2tony2

from abc import ABC, abstractmethod
import asyncio
import asyncpg
import time

class AsyncDatabaseManager(ABC):
    @abstractmethod
    async def query(self, dsn, sql):
        """An asynchronous context manager for executing and timing database queries."""
        pass

2tony2

class DataFrame:
    name: str
    rows: int

    def __init__(self, name: str, rows: int):
        self.name = name
        self.rows = rows

def process_dataframe(df: DataFrame) -> str:
    return f"Processing {df.name} with {df.rows} rows."

2tony2

from typing import get_origin, get_args

# Define some generic and non-generic types
generic_type = list[int]
non_generic_type = int
generic_type_without_params = list

# Function to check if a type is a generic type
def is_generic_type(tp):
    return get_origin(tp) is not None or hasattr(tp, '__origin__') and tp.__origin__ is not None

2tony2

from typing import Annotated
from typing_extensions import Annotated

# Define an annotated type with metadata
Age = Annotated[int, "Must be a non-negative integer"]

def check_age(age: Age):
    if age < 0:
        raise ValueError("Age must be a non-negative integer")
    print(f"Age is valid: {age}")

2tony2

from collections.abc import Callable, Awaitable

# Function with callable parameter
def feeder(get_next_item: Callable[[], str]) -> None:
    ...  # Body

# Function with multiple callable parameters
def async_query(on_success: Callable[[int], None],
                on_error: Callable[[int, Exception], None]) -> None:
    ...  # Body

2tony2

def concat(x: str, y: str) -> str:
    return x + y

x: Callable[..., str]
x = str     # OK
x = concat  # Also OK

2tony2

class SQLDataSource(DataSource):
    connection_params = {}

    @classmethod
    def configure(cls, config):
        cls.connection_params = config
        return f"Configured SQL database with parameters: {cls.connection_params}"

    def execute_query(self, query):
        return f"Executing SQL query: '{query}' with parameters {self.connection_params}"

2tony2

from typing import Any

def process_item(item: Any) -> None:
    print(f"Processing item: {item}")

# Usage examples
process_item(123)          # Processing an integer
process_item("Hello")      # Processing a string
process_item([1, 2, 3])    # Processing a list

2tony2

from typing import TypeVar, List

T = TypeVar('T')

def duplicate(item: T) -> List[T]:
    return [item, item]

# Usage examples
print(duplicate(123))          # [123, 123]
print(duplicate("Hello"))      # ['Hello', 'Hello']

2tony2

from typing import Union

def process_record(record_id: Union[str, int]) -> str:
    if isinstance(record_id, str):
        return f"Processing record with string ID: {record_id}"
    elif isinstance(record_id, int):
        return f"Processing record with integer ID: {record_id}"
    else:
        return "Unsupported record ID type"