add language pattern notes: typing, async, pydantic, patterns

restructure python notes to cover the language itself, not just tooling:
- typing.md: modern type hints, Protocol, generics, ParamSpec
- async.md: context managers, ContextVar, concurrency patterns
- pydantic.md: models, validators, discriminated unions, settings
- patterns.md: class design, decorators, error handling

rewrite README as friendly introduction with language/ecosystem split

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <[e-mail]>

languages/python/README.md

···
 # python
 
-notes on python patterns - tooling, project structure, and building things.
+notes on writing python - the language itself, the ecosystem around it, and patterns that work.
 
-## topics
+python is easy to start but has depth. these notes focus on modern python (3.10+) with strong typing, async patterns, and the tooling that makes it feel like a proper systems language.
+
+## the language
+
+how to write python code:
+
+- [typing](./typing.md) - modern type hints, Protocol, generics
+- [async](./async.md) - context managers, ContextVar, patterns
+- [pydantic](./pydantic.md) - models, validation, settings
+- [patterns](./patterns.md) - class design, decorators, error handling
+
+## the ecosystem
+
+tooling and project structure:
 
 - [uv](./uv.md) - cargo for python
 - [project setup](./project-setup.md) - src/ layout, pyproject.toml, justfile
-- [pydantic-settings](./pydantic-settings.md) - centralized typed configuration
 - [tooling](./tooling.md) - ruff, ty, pre-commit
 - [mcp](./mcp.md) - building MCP servers with fastmcp
 
