Python

Python

Overview

Modern Python development patterns including type hints, async programming, and Pythonic idioms.

Type Hints

Basic Types

from typing import (
    Optional, Union, List, Dict, Set, Tuple,
    TypeVar, Generic, Callable, Any,
    Literal, TypedDict, Protocol
)
from dataclasses import dataclass
from datetime import datetime

# Basic type hints
def greet(name: str) -> str:
    return f"Hello, {name}!"

# Optional (can be None)
def find_user(user_id: str) -> Optional['User']:
    return users.get(user_id)

# Union types
def process(value: Union[str, int]) -> str:
    return str(value)

# Python 3.10+ union syntax
def process_new(value: str | int | None) -> str:
    return str(value) if value else ""

# Collections
def process_items(
    items: List[str],
    mapping: Dict[str, int],
    unique: Set[str],
    pair: Tuple[str, int]
) -> None:
    pass

# Python 3.9+ built-in generics
def process_items_new(
    items: list[str],
    mapping: dict[str, int],
    unique: set[str]
) -> None:
    pass

Advanced Types

# TypeVar for generics
T = TypeVar('T')
K = TypeVar('K')
V = TypeVar('V')

def first(items: list[T]) -> T | None:
    return items[0] if items else None

# Generic classes
class Repository(Generic[T]):
    def __init__(self) -> None:
        self._items: dict[str, T] = {}

    def get(self, id: str) -> T | None:
        return self._items.get(id)

    def save(self, id: str, item: T) -> None:
        self._items[id] = item

# TypedDict for structured dicts
class UserDict(TypedDict):
    id: str
    name: str
    email: str
    age: int  # Required
    nickname: str  # Required

class PartialUserDict(TypedDict, total=False):
    nickname: str  # Optional

# Literal types
Mode = Literal["read", "write", "append"]

def open_file(path: str, mode: Mode) -> None:
    pass

# Protocol (structural typing)
class Readable(Protocol):
    def read(self) -> str: ...

def process_readable(source: Readable) -> str:
    return source.read()

# Callable types
Handler = Callable[[str, int], bool]
AsyncHandler = Callable[[str], 'Awaitable[bool]']

def register_handler(handler: Handler) -> None:
    pass

Dataclasses

from dataclasses import dataclass, field, asdict, astuple
from typing import ClassVar
from datetime import datetime

@dataclass
class User:
    id: str
    email: str
    name: str
    created_at: datetime = field(default_factory=datetime.now)
    tags: list[str] = field(default_factory=list)
    _cache: dict = field(default_factory=dict, repr=False, compare=False)

    # Class variable (not instance field)
    MAX_TAGS: ClassVar[int] = 10

    def __post_init__(self):
        # Validation after init
        if len(self.tags) > self.MAX_TAGS:
            raise ValueError(f"Too many tags (max {self.MAX_TAGS})")

# Frozen (immutable)
@dataclass(frozen=True)
class Point:
    x: float
    y: float

    def distance_from_origin(self) -> float:
        return (self.x ** 2 + self.y ** 2) ** 0.5

# Slots for memory efficiency
@dataclass(slots=True)
class LightweightUser:
    id: str
    name: str

# Convert to dict/tuple
user = User(id="1", email="test@example.com", name="Test")
user_dict = asdict(user)
user_tuple = astuple(user)

Decorators

from functools import wraps
from typing import TypeVar, Callable, ParamSpec
import time

P = ParamSpec('P')
R = TypeVar('R')

# Basic decorator
def timer(func: Callable[P, R]) -> Callable[P, R]:
    @wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"{func.__name__} took {elapsed:.4f}s")
        return result
    return wrapper

# Decorator with arguments
def retry(max_attempts: int = 3, delay: float = 1.0):
    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        @wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            last_exception: Exception | None = None
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_exception = e
                    if attempt < max_attempts - 1:
                        time.sleep(delay)
            raise last_exception
        return wrapper
    return decorator

# Class decorator
def singleton(cls):
    instances = {}
    @wraps(cls)
    def get_instance(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]
    return get_instance

# Usage
@timer
@retry(max_attempts=3, delay=0.5)
def fetch_data(url: str) -> dict:
    # ... fetch logic
    pass

@singleton
class Database:
    def __init__(self, connection_string: str):
        self.connection_string = connection_string

Async Programming

import asyncio
from typing import AsyncIterator
import aiohttp

# Async function
async def fetch_url(url: str) -> str:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

# Parallel execution
async def fetch_all(urls: list[str]) -> list[str]:
    tasks = [fetch_url(url) for url in urls]
    return await asyncio.gather(*tasks)

# With error handling
async def fetch_all_safe(urls: list[str]) -> list[str | None]:
    tasks = [fetch_url(url) for url in urls]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return [r if isinstance(r, str) else None for r in results]