···
 |---------|------------|
 | [pdsx](https://github.com/zzstoatzz/pdsx) | ATProto MCP server and CLI |
 | [plyr-python-client](https://github.com/zzstoatzz/plyr-python-client) | uv workspace, multi-package repo |
-| [pmgfal](https://github.com/zzstoatzz/pmgfal) | Rust+Python with maturin |
 | [prefect-mcp-server-demo](https://github.com/zzstoatzz/prefect-mcp-server-demo) | MCP server patterns |
-| [typsht](https://github.com/zzstoatzz/typsht) | parallel type checking tool |
 
 and studying:
 

languages/python/async.md

···
+# async
+
+async python patterns - context managers, concurrency, and request-scoped state.
+
+## async context managers
+
+the core pattern for resource lifecycle:
+
+```python
+from contextlib import asynccontextmanager
+from typing import AsyncIterator
+
+@asynccontextmanager
+async def get_client(require_auth: bool = False) -> AsyncIterator[Client]:
+    """acquire a client, ensure cleanup."""
+    client = Client()
+
+    if require_auth:
+        await client.login()
+
+    try:
+        yield client
+    finally:
+        await client.close()
+```
+
+usage:
+
+```python
+async with get_client(require_auth=True) as client:
+    await client.do_something()
+# client.close() called automatically
+```
+
+## class-based context managers
+
+when you need state:
+
+```python
+class AsyncClient:
+    async def __aenter__(self) -> "AsyncClient":
+        self._session = await self._create_session()
+        return self
+
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: TracebackType | None,
+    ) -> None:
+        await self._session.close()
+```
+
+## AsyncExitStack
+
+compose multiple context managers:
+
+```python
+from contextlib import AsyncExitStack
+
+async def with_dependencies() -> dict[str, Any]:
+    async with AsyncExitStack() as stack:
+        db = await stack.enter_async_context(get_database())
+        cache = await stack.enter_async_context(get_cache())
+
+        return {"db": db, "cache": cache}
+```
+
+useful when the number of context managers is dynamic.
+
+## ContextVar
+
+request-scoped state without passing through every function:
+
+```python
+from contextvars import ContextVar
+
+_current_user: ContextVar[User | None] = ContextVar("current_user", default=None)
+
+def get_current_user() -> User:
+    user = _current_user.get()
+    if user is None:
+        raise RuntimeError("no user in context")
+    return user
+
+async def with_user(user: User):
+    token = _current_user.set(user)
+    try:
+        yield
+    finally:
+        _current_user.reset(token)
+```
+
+each async task gets its own copy. no global state pollution.
+
+## semaphore for concurrency
+
+limit concurrent operations:
+
+```python
+import asyncio
+
+async def batch_process(items: list[str], concurrency: int = 10) -> list[Result]:
+    semaphore = asyncio.Semaphore(concurrency)
+
+    async def process_one(item: str) -> Result:
+        async with semaphore:
+            return await do_work(item)
+
+    return await asyncio.gather(*[process_one(item) for item in items])
+```
+
+## gather with error handling
+
+don't let one failure kill everything:
+
+```python
+results = await asyncio.gather(
+    *tasks,
+    return_exceptions=True,
+)
+
+successes = [r for r in results if not isinstance(r, Exception)]
+failures = [r for r in results if isinstance(r, Exception)]
+```
+
+## shield for critical cleanup
+
+prevent cancellation during important operations:
+
+```python
+async def __aexit__(self, *args) -> None:
+    # don't let cancellation interrupt disconnect
+    await asyncio.shield(self._connection.disconnect())
+```
+
+## streaming responses
+
+async iteration over chunks:
+
+```python
+async with client.stream("GET", url) as response:
+    async for line in response.aiter_lines():
+        if line.startswith("data: "):
+            yield json.loads(line[6:])
+```
+
+## background tasks
+
+fire and forget with cleanup:
+
+```python
+class Worker:
+    async def __aenter__(self) -> "Worker":
+        self._heartbeat_task = asyncio.create_task(
+            self._heartbeat(),
+            name="worker.heartbeat",
+        )
+        return self
+
+    async def __aexit__(self, *args) -> None:
+        self._heartbeat_task.cancel()
+        with suppress(asyncio.CancelledError):
+            await self._heartbeat_task
+```
+
+sources:
+- [pdsx/src/pdsx/mcp/client.py](https://github.com/zzstoatzz/pdsx/blob/main/src/pdsx/mcp/client.py)
+- [docket/src/docket/worker.py](https://github.com/chrisguidry/docket/blob/main/src/docket/worker.py)

languages/python/patterns.md

···
+# patterns
+
+class design, decorators, error handling, and other structural patterns.
+
+## private internals
+
+keep implementation details in `_internal/`:
+
+```
+src/mypackage/
+├── __init__.py          # public API
+├── cli.py               # entry point
+└── _internal/           # implementation details
+    ├── config.py
+    ├── operations.py
+    └── types.py
+```
+
+`__init__.py` re-exports the public interface. users import from the package, not from internals.
+
+## dataclasses for DTOs
+
+simple data containers that don't need validation:
+
+```python
+from dataclasses import dataclass
+
+@dataclass
+class BatchResult:
+    successful: list[str]
+    failed: list[tuple[str, Exception]]
+
+    @property
+    def total(self) -> int:
+        return len(self.successful) + len(self.failed)
+```
+
+lighter than pydantic when you control the data source.
+
+## base classes with parallel implementations
+
+when you need both sync and async:
+
+```python
+class _BaseClient:
+    def __init__(self, *, token: str | None = None):
+        self._token = token or get_settings().token
+
+class Client(_BaseClient):
+    def get(self, url: str) -> dict:
+        return httpx.get(url, headers=self._headers).json()
+
+class AsyncClient(_BaseClient):
+    async def get(self, url: str) -> dict:
+        async with httpx.AsyncClient() as client:
+            return (await client.get(url, headers=self._headers)).json()
+```
+
+shared logic in base, divergent implementation in subclasses.
+
+## fluent interfaces
+
+chainable methods that return `self`:
+
+```python
+class Track:
+    def __init__(self, source: str):
+        self._source = source
+        self._effects: list[str] = []
+
+    def volume(self, level: float) -> "Track":
+        self._effects.append(f"volume={level}")
+        return self
+
+    def lowpass(self, freq: float) -> "Track":
+        self._effects.append(f"lowpass=f={freq}")
+        return self
+
+# usage
+track.volume(0.8).lowpass(600).fade_in(0.5)
+```
+
+## factory classmethods
+
+alternative constructors:
+
+```python
+@dataclass
+class URIParts:
+    repo: str
+    collection: str
+    rkey: str
+
+    @classmethod
+    def from_uri(cls, uri: str) -> "URIParts":
+        # at://did:plc:xxx/app.bsky.feed.post/abc
+        parts = uri.replace("at://", "").split("/")
+        return cls(repo=parts[0], collection=parts[1], rkey=parts[2])
+```
+
+## keyword-only arguments
+
+force callers to name arguments for clarity:
+
+```python
+def batch_create(
+    client: Client,
+    collection: str,
+    records: list[dict],
+    *,  # everything after is keyword-only
+    concurrency: int = 10,
+    fail_fast: bool = False,
+) -> BatchResult:
+    ...
+
+# must use: batch_create(client, "posts", items, concurrency=5)
+# not: batch_create(client, "posts", items, 5)
+```
+
+## custom exceptions
+
+with helpful context:
+
+```python
+class AuthenticationRequired(Exception):
+    def __init__(self, operation: str = "this operation"):
+        super().__init__(
+            f"{operation} requires authentication.\n\n"
+            "Set ATPROTO_HANDLE and ATPROTO_PASSWORD environment variables."
+        )
+```
+
+the message tells you what to do, not just what went wrong.
+
+## exception hierarchies
+
+for structured error handling:
+
+```python
+class AppError(Exception):
+    """base for all app errors."""
+
+class ValidationError(AppError):
+    """input validation failed."""
+
+class ResourceError(AppError):
+    """resource operation failed."""
+
+# callers can catch AppError for all, or specific types
+```
+
+## decorators that modify signatures
+
+add parameters dynamically:
+
+```python
+import inspect
+from functools import wraps
+
+def filterable(fn):
+    """add _filter parameter for jmespath filtering."""
+    @wraps(fn)
+    async def wrapper(*args, _filter: str | None = None, **kwargs):
+        result = await fn(*args, **kwargs)
+        if _filter:
+            import jmespath
+            return jmespath.search(_filter, result)
+        return result
+
+    # modify signature to include new param
+    sig = inspect.signature(fn)
+    params = list(sig.parameters.values())
+    params.append(inspect.Parameter(
+        "_filter",
+        inspect.Parameter.KEYWORD_ONLY,
+        default=None,
+    ))
+    wrapper.__signature__ = sig.replace(parameters=params)
+    return wrapper
+```
+
+the wrapper's signature now includes `_filter`. IDEs and schema generators see it.
+
+## module-level singletons
+
+instantiate once, import everywhere:
+
+```python
+# config.py
+from functools import lru_cache
+
+@lru_cache
+def get_settings() -> Settings:
+    return Settings()
+
+settings = get_settings()
+
+# console.py
+from rich.console import Console
+
+console = Console()
+```
+
+## lowercase aesthetic
+
+docstrings, comments, and output in lowercase:
+
+```python
+"""list records in a collection."""
+
+def list_records(collection: str) -> list[dict]:
+    """list records from the specified collection.
+
+    args:
+        collection: the collection to list from
+
+    returns:
+        list of record dictionaries
+    """
+```
+
+consistent voice throughout.
+
+sources:
+- [pdsx](https://github.com/zzstoatzz/pdsx)
+- [plyr-python-client](https://github.com/zzstoatzz/plyr-python-client)
+- [docket](https://github.com/chrisguidry/docket)

languages/python/pydantic-settings.md

···
-# pydantic-settings
-
-replace scattered `os.getenv()` calls with a typed, validated settings class.
-
-## the problem
-
-```python
-import os
-
-REDIS_HOST = os.getenv("REDIS_HOST")
-REDIS_PORT = os.getenv("REDIS_PORT")
-
-if not REDIS_HOST or not REDIS_PORT:
-    raise ValueError("REDIS_HOST and REDIS_PORT must be set")
-
-# REDIS_PORT is still a string here
-```
-
-issues:
-- validation happens where you use the value, not at startup
-- no type coercion (port is a string)
-- configuration scattered across files
-- easy to forget validation
-
-## the solution
-
-```python
-from pydantic import Field, SecretStr
-from pydantic_settings import BaseSettings, SettingsConfigDict
-
-class Settings(BaseSettings):
-    model_config = SettingsConfigDict(
-        env_file=".env",
-        extra="ignore",
-    )
-
-    redis_host: str
-    redis_port: int = Field(ge=0)
-    openai_api_key: SecretStr
-
-settings = Settings()
-```
-
-now:
-- missing `REDIS_HOST` fails immediately at import
-- `redis_port` is coerced to int and validated >= 0
-- `openai_api_key` won't accidentally appear in logs
-- all configuration in one place
-
-## field aliases
-
-when env var names don't match your preferred attribute names:
-
-```python
-class Settings(BaseSettings):
-    current_user: str = Field(alias="USER")
-```
-
-reads from `$USER`, accessed as `settings.current_user`.
-
-## secrets
-
-`SecretStr` prevents accidental exposure:
-
-```python
-class Settings(BaseSettings):
-    api_key: SecretStr
-
-settings = Settings()
-print(settings.api_key)              # SecretStr('**********')
-print(settings.api_key.get_secret_value())  # actual value
-```
-
-## contextual serialization
-
-when you need to unmask secrets for subprocesses:
-
-```python
-from pydantic import Secret, SerializationInfo
-
-def maybe_unmask(v: Secret[str], info: SerializationInfo) -> str | Secret[str]:
-    if info.context and info.context.get("unmask"):
-        return v.get_secret_value()
-    return v
-
-# usage
-settings.model_dump(context={"unmask": True})
-```
-
-## nested settings
-
-for larger projects:
-
-```python
-class DatabaseSettings(BaseSettings):
-    host: str = "localhost"
-    port: int = 5432
-
-class Settings(BaseSettings):
-    database: DatabaseSettings = Field(default_factory=DatabaseSettings)
-```
-
-## why fail-fast matters
-
-with `os.getenv()`, you find out about missing config when the code path executes - maybe in production, at 2am.
-
-with pydantic-settings, invalid configuration fails at startup. deploy fails, not runtime.
-
-sources:
-- [how to use pydantic-settings](https://blog.zzstoatzz.io/how-to-use-pydantic-settings/)

languages/python/pydantic.md

···
+# pydantic
+
+data validation, serialization, and settings. the foundation for typed python.
+
+## BaseModel basics
+
+```python
+from pydantic import BaseModel, Field, ConfigDict
+
+class Track(BaseModel):
+    model_config = ConfigDict(extra="forbid")  # reject unknown fields
+
+    title: str
+    album: str | None = None
+    tags: list[str] = Field(default_factory=list)
+```
+
+`extra="forbid"` catches typos. if someone passes `titl` instead of `title`, it fails.
+
+## field aliases
+
+when external data uses different names:
+
+```python
+class Track(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    audio_url: str | None = Field(default=None, alias="r2_url")
+```
+
+accepts both `audio_url` and `r2_url` as input.
+
+## validators
+
+transform or check data:
+
+```python
+from pydantic import field_validator, model_validator
+from typing import Self
+
+class Document(BaseModel):
+    text: str
+    tokens: int | None = None
+
+    @field_validator("text", mode="before")
+    @classmethod
+    def strip_whitespace(cls, v: str) -> str:
+        return v.strip()
+
+    @model_validator(mode="after")
+    def compute_tokens(self) -> Self:
+        if self.tokens is None:
+            self.tokens = len(self.text.split())
+        return self
+```
+
+`mode="before"` runs before pydantic's own validation. `mode="after"` runs after the model is constructed.
+
+## discriminated unions
+
+polymorphic types with a type field:
+
+```python
+from typing import Annotated, Literal
+from pydantic import Field
+
+class TrackResult(BaseModel):
+    type: Literal["track"] = "track"
+    title: str
+
+class ArtistResult(BaseModel):
+    type: Literal["artist"] = "artist"
+    name: str
+
+SearchResult = Annotated[
+    TrackResult | ArtistResult,
+    Field(discriminator="type"),
+]
+
+def parse_results(data: list[dict]) -> list[SearchResult]:
+    from pydantic import TypeAdapter
+    adapter = TypeAdapter(list[SearchResult])
+    return adapter.validate_python(data)
+```
+
+pydantic looks at `type` to decide which model to use.
+
+## TypedDict for loose structures
+
+when you don't need full model machinery:
+
+```python
+from typing import TypedDict
+
+class RecordResponse(TypedDict):
+    uri: str
+    cid: str | None
+    value: dict[str, Any]
+```
+
+lighter than BaseModel, still typed.
+
+## settings from environment
+
+replace `os.getenv()` with validated configuration:
+
+```python
+from pydantic import Field, SecretStr
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(
+        env_prefix="APP_",
+        env_file=".env",
+        extra="ignore",
+    )
+
+    database_url: str
+    api_key: SecretStr
+    debug: bool = False
+
+settings = Settings()
+```
+
+- `env_prefix` means `APP_DATABASE_URL` maps to `database_url`
+- `SecretStr` hides values in logs
+- missing required fields fail at import, not at runtime
+
+## cached singleton
+
+```python
+from functools import lru_cache
+
+@lru_cache
+def get_settings() -> Settings:
+    return Settings()
+```
+
+one instance, created once, reused everywhere.
+
+## nested settings
+
+for complex configuration:
+
+```python
+class DatabaseSettings(BaseSettings):
+    host: str = "localhost"
+    port: int = 5432
+
+class Settings(BaseSettings):
+    database: DatabaseSettings = Field(default_factory=DatabaseSettings)
+    cache: CacheSettings = Field(default_factory=CacheSettings)
+```
+
+## factory methods
+
+alternative constructors:
+
+```python
+class Config(BaseModel):
+    name: str
+    values: dict[str, Any]
+
+    @classmethod
+    def from_file(cls, path: Path) -> "Config":
+        data = json.loads(path.read_text())
+        return cls.model_validate(data)
+```
+
+## TypeAdapter for non-model validation
+
+validate without a full model:
+
+```python
+from pydantic import TypeAdapter
+
+adapter = TypeAdapter(list[int])
+result = adapter.validate_python(["1", "2", "3"])  # [1, 2, 3]
+```
+
+useful for validating function arguments or API responses.
+
+sources:
+- [pdsx/src/pdsx/_internal/types.py](https://github.com/zzstoatzz/pdsx/blob/main/src/pdsx/_internal/types.py)
+- [plyr-python-client/packages/plyrfm/src/plyrfm/_internal/types.py](https://github.com/zzstoatzz/plyr-python-client)
+- [how to use pydantic-settings](https://blog.zzstoatzz.io/how-to-use-pydantic-settings/)

languages/python/typing.md

···
+# typing
+
+modern python type hints - not just for IDEs, but for thinking clearly about code.
+
+## basics
+
+use `|` for unions, not `Optional`:
+
+```python
+# yes
+def fetch(url: str, timeout: float | None = None) -> dict[str, Any]: ...
+
+# no
+from typing import Optional
+def fetch(url: str, timeout: Optional[float] = None) -> Dict[str, Any]: ...
+```
+
+`from __future__ import annotations` at the top of every file. this defers evaluation so you can reference types before they're defined.
+
+## TypedDict
+
+for structured dictionaries where you know the shape:
+
+```python
+from typing import TypedDict
+
+class RecordResponse(TypedDict):
+    uri: str
+    cid: str | None
+    value: dict[str, Any]
+
+def get_record(uri: str) -> RecordResponse:
+    ...
+```
+
+better than `dict[str, Any]` because the structure is documented and checked.
+
+## Annotated
+
+add metadata to types, especially useful with pydantic:
+
+```python
+from typing import Annotated
+from pydantic import Field
+
+FilterParam = Annotated[
+    str | None,
+    Field(description="jmespath expression to filter results"),
+]
+
+def list_records(collection: str, _filter: FilterParam = None) -> list[dict]:
+    ...
+```
+
+the metadata travels with the type. frameworks like fastmcp use it for schema generation.
+
+## Protocol
+
+duck typing with type safety. define what methods something needs, not what class it is:
+
+```python
+from typing import Protocol, runtime_checkable
+
+@runtime_checkable
+class Closeable(Protocol):
+    def close(self) -> None: ...
+
+def cleanup(resource: Closeable) -> None:
+    resource.close()
+```
+
+`@runtime_checkable` lets you use `isinstance()` checks.
+
+## generics
+
+for functions that work with any type:
+
+```python
+from typing import TypeVar
+
+T = TypeVar("T")
+
+def first(items: list[T]) -> T | None:
+    return items[0] if items else None
+```
+
+with bounds:
+
+```python
+from pydantic import BaseModel
+
+ModelT = TypeVar("ModelT", bound=BaseModel)
+
+def parse(data: dict, model: type[ModelT]) -> ModelT:
+    return model.model_validate(data)
+```
+
+## ParamSpec
+
+for decorators that preserve function signatures:
+
+```python
+from typing import ParamSpec, TypeVar, Callable, Awaitable
+from functools import wraps
+
+P = ParamSpec("P")
+R = TypeVar("R")
+
+def logged(fn: Callable[P, Awaitable[R]]) -> Callable[P, Awaitable[R]]:
+    @wraps(fn)
+    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
+        print(f"calling {fn.__name__}")
+        return await fn(*args, **kwargs)
+    return wrapper
+```
+
+this preserves the original function's parameter types in the wrapper.
+
+## overload
+
+when return type depends on input:
+
+```python
+from typing import overload, Literal
+
+@overload
+def fetch(url: str, raw: Literal[True]) -> bytes: ...
+@overload
+def fetch(url: str, raw: Literal[False] = False) -> dict: ...
+
+def fetch(url: str, raw: bool = False) -> bytes | dict:
+    response = httpx.get(url)
+    return response.content if raw else response.json()
+```
+
+IDEs and type checkers know that `fetch(url, raw=True)` returns `bytes`.
+
+## TYPE_CHECKING
+
+avoid runtime import costs for types only needed for hints:
+
+```python
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from heavy_module import ExpensiveClass
+
+def process(item: "ExpensiveClass") -> None:
+    ...
+```
+
+the string quote is needed because the import doesn't happen at runtime.
+
+## import organization
+
+```python
+"""module docstring."""
+
+from __future__ import annotations
+
+import stdlib_module
+from typing import TYPE_CHECKING
+
+from third_party import thing
+
+from local_package import helper
+
+if TYPE_CHECKING:
+    from expensive import Type
+```
+
+sources:
+- [pdsx/src/pdsx/_internal/types.py](https://github.com/zzstoatzz/pdsx/blob/main/src/pdsx/_internal/types.py)
+- [fastmcp/src/fastmcp/tools/tool.py](https://github.com/jlowin/fastmcp/blob/main/src/fastmcp/tools/tool.py)