# Async context manager
class AsyncDatabase:
    async def __aenter__(self) -> 'AsyncDatabase':
        await self.connect()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
        await self.disconnect()

    async def connect(self) -> None:
        print("Connecting...")

    async def disconnect(self) -> None:
        print("Disconnecting...")

# Async generator
async def paginate(
    fetch_page: Callable[[int], 'Awaitable[list[T]]']
) -> AsyncIterator[T]:
    page = 1
    while True:
        items = await fetch_page(page)
        if not items:
            break
        for item in items:
            yield item
        page += 1

# Using async for
async def process_all_items():
    async for item in paginate(fetch_page):
        await process_item(item)

# Semaphore for rate limiting
async def fetch_with_limit(urls: list[str], max_concurrent: int = 10):
    semaphore = asyncio.Semaphore(max_concurrent)

    async def fetch_limited(url: str) -> str:
        async with semaphore:
            return await fetch_url(url)

    return await asyncio.gather(*[fetch_limited(url) for url in urls])

Context Managers

from contextlib import contextmanager, asynccontextmanager
from typing import Generator, AsyncGenerator

# Class-based context manager
class Timer:
    def __init__(self, name: str):
        self.name = name
        self.start: float = 0
        self.elapsed: float = 0

    def __enter__(self) -> 'Timer':
        self.start = time.perf_counter()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
        self.elapsed = time.perf_counter() - self.start
        print(f"{self.name}: {self.elapsed:.4f}s")

# Generator-based context manager
@contextmanager
def timer(name: str) -> Generator[None, None, None]:
    start = time.perf_counter()
    try:
        yield
    finally:
        elapsed = time.perf_counter() - start
        print(f"{name}: {elapsed:.4f}s")

# Async context manager
@asynccontextmanager
async def async_timer(name: str) -> AsyncGenerator[None, None]:
    start = time.perf_counter()
    try:
        yield
    finally:
        elapsed = time.perf_counter() - start
        print(f"{name}: {elapsed:.4f}s")

# Usage
with timer("operation"):
    do_something()

async with async_timer("async_operation"):
    await do_something_async()

Itertools and Generators

from itertools import (
    chain, islice, groupby, takewhile, dropwhile,
    combinations, permutations, product, accumulate
)
from typing import Iterator, Iterable

# Generator function
def fibonacci() -> Iterator[int]:
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b

# Take first n
first_10_fib = list(islice(fibonacci(), 10))

# Generator expression
squares = (x ** 2 for x in range(10))

# Chain multiple iterables
all_items = chain(list1, list2, list3)

# Group by
data = [
    {"type": "a", "value": 1},
    {"type": "a", "value": 2},
    {"type": "b", "value": 3},
]
for key, group in groupby(sorted(data, key=lambda x: x["type"]), key=lambda x: x["type"]):
    print(f"{key}: {list(group)}")

# Batching
def batch(iterable: Iterable[T], size: int) -> Iterator[list[T]]:
    iterator = iter(iterable)
    while batch := list(islice(iterator, size)):
        yield batch

# Sliding window
def sliding_window(iterable: Iterable[T], size: int) -> Iterator[tuple[T, ...]]:
    from collections import deque
    iterator = iter(iterable)
    window = deque(islice(iterator, size), maxlen=size)
    if len(window) == size:
        yield tuple(window)
    for item in iterator:
        window.append(item)
        yield tuple(window)

Error Handling

from typing import TypeVar, Generic
from dataclasses import dataclass

T = TypeVar('T')
E = TypeVar('E', bound=Exception)

# Custom exceptions
class AppError(Exception):
    def __init__(self, message: str, code: str):
        super().__init__(message)
        self.code = code

class ValidationError(AppError):
    def __init__(self, message: str, fields: dict[str, list[str]]):
        super().__init__(message, "VALIDATION_ERROR")
        self.fields = fields

# Result type pattern
@dataclass
class Ok(Generic[T]):
    value: T

    def is_ok(self) -> bool:
        return True

    def is_err(self) -> bool:
        return False

@dataclass
class Err(Generic[E]):
    error: E

    def is_ok(self) -> bool:
        return False

    def is_err(self) -> bool:
        return True

Result = Ok[T] | Err[E]

def parse_int(s: str) -> Result[int, ValueError]:
    try:
        return Ok(int(s))
    except ValueError as e:
        return Err(e)

# Exception chaining
try:
    process_data()
except ValueError as e:
    raise AppError("Failed to process data", "PROCESS_ERROR") from e

Related Skills

• [[ai-ml-integration]] - ML/AI with Python
• [[backend]] - FastAPI/Django
• [[automation-scripts]] - Scripting and automation

Converted and distributed by TomeVault — claim your Tome and manage your conversions.