From c4273fe98fce02feb4cda7173645cb90f4381717 Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Sat, 30 May 2026 19:12:18 +0900 Subject: [PATCH] docs(omo-codex): batch 95 (14 files) --- .../programming/references/python/README.md | 314 +++++++++++++ .../references/python/async-anyio.md | 442 ++++++++++++++++++ .../references/python/data-modeling.md | 233 +++++++++ .../references/python/data-processing.md | 133 ++++++ .../references/python/error-handling.md | 218 +++++++++ .../references/python/fastapi-stack.md | 316 +++++++++++++ .../references/python/httpx2-optimization.md | 360 ++++++++++++++ .../references/python/libraries.md | 307 ++++++++++++ .../references/python/one-liners.md | 268 +++++++++++ .../references/python/orjson-stack.md | 378 +++++++++++++++ .../references/python/pydantic-ai.md | 285 +++++++++++ .../references/python/pyproject-strict.md | 232 +++++++++ .../references/python/textual-tui.md | 201 ++++++++ .../references/python/type-patterns.md | 176 +++++++ 14 files changed, 3863 insertions(+) create mode 100644 packages/omo-codex/plugin/skills/programming/references/python/README.md create mode 100644 packages/omo-codex/plugin/skills/programming/references/python/async-anyio.md create mode 100644 packages/omo-codex/plugin/skills/programming/references/python/data-modeling.md create mode 100644 packages/omo-codex/plugin/skills/programming/references/python/data-processing.md create mode 100644 packages/omo-codex/plugin/skills/programming/references/python/error-handling.md create mode 100644 packages/omo-codex/plugin/skills/programming/references/python/fastapi-stack.md create mode 100644 packages/omo-codex/plugin/skills/programming/references/python/httpx2-optimization.md create mode 100644 packages/omo-codex/plugin/skills/programming/references/python/libraries.md create mode 100644 packages/omo-codex/plugin/skills/programming/references/python/one-liners.md create mode 100644 packages/omo-codex/plugin/skills/programming/references/python/orjson-stack.md create mode 100644 packages/omo-codex/plugin/skills/programming/references/python/pydantic-ai.md create mode 100644 packages/omo-codex/plugin/skills/programming/references/python/pyproject-strict.md create mode 100644 packages/omo-codex/plugin/skills/programming/references/python/textual-tui.md create mode 100644 packages/omo-codex/plugin/skills/programming/references/python/type-patterns.md diff --git a/packages/omo-codex/plugin/skills/programming/references/python/README.md b/packages/omo-codex/plugin/skills/programming/references/python/README.md new file mode 100644 index 000000000..0b3ebf640 --- /dev/null +++ b/packages/omo-codex/plugin/skills/programming/references/python/README.md @@ -0,0 +1,314 @@ + +# Python Programmer + +Modern Python. Type-strict, stack-first, async-correct. + +## Philosophy + +The type checker is your compiler. Make illegal states unrepresentable. Parse at boundaries. Own resources explicitly. Every function has a contract; the type system enforces it. + +## Hard rules + +These are deliberate project choices. Violations are always wrong, not "style preferences". + +### Tooling + +| Category | Use | Never | +|---|---|---| +| Package manager | `uv` | pip, poetry, conda, pipenv | +| Type checker | `basedpyright` (`typeCheckingMode = "all"`) | pyright, mypy | +| Linter + formatter | `ruff` (`select = ["ALL"]`) | flake8, black, isort, autopep8 | +| Async runtime | `anyio` | `import asyncio` | +| Data | `polars` + `duckdb` + `numpy` | pandas | +| Web framework | FastAPI + Pydantic v2 | Flask, Django REST | +| ORM | SQLAlchemy 2.x async | Django ORM, Tortoise | +| HTTP client | [`httpx2`](https://github.com/pydantic/httpx2) | requests, aiohttp, httpx | +| Testing | `pytest` | unittest | +| CLI | `typer` + `rich` | argparse, click, fire | + +### The iron list + +1. **Frozen by default** — `@dataclass(frozen=True, slots=True)`. Pydantic: `model_config = ConfigDict(frozen=True)`. Mutable only when mutation is the documented purpose. +2. **NewType for distinct IDs** — `UserId = NewType("UserId", int)`. Never pass raw `int` where a branded type exists. +3. **`match` only for variants, `if` only for booleans** — **NEVER** use `if/elif/else` to discriminate on type (`isinstance`), enum value, or literal variant. `match/case` is mandatory for these — non-negotiable. **ALWAYS** end with `case unreachable: assert_never(unreachable)` — bare `case _: pass` and `case _: raise ValueError` are banned (they silently swallow new variants). `if/else` is fine only for boolean expressions, range checks, and predicate calls that aren't variant discrimination. See "Why `if/elif` on variants is banned" below for examples. +4. **Protocol over ABC** — `typing.Protocol` for interfaces. ABC only when you need shared method implementation. +5. **No raw dicts in signatures** — params and returns use `TypedDict`, `dataclass`, or Pydantic model. Internal scratch dicts are fine. +6. **Parse, don't validate** — constructors produce typed objects or raise. Never pass unvalidated data deeper into the call stack. +7. **Typed errors** — error types are dataclasses or exceptions with typed fields. Never `raise ValueError("something")` with a bare string. Use union returns when the caller is within 1-2 call levels and must handle the outcome (repository → service). Use exceptions when the error should propagate up many layers to a boundary handler (service → HTTP handler). +8. **Final for constants** — module-level constants use `Final`. Mutable module globals are a code smell. +9. **Explicit None** — annotate `-> X | None`. Never return `None` from a function whose signature omits it. +10. **Context managers for resources** — files, DB connections, HTTP clients, locks. No manual `.close()`. +11. **No Any, no object** — both are banned as type annotations. `object` erases all structural information (zero callable attributes, zero narrowing). Use `Protocol` (structural typing), `TypeVar` (generic pass-through), explicit union (known variants), or `TypedDict` (dict shapes). +12. **No cast** — `cast()` is banned. Redesign the types. +13. **No type: ignore** — fix the type error. The checker is right; you are wrong. +14. **No broad except** — `except Exception` and `except BaseException` are banned. Catch the **specific** exception you expect. A broad catch swallows bugs you need to see — `KeyError`, `AttributeError`, `TypeError` all vanish silently. If you genuinely need a catch-all at a top-level boundary (CLI entry, HTTP handler), use `# noqa: BROAD_EXCEPT_OK` and log + re-raise. + +### Typing and safety + +- `basedpyright` in `typeCheckingMode = "all"`. Every public function has full annotations. Internal helpers: annotate return type; parameter types may be inferred. +- `ruff` with `select = ["ALL"]`. Override specific rules per project in `pyproject.toml`, never globally disable the strict baseline. +- Every new function must have a `docstring` unless its name + signature makes it completely obvious (e.g. `def full_name(first: str, last: str) -> str:`). +- Use `X | Y` union syntax (PEP 604), never `Union[X, Y]` or `Optional[X]`. + +### Why `object` is banned + +`object` pretends to be safe ("it's the top type!") but gives **zero** narrowing and **zero** attributes. Even `Any` is more honest — it admits the boundary is untyped. + +```python +# BANNED +def process(data: object) -> object: ... +def store(items: list[object]) -> None: ... +results: dict[str, object] = {} + +# GOOD — Protocol for structural typing +class Serializable(Protocol): + def serialize(self) -> bytes: ... +def process(data: Serializable) -> ProcessResult: ... + +# GOOD — TypeVar for generic pass-through +def identity[T](x: T) -> T: ... +def first[T](items: Sequence[T]) -> T: ... + +# GOOD — explicit union for known variants +def parse(raw: str | bytes) -> Document: ... +``` + +### Why `if/elif` on variants is banned + +`if/elif/else` chains on type, enum, or literal values lose compile-time exhaustiveness. When a new variant is added, nothing warns you. `match/case` + `assert_never` does. + +```python +# BANNED — if/elif for type discrimination +if isinstance(event, Click): + handle_click(event.x, event.y) +elif isinstance(event, Scroll): + handle_scroll(event.delta) +else: + raise ValueError(f"Unknown: {event}") # runtime bomb + +# BANNED — if/elif for enum discrimination +if status == Status.PENDING: + start_review() +elif status == Status.ACTIVE: + continue_processing() +elif status == Status.CLOSED: + archive() + +# BANNED — non-exhaustive match (swallows new variants) +match event: + case Click(x, y): handle_click(x, y) + case _: pass + +# GOOD — exhaustive match with assert_never +match event: + case Click(x=x, y=y): + handle_click(x, y) + case Scroll(delta=delta): + handle_scroll(delta) + case unreachable: + assert_never(unreachable) + +# GOOD — enum match +match status: + case Status.PENDING: start_review() + case Status.ACTIVE: continue_processing() + case Status.CLOSED: archive() + case unreachable: assert_never(unreachable) +``` + +`if/else` is fine for boolean conditions and range checks — things that aren't variant discrimination: + +```python +# FINE — boolean, not variant +if age >= 18: + grant_access() +else: + deny_access() +``` + +### Why broad `except` is banned + +`except Exception` catches **every** non-system exception — `KeyError`, `TypeError`, `AttributeError`, `ValueError` all vanish. You lose the stack trace that would have told you exactly what went wrong. The fix is always to name the exception you expect. + +```python +# BANNED — swallows bugs +try: + result = api.fetch(url) +except Exception as e: + logger.error(e) + return None + +# BANNED — catch-and-ignore +try: + parse(data) +except Exception: + pass + +# GOOD — catch what you expect +try: + result = api.fetch(url) +except httpx.HTTPStatusError as e: + logger.error("API %d: %s", e.response.status_code, e.request.url) + return None +except httpx.ConnectError: + raise ServiceUnavailableError(service="api") from None + +# GOOD — top-level boundary (only place broad catch is acceptable) +def main() -> int: # noqa: BROAD_EXCEPT_OK + try: + return run() + except Exception: + logger.exception("unhandled error") + return 1 +``` + +### Async + +- `import asyncio` is **BANNED**. Use `import anyio`. +- For background tasks, use `anyio.create_task_group`. Never fire-and-forget with `asyncio.create_task`. +- For concurrency gates, use `anyio.CapacityLimiter` (not `asyncio.Semaphore`). +- Load `async-anyio.md` when writing async code for the full pattern library. + +### Data modeling — which container, when + +All model fields carry type annotations. No `Any`, no untyped dicts in public APIs. +Use `polars` + `duckdb` for data. pandas is never the right answer in this stack. + +| Situation | Use | +|---|---| +| User input, API request/response | `Pydantic BaseModel (frozen=True)` | +| Internal value object (no I/O) | `@dataclass(frozen=True, slots=True)` | +| Function with multiple outcomes | Union of frozen dataclasses + `match` | +| Dict shape for JSON compat / `**kwargs` | `TypedDict` | +| Fixed constants | `StrEnum` / `IntEnum` | +| Distinct primitive (UserId vs MovieId) | `NewType` | +| Contract / capability | `Protocol` | +| Contract + shared implementation | `ABC` | +| ORM model (SQLAlchemy) | `Mapped[]` — inherently mutable, `# noqa: MUTABLE_OK` | +| Config from env vars | `pydantic-settings BaseSettings` | + +**The one rule**: data crosses trust boundary → Pydantic. Everything else → dataclass. + +Load `data-modeling.md` for the full decision flowchart and comparison matrix. + +### When frozen=True does not apply + +- **ORM models** — SQLAlchemy `Mapped[]` requires mutation. Use `# noqa: MUTABLE_OK`. +- **Builder / accumulator** — object exists to be mutated (counter, buffer, state machine). Docstring must explain why. +- **Pydantic Settings** — tests override fields. Mutable is acceptable. + +If you need `# noqa: MUTABLE_OK`, the class docstring must say why mutation is required. + +### Libraries + +Canonical defaults (override only if `pyproject.toml` explicitly picks something else): + +| Domain | Library | Reason | +|---|---|---| +| CLI | `typer` | Type-annotated CLI from function sigs | +| Pretty output | `rich` | Tables, progress, tracebacks, markdown | +| HTTP client | [`httpx2`](https://github.com/pydantic/httpx2) | Next-gen HTTP client (Pydantic stewardship), HTTP/2, brotli+zstd. Always `httpx2[http2,brotli,zstd]`. See `httpx2-optimization.md` | +| Validation | `pydantic` v2 | Fast native validator, JSON Schema | +| Web API | `fastapi` | Async, Pydantic-native, OpenAPI | +| ORM | `sqlalchemy` 2.x async | `Mapped[]` types, async sessions | +| DB driver (Postgres) | `asyncpg` (via SQLAlchemy) | Fastest PG driver | +| AI agents | `pydantic-ai` | Typed deps, structured output | +| TUI | `textual` | Rich-based, CSS layout, widgets | +| Logging | `rich.logging.RichHandler` | Pretty; swap to `structlog` in prod | + +## pyproject.toml — the one true config + +Scaffold a new project with all strict defaults pre-configured: + +```bash +uv run ../../scripts/python/new-project.py myproject +uv run ../../scripts/python/new-project.py myproject --path ./workspace +uv run ../../scripts/python/new-project.py myproject --lib # publishable library +``` + +Creates via `uv init`, then injects basedpyright `typeCheckingMode = "all"` + ruff `select = ["ALL"]` + pytest strict. Cross-platform (macOS, Linux, Windows). + +For manual setup: `uv init --app myproject`, then load `pyproject-strict.md`. + +## PEP 723 — inline script metadata (mandatory for ALL scripts) + +Every `.py` script — even throwaway — MUST use PEP 723 inline metadata with the `# ─── How to run ───` comment block. No venv, no `requirements.txt`. The script IS the environment spec. A script without the usage comment block is incomplete. + +Scaffold with: `uv run ../../scripts/python/new-script.py --deps "httpx2[http2,brotli,zstd]"` (writes to temp dir by default, `--output` for specific path). + +Load `one-liners.md` for full patterns, examples, and anti-patterns. + +## Reference loading + +Load on demand — not all at once. + +| Need | Load | +|---|---| +| Full pyproject.toml config | `pyproject-strict.md` | +| Type patterns (NewType, Final, enums, narrowing) | `type-patterns.md` | +| Data modeling (container choice, frozen, parse-don't-validate) | `data-modeling.md` | +| Error handling (typed errors, union returns, exhaustive match) | `error-handling.md` | +| Async patterns (anyio) | `async-anyio.md` | +| Data processing (polars / duckdb) | `data-processing.md` | +| FastAPI + SQLAlchemy stack | `fastapi-stack.md` | +| Library decision tree | `libraries.md` | +| **httpx2 optimization** (MUST load for any network code) | `httpx2-optimization.md` | +| **orjson** (when JSON is in the hot path; FastAPI/Pydantic v2 integration) | `orjson-stack.md` | +| One-liner scripts (PEP 723) | `one-liners.md` | +| PydanticAI agents | `pydantic-ai.md` | +| Textual TUI | `textual-tui.md` | + +## httpx2 — mandatory for ALL network requests + +Every outgoing HTTP call MUST use [`httpx2`](https://github.com/pydantic/httpx2) (`httpx2[http2,brotli,zstd]`). Never `requests`, never `aiohttp`, never the original `httpx`. + +**ALL optimizations are ON by default — not optional, not progressive, not "nice to have".** A bare `httpx2.AsyncClient()` is a bug — treat it like a lint violation. The correct way is the factory pattern in `httpx2-optimization.md` with: HTTP/2 enabled, tuned connection pool (200/40/30s), split timeouts (5/30/10/10), transport retries (3), TCP_NODELAY, follow_redirects, and event hooks for observability. + +When writing or reviewing ANY network code, **ALWAYS load `httpx2-optimization.md`** and use the factory pattern verbatim. No exceptions. + +## No-excuse audit + +Violations caught by `../../scripts/python/check-no-excuse-rules.py`. Run after every edit session. + +| Rule ID | Catches | Opt-out | +|---|---|---| +| `cast-any` | `cast(Any, ...)` | None — redesign types | +| `type-ignore` | `# type: ignore` | None — fix the type | +| `pyright-ignore` | `# pyright: ignore` | None — fix the type | +| `bare-except` | `except:` with no class | None — name the exception | +| `silent-except` | `except X: pass` / `except X: ...` | None — handle or re-raise | +| `no-asyncio` | `import asyncio` | `# noqa: ANYIO_OK` | +| `no-pandas` | `import pandas` | `# noqa: PANDAS_OK` | +| `mutable-dataclass` | `@dataclass` without `frozen=True` | `# noqa: MUTABLE_OK` | +| `missing-slots` | `@dataclass` without `slots=True` | `# noqa: SLOTS_OK` | +| `raw-dict-return` | `-> dict` in function return type | `# noqa: DICT_OK` | +| `missing-assert-never` | `match` block without `assert_never` default | `# noqa: MATCH_OK` | +| `generic-exception` | `raise ValueError("...")` / `raise TypeError("...")` with bare string | `# noqa: GENERIC_ERR_OK` | +| `no-object` | `object` used as type annotation (param, return, generic arg) | `# noqa: OBJECT_OK` | +| `if-elif-on-variant` | `if isinstance()`/`if x == Enum.V` chain that should be `match/case` | `# noqa: IF_VARIANT_OK` | +| `oversized-module` | File exceeds 250 pure LOC (non-blank, non-comment) | `# noqa: SIZE_OK` | +| `broad-except` | `except Exception` / `except BaseException` (too broad) | `# noqa: BROAD_EXCEPT_OK` | + +Fix every violation before declaring work done. basedpyright + ruff strict config catches the rest. + +## In tests + +Tests are strict too, with these exceptions (already configured in `pyproject.toml` per-file-ignores): + +| In tests you may | Why | +|---|---| +| Use `assert` | That's how pytest works (`S101` ignored) | +| Use magic numbers | Test data (`PLR2004` ignored) | +| Access `_private` members | Testing internals (`SLF001` ignored) | +| Skip docstrings | Test names are the docs (`D` ignored) | +| Have unused function args | Fixtures (`ARG` ignored) | + +Tests still follow the iron list — frozen dataclasses, typed errors, exhaustive match. If test fixtures need mutable state, use `# noqa: MUTABLE_OK` on the fixture class. + +## Existing codebases + +When editing an existing file that doesn't follow these rules: **write new code in strict style, don't refactor existing code in the same change.** Mixing feature work with style migration makes reviews harder and bugs likelier. + +## Activation + +This skill activates whenever you are writing or modifying any `.py` file. Even one-off scripts get the strict treatment — that is the whole point of PEP 723 + uv: production hygiene with throwaway ergonomics. diff --git a/packages/omo-codex/plugin/skills/programming/references/python/async-anyio.md b/packages/omo-codex/plugin/skills/programming/references/python/async-anyio.md new file mode 100644 index 000000000..1af6df53b --- /dev/null +++ b/packages/omo-codex/plugin/skills/programming/references/python/async-anyio.md @@ -0,0 +1,442 @@ +# AnyIO Reference: Replacing asyncio Idioms + +> **Skill mandate**: `import asyncio` is BANNED. Use `import anyio` exclusively. +> This reference targets AnyIO 4.x (2026 Python projects). + +--- + +## 1. Task Groups (The Core Primitive) + +AnyIO uses **structured concurrency** via task groups. A task group is an async context manager that guarantees all child tasks finish before the block exits. + +### `start_soon` — fire-and-forget + +```python +import anyio + +async def worker(n: int) -> None: + await anyio.sleep(1) + print(f"task {n} done") + +async def main() -> None: + async with anyio.create_task_group() as tg: + for i in range(3): + tg.start_soon(worker, i) + print("all tasks finished") + +anyio.run(main) +``` + +**Signature**: `tg.start_soon(func, *args, name=None)` +- `func` must be a **coroutine function** (not a coroutine object). +- `name` is optional, for introspection/debugging. +- No return value; exceptions propagate as `ExceptionGroup` on exit. + +### `start` — wait for ready signal + +Use when a task must initialize before the caller proceeds (e.g., starting a server and then connecting to it). + +```python +from anyio import TASK_STATUS_IGNORED, create_task_group, run +from anyio.abc import TaskStatus + +async def start_server(port: int, *, task_status: TaskStatus[None] = TASK_STATUS_IGNORED) -> None: + listener = await anyio.create_tcp_listener(local_host="127.0.0.1", local_port=port) + task_status.started() # unblocks tg.start() + await listener.serve(handler) + +async def main() -> None: + async with create_task_group() as tg: + await tg.start(start_server, 8080) # blocks until task_status.started() + # server is guaranteed ready here + async with await anyio.connect_tcp("127.0.0.1", 8080) as client: + ... + +run(main) +``` + +**Rule of thumb**: +- Use `start_soon` when you don't need to know when the task is ready. +- Use `start` when the task must signal readiness before you continue. + +### `create_task` — retrieving return values (AnyIO 4.14+) + +```python +async def add(x: int, y: int) -> int: + return x + y + +async def main() -> None: + async with anyio.create_task_group() as tg: + handle = tg.create_task(add(2, 4)) + result = await handle # == 6 + print(handle.return_value) # also 6 + +anyio.run(main) +``` + +**Signature**: `tg.create_task(coro, *, name=None, context=None) -> TaskHandle[T]` +- Returns a `TaskHandle` you can `await` for the result. +- If the task raises, awaiting raises `TaskFailed` (or `TaskCancelled`). +- This is the canonical replacement for `asyncio.gather` when you need results. + +--- + +## 2. asyncio → anyio Cheat Sheet + +| asyncio | anyio | Notes | +|---------|-------|-------| +| `asyncio.gather(a, b, c)` | `tg.create_task(a); tg.create_task(b); tg.create_task(c); results = [await h for h in handles]` | No direct gather; structured concurrency requires explicit task group scope. For fire-and-forget, use `tg.start_soon`. | +| `asyncio.create_task(coro)` | `tg.start_soon(func, *args)` or `tg.create_task(coro)` | `start_soon` takes a coroutine **function** + args. `create_task` takes a coroutine **object** and returns a handle. | +| `asyncio.sleep(n)` | `anyio.sleep(n)` | Identical semantics. | +| `asyncio.wait_for(coro, timeout)` | `with anyio.fail_after(timeout): await coro` | Raises `TimeoutError`. Use `move_on_after` for silent timeout. | +| `asyncio.Event()` | `anyio.Event()` | AnyIO events are **not reusable**; create a new one instead of `.clear()`. | +| `asyncio.Lock()` | `anyio.Lock()` | Use `async with lock:`. Pass `fast_acquire=True` if performance-critical. | +| `asyncio.Semaphore(n)` | `anyio.Semaphore(n)` | Same. Pass `fast_acquire=True` if performance-critical. | +| `asyncio.Condition()` | `anyio.Condition()` | Same semantics. | +| `asyncio.run(main())` | `anyio.run(main)` | Backend-agnostic entry point. | +| `asyncio.Queue(maxsize=N)` | `anyio.create_memory_object_stream[T](max_buffer_size=N)` | Returns `(send_stream, receive_stream)`. Supports `async for` on receive end. | +| `asyncio.to_thread(fn, *args)` | `anyio.to_thread.run_sync(fn, *args)` | Supports `abandon_on_cancel=True` and custom `limiter`. | +| `asyncio.run_coroutine_threadsafe(coro, loop)` | `anyio.from_thread.run(func, *args)` | Call async code from a worker thread. | +| `loop.call_soon_threadsafe(callback)` | `anyio.from_thread.run_sync(func, *args)` | Call sync code in event loop thread from worker thread, **with return value**. | +| `asyncio.shield(coro)` | `with anyio.CancelScope(shield=True): ...` | AnyIO shielding does not orphan tasks. | +| `asyncio.timeout(delay)` | `with anyio.fail_after(delay): ...` | AnyIO uses level cancellation, not edge cancellation. | +| `asyncio.CancelledError` | `anyio.get_cancelled_exc_class()` | Use this to catch cancellation portably across backends. | + +--- + +## 3. Cancellation & CancelScope + +AnyIO uses **level cancellation** (inspired by Trio), not asyncio's **edge cancellation**. + +- **Edge cancellation** (asyncio): A `CancelledError` is injected once. If caught and not re-raised, the task keeps running. +- **Level cancellation** (anyio): As long as a task is inside an effectively cancelled scope, every yield point raises a new cancellation exception. + +### Basic CancelScope + +```python +from anyio import CancelScope, create_task_group, get_cancelled_exc_class, sleep, run + +async def worker() -> None: + try: + await sleep(10) + except get_cancelled_exc_class(): + print("cancelled!") + raise # ALWAYS re-raise cancellation exceptions + +async def main() -> None: + async with create_task_group() as tg: + tg.start_soon(worker) + await sleep(0.1) + tg.cancel_scope.cancel() # cancels all children + +run(main) +``` + +### Shielding + +Shield a block from external cancellation. Essential for cleanup. + +```python +from anyio import CancelScope, create_task_group, sleep, run + +async def main() -> None: + async with create_task_group() as tg: + with CancelScope(shield=True): + tg.start_soon(some_task) + tg.cancel_scope.cancel() # shielded block is protected + await sleep(1) # this still runs + +run(main) +``` + +**Combine with timeouts for graceful shutdown**: + +```python +from anyio import CancelScope, move_on_after + +async def do_something(resource) -> None: + try: + await run_async_stuff() + except BaseException: + # Allow up to 10s for cleanup, then move on + with move_on_after(10, shield=True): + await resource.aclose() + raise +``` + +### Structured Concurrency Guarantee + +A task group contains its own `CancelScope`. If any child task raises an exception: +1. The task group's cancel scope is cancelled. +2. All other child tasks receive cancellation. +3. The task group waits for all children to finish. +4. The original exception (wrapped in `ExceptionGroup` if multiple) is re-raised. + +--- + +## 4. Timeouts + +Two context managers. Both create a `CancelScope` internally. + +### `fail_after` — raises on timeout + +```python +from anyio import fail_after, sleep, run + +async def main() -> None: + try: + with fail_after(5) as scope: + await sleep(10) + except TimeoutError: + print("timed out") + print(scope.cancelled_caught) # True + +run(main) +``` + +### `move_on_after` — silent timeout + +```python +from anyio import move_on_after, sleep, run + +async def main() -> None: + with move_on_after(5) as scope: + await sleep(10) + print("this never prints") + + print("exited scope, cancelled =", scope.cancelled_caught) + +run(main) +``` + +### Combined with shielding + +```python +from anyio import move_on_after + +# Give cleanup 10 seconds, but don't let outer cancellation interrupt it +with move_on_after(10, shield=True): + await resource.aclose() +``` + +--- + +## 5. Memory Object Streams (Queue Replacement) + +Replaces `asyncio.Queue` with a safer, typed, structured-concurrency-friendly construct. + +```python +from anyio import create_task_group, create_memory_object_stream, run +from anyio.streams.memory import MemoryObjectReceiveStream + +async def consumer(stream: MemoryObjectReceiveStream[str]) -> None: + async with stream: # closes receive end on exit + async for item in stream: + print("received", item) + +async def main() -> None: + # Type-annotated stream creation (AnyIO 4+ syntax) + send_stream, receive_stream = create_memory_object_stream[str](max_buffer_size=10) + + async with create_task_group() as tg: + tg.start_soon(consumer, receive_stream) + async with send_stream: + for i in range(5): + await send_stream.send(f"item {i}") + # send_stream closed → consumer's async for loop exits naturally + +run(main) +``` + +**Key differences from `asyncio.Queue`**: +- **Bounded by default**: `max_buffer_size=0` means send blocks until a receiver is ready. +- **Cloneable**: Each producer/consumer can close its own clone. The stream only ends when **all** clones of one end are closed. +- **Async iterable**: `async for item in receive_stream:` works out of the box. +- **Type-safe**: Generic `create_memory_object_stream[T]()`. +- **Synchronous close**: Both `close()` and `async with` work. + +--- + +## 6. Backend Selection + +AnyIO is backend-agnostic. Code written against AnyIO APIs runs on both asyncio and Trio. + +```python +import anyio + +async def main() -> None: + print("running on", anyio.current_async_library()) + await anyio.sleep(1) + +# Default backend (asyncio) +anyio.run(main) + +# Explicit backend +anyio.run(main, backend="trio") +anyio.run(main, backend="asyncio", backend_options={"debug": True}) +``` + +**Library design rule**: Never hardcode a backend. Let the application choose via `anyio.run()`. Libraries should only import `anyio` and avoid backend-specific APIs. + +--- + +## 7. Compatibility with asyncio-only libraries + +### Using asyncio libraries under the asyncio backend + +If a third-party library exposes only an asyncio interface (returns asyncio coroutine objects), it works directly under the asyncio backend because AnyIO runs on top of asyncio's event loop: + +```python +import anyio +import some_asyncio_only_lib # returns asyncio.Future/coroutine objects + +async def main() -> None: + # This works because under the asyncio backend, await passes through + result = await some_asyncio_only_lib.fetch_data() + +anyio.run(main, backend="asyncio") +``` + +**Important**: This only works on the `asyncio` backend. On the `trio` backend, asyncio-native objects will not work. + +### When you MUST use asyncio APIs + +Some APIs have no AnyIO equivalent and require direct event loop access: + +| Scenario | asyncio API | AnyIO approach | +|----------|-------------|----------------| +| Signal handlers | `loop.add_signal_handler()` | `anyio.open_signal_receiver()` | +| Custom protocols | `asyncio.Protocol` | Use AnyIO streams / sockets | +| Direct Future manipulation | `asyncio.Future` | Avoid; use AnyIO primitives | +| Eager task factories | `asyncio.eager_task_factory` | Experimental in AnyIO; avoid | + +If you absolutely need the running loop: + +```python +import asyncio + +async def main() -> None: + loop = asyncio.get_running_loop() + # ... do something loop-specific ... + # WARNING: this breaks backend-agnosticism + +anyio.run(main, backend="asyncio") +``` + +**Best practice**: Wrap asyncio-only code in a backend-agnostic facade, and document that the feature requires the asyncio backend. + +--- + +## 8. Idiomatic Code Snippets + +### Snippet 1: Parallel HTTP requests with timeout and cleanup + +```python +import anyio + +async def fetch(url: str) -> bytes: + await anyio.sleep(0.5) # simulate + return b"data" + +async def main() -> None: + urls = ["a", "b", "c"] + async with anyio.create_task_group() as tg: + with anyio.move_on_after(5): + for url in urls: + tg.start_soon(fetch, url) + # All tasks are cancelled on timeout; task group waits for cleanup + +anyio.run(main) +``` + +### Snippet 2: Producer-consumer with memory object stream + +```python +import anyio +from anyio.streams.memory import MemoryObjectReceiveStream + +async def producer(send_stream: anyio.streams.memory.MemoryObjectSendStream[int]) -> None: + async with send_stream: + for i in range(100): + await send_stream.send(i) + +async def consumer(receive_stream: MemoryObjectReceiveStream[int]) -> None: + async with receive_stream: + async for item in receive_stream: + print(f"consumed {item}") + +async def main() -> None: + send, receive = anyio.create_memory_object_stream[int](max_buffer_size=5) + async with anyio.create_task_group() as tg: + tg.start_soon(producer, send) + tg.start_soon(consumer, receive) + +anyio.run(main) +``` + +### Snippet 3: Calling sync code from async + +```python +import time +import anyio + +async def main() -> None: + # Run blocking function in worker thread + result = await anyio.to_thread.run_sync(time.sleep, 2) + print("done") + +anyio.run(main) +``` + +### Snippet 4: Calling async code from a worker thread + +```python +import anyio + +def blocking_callback() -> None: + # Inside a worker thread, call back into the event loop + anyio.from_thread.run(anyio.sleep, 1) + anyio.from_thread.run_sync(print, "hello from thread") + +async def main() -> None: + await anyio.to_thread.run_sync(blocking_callback) + +anyio.run(main) +``` + +### Snippet 5: Graceful shutdown with shielded cleanup + +```python +import anyio + +async def worker() -> None: + try: + await anyio.sleep_forever() + except anyio.get_cancelled_exc_class(): + with anyio.CancelScope(shield=True): + await anyio.sleep(0.5) # cleanup + print("cleaned up") + raise + +async def main() -> None: + async with anyio.create_task_group() as tg: + tg.start_soon(worker) + await anyio.sleep(1) + tg.cancel_scope.cancel() + +anyio.run(main) +``` + +--- + +## Sources + +- AnyIO Documentation (stable): https://anyio.readthedocs.io/en/stable/ +- AnyIO GitHub (HEAD `cb245dba`): https://github.com/agronholm/anyio +- Task Groups: https://anyio.readthedocs.io/en/stable/tasks.html +- Cancellation & Timeouts: https://anyio.readthedocs.io/en/stable/cancellation.html +- Streams: https://anyio.readthedocs.io/en/stable/streams.html +- Synchronization: https://anyio.readthedocs.io/en/stable/synchronization.html +- Threads: https://anyio.readthedocs.io/en/stable/threads.html +- Basics / Backends: https://anyio.readthedocs.io/en/stable/basics.html +- Design Rationale (why asyncio is problematic): https://anyio.readthedocs.io/en/stable/why.html diff --git a/packages/omo-codex/plugin/skills/programming/references/python/data-modeling.md b/packages/omo-codex/plugin/skills/programming/references/python/data-modeling.md new file mode 100644 index 000000000..48b46f8eb --- /dev/null +++ b/packages/omo-codex/plugin/skills/programming/references/python/data-modeling.md @@ -0,0 +1,233 @@ +# Data Modeling + +Which container to use, how to structure data, and why frozen is the default. + +--- + +## Decision flowchart + +``` +Is it a fixed set of named constants? + YES → StrEnum / IntEnum + NO ↓ +Is it just branding a primitive (int, str, float)? + YES → NewType("X", base) + NO ↓ +Is it an interface / contract ("this thing can do X")? + ├─ Shape only, no shared code → Protocol + └─ Shared method implementation needed → ABC + NO ↓ +Does the data cross a trust boundary (user input, API, file, external DB)? + YES → pydantic.BaseModel (frozen=True) — validates + serializes + NO ↓ +Is it a dict shape needed for JSON compat / **kwargs typing? + YES → TypedDict + NO ↓ +Is it structured data with named fields? + YES → @dataclass(frozen=True, slots=True) + NO ↓ +Is it a tuple with positional semantics (x, y coords / DB row)? + YES → NamedTuple + NO → you probably don't need a new type +``` + +--- + +## Container reference + +### @dataclass — internal value object + +The default for structured data inside your codebase. Zero overhead, no framework coupling. + +```python +from dataclasses import dataclass +from typing import NewType + +UserId = NewType("UserId", int) + +@dataclass(frozen=True, slots=True) +class User: + id: UserId + name: str + email: str + +@dataclass(frozen=True, slots=True) +class Point: + x: float + y: float +``` + +Always `frozen=True, slots=True`. Mutable only when mutation is the documented purpose — opt out with `# noqa: MUTABLE_OK`. + +### Pydantic BaseModel — trust boundary guardian + +Use when data enters or leaves your system. Validates at construction, serializes to JSON, generates OpenAPI schema. + +```python +from pydantic import BaseModel, ConfigDict, EmailStr + +class CreateUserRequest(BaseModel): + model_config = ConfigDict(frozen=True) + + name: str + email: EmailStr + age: int + +class UserResponse(BaseModel): + model_config = ConfigDict(frozen=True) + + id: int + name: str + email: str +``` + +**The one rule**: data crosses a trust boundary → Pydantic. Everything else → dataclass. +Never use Pydantic for internal-only data just because it's convenient. The validation cost is real. + +### TypedDict — dict that knows its shape + +Use when the value must stay a `dict` at runtime — JSON blobs, `**kwargs`, third-party APIs expecting dicts. + +```python +from typing import TypedDict, NotRequired + +class Headers(TypedDict): + content_type: str + authorization: NotRequired[str] + +def make_request(url: str, headers: Headers) -> None: ... + +make_request("https://api.example.com", {"content_type": "application/json"}) +``` + +### Protocol — structural interface + +"Anything that has method X" — no inheritance required. + +```python +from typing import Protocol + +class Renderable(Protocol): + def render(self) -> str: ... + +class Saveable(Protocol): + async def save(self) -> None: ... + +@dataclass(frozen=True, slots=True) +class MarkdownDoc: + content: str + def render(self) -> str: + return self.content + +def publish(doc: Renderable) -> None: + print(doc.render()) # MarkdownDoc works — no inheritance needed +``` + +Default to Protocol for interfaces. ABC only when you need shared method implementations. + +### ABC — interface with shared code + +Only when Protocol isn't enough. + +```python +from abc import ABC, abstractmethod + +class BaseRepository(ABC): + @abstractmethod + async def get(self, id: int) -> Model | None: ... + + @abstractmethod + async def save(self, model: Model) -> None: ... + + async def get_or_raise(self, id: int) -> Model: + result = await self.get(id) + if result is None: + msg = f"{type(self).__name__}: id {id} not found" + raise LookupError(msg) + return result +``` + +### NamedTuple — positional + named (rare) + +Only when you need tuple protocol (unpacking, indexing). + +```python +from typing import NamedTuple + +class Coordinate(NamedTuple): + x: float + y: float + +x, y = Coordinate(1.0, 2.0) # tuple unpacking +``` + +99% of the time, `@dataclass(frozen=True, slots=True)` is better. + +--- + +## Quick lookup + +| Situation | Use | Why | +|---|---|---| +| User input, API request/response | `Pydantic BaseModel` | Validation, JSON schema, serialization | +| DB row ↔ Python (ORM) | SQLAlchemy `Mapped[]` model | ORM integration, async session | +| Internal value object | `@dataclass(frozen=True, slots=True)` | Zero overhead, no validation needed | +| Multiple outcomes from function | Union of frozen dataclasses | Distinct types for `match` | +| Dict shape for JSON / `**kwargs` | `TypedDict` | Stays a dict at runtime | +| Fixed constants | `StrEnum` / `IntEnum` | Exhaustive match, no typos | +| Distinct primitive | `NewType("X", int)` | Zero runtime cost, type-level only | +| Contract / capability | `Protocol` | Structural typing, no inheritance | +| Contract + shared impl | `ABC` | When Protocol isn't enough | + +--- + +## Comparison matrix + +| Feature | dataclass | Pydantic | TypedDict | Protocol | NamedTuple | NewType | Enum | +|---|---|---|---|---|---|---|---| +| Validation | - | ✓ | - | - | - | - | - | +| JSON serialization | manual | built-in | native dict | - | - | - | `.value` | +| Immutable | frozen=True | frozen=True | - (dict) | N/A | always | N/A | always | +| Runtime cost | ~zero | validation | zero | zero | ~zero | zero | ~zero | +| `match` support | ✓ | ✓ | - | - | ✓ | - | ✓ | +| `slots` support | ✓ | - | - | - | - | - | - | + +--- + +## Parse, don't validate + +Validate at the boundary. Inside the boundary, types are proof of validity. + +```python +# BAD — validate then pass raw data +def process_email(email: str) -> None: + if "@" not in email: + raise ValueError("invalid email") + # still a raw str everywhere downstream + +# GOOD — parse into typed value at boundary +from typing import NewType + +Email = NewType("Email", str) + +def parse_email(raw: str) -> Email: + if "@" not in raw or "." not in raw.split("@")[1]: + msg = f"invalid email: {raw}" + raise ValueError(msg) + return Email(raw.lower().strip()) + +# Downstream only sees Email, never raw str +def send_welcome(email: Email) -> None: ... +``` + +With Pydantic this happens automatically — `EmailStr` is already a parsed type. Once constructed, `.email` is always valid. No re-validation needed. + +--- + +## Sources + +- Python docs: [dataclasses](https://docs.python.org/3/library/dataclasses.html) +- Pydantic v2: [docs.pydantic.dev](https://docs.pydantic.dev/latest/) +- Python docs: [typing — Protocol](https://docs.python.org/3/library/typing.html#typing.Protocol) +- Python docs: [typing — TypedDict](https://docs.python.org/3/library/typing.html#typing.TypedDict) +- Alexis King: [Parse, don't validate](https://lexi-lambda.github.io/blog/2019/11/05/parse-don-t-validate/) diff --git a/packages/omo-codex/plugin/skills/programming/references/python/data-processing.md b/packages/omo-codex/plugin/skills/programming/references/python/data-processing.md new file mode 100644 index 000000000..299b30cd1 --- /dev/null +++ b/packages/omo-codex/plugin/skills/programming/references/python/data-processing.md @@ -0,0 +1,133 @@ +# Data Processing — Polars + DuckDB + +## The rule + +NEVER pandas. Polars (with numpy) plus DuckDB. Pandas is 10-50x slower, has weaker types, and the modern Python data ecosystem has moved on. + +## Quick decision tree + +| Operation | Use | Why | +|---|---|---| +| `.csv` / `.parquet` / `.json` direct query | DuckDB | Zero memory load, SQL ergonomics | +| `.duckdb` file | DuckDB | Native format | +| Filter (any size) | Polars | 128x faster than DuckDB for filtering | +| Sort | Polars | 12x faster | +| Multi-table join | DuckDB | 3x faster, more join types | +| Heavy GROUP BY aggregation | DuckDB | 4x faster on large datasets | +| Window function | Polars | 3-5x faster | +| Pivot / melt / string ops | Polars | 2x faster | +| Larger than RAM | Polars streaming or DuckDB out-of-core | Both handle OOM | +| Mixed pipeline | Hybrid (zero-copy via Arrow) | Use each tool's strengths | + +For the deep version (per-operation benchmarks, OOM strategies, full execution templates), load the **`data-scientist`** skill - it lives in this same skill set and is the source of truth for performance numbers. + +## Standard imports + +```python +import numpy as np +import polars as pl +import duckdb +``` + +## DuckDB direct file query (zero memory load) + +```python +result = duckdb.sql(""" + SELECT category, SUM(amount) AS total + FROM 'data.csv' + WHERE date >= '2026-01-01' + GROUP BY category + ORDER BY total DESC +""").pl() # zero-copy → Polars DataFrame +``` + +`.pl()` returns Polars; `.df()` would return pandas - never use `.df()`. + +## Polars lazy pipeline + +```python +result = ( + pl.scan_csv("data.csv") # lazy, no read yet + .filter(pl.col("amount") > 1000) + .filter(pl.col("status") == "active") + .sort("amount", descending=True) + .head(100) + .collect() # execute optimised plan +) +``` + +`scan_*` over `read_*` for files; `lazy()` then `collect()` for in-memory frames. Polars optimises the entire plan before execution (predicate pushdown, projection pushdown, common subexpression elimination). + +## Streaming for OOM data + +```python +result = ( + pl.scan_csv("huge.csv") + .filter(pl.col("active")) + .group_by("category") + .agg([ + pl.len().alias("count"), + pl.sum("amount").alias("total"), + ]) + .collect(streaming=True) +) +``` + +## Hybrid pipeline (most realistic shape) + +```python +# Phase 1: DuckDB for the join (3x faster) +joined = duckdb.sql(""" + SELECT o.*, c.region, p.category + FROM 'orders.parquet' o + JOIN 'customers.parquet' c ON o.customer_id = c.id + JOIN 'products.parquet' p ON o.product_id = p.id +""").pl() + +# Phase 2: Polars for filtering and transformation (128x + 2x faster) +processed = ( + joined + .filter(pl.col("amount") > 100) + .with_columns([ + (pl.col("amount") * 1.1).alias("amount_with_tax"), + ]) +) + +# Phase 3: DuckDB for final aggregation (4x faster) - register Polars frame by name +duckdb.register("processed", processed) +final = duckdb.sql(""" + SELECT region, category, SUM(amount_with_tax) AS revenue + FROM processed + GROUP BY region, category + ORDER BY revenue DESC +""").pl() +``` + +## Type safety with Polars + +Polars supports schema overrides at read time, and `.cast()` for explicit conversion. Avoid implicit coercion in hot paths. + +```python +schema = {"id": pl.Int64, "amount": pl.Float64, "date": pl.Date} +df = pl.read_csv("data.csv", schema_overrides=schema) +``` + +basedpyright understands `polars-stubs`, which ship with polars itself. No extra type stubs to install. + +## Things you might miss from pandas (and how to do them in Polars) + +| pandas | polars | +|---|---| +| `df.iloc[5]` | `df.row(5)` (named tuple) or `df[5]` (single-row frame) | +| `df.loc[df["x"] > 5]` | `df.filter(pl.col("x") > 5)` | +| `df["x"].apply(fn)` | `df["x"].map_elements(fn)` (slow path) or use native expressions | +| `df.merge(...)` | `df.join(other, on="key")` | +| `df.groupby(...).agg(...)` | `df.group_by(...).agg(...)` | +| `pd.read_csv(...).dtypes` | `pl.read_csv(...).schema` | +| `df.to_dict("records")` | `df.to_dicts()` | + +## Sources + +- Polars docs: +- DuckDB Python API: +- Cross-reference - this skill set's `data-scientist` skill (load it for the deep version) diff --git a/packages/omo-codex/plugin/skills/programming/references/python/error-handling.md b/packages/omo-codex/plugin/skills/programming/references/python/error-handling.md new file mode 100644 index 000000000..44f58befd --- /dev/null +++ b/packages/omo-codex/plugin/skills/programming/references/python/error-handling.md @@ -0,0 +1,218 @@ +# Error Handling + +Typed errors, exhaustive matching, union returns, and resource safety. + +--- + +## Typed errors — no bare strings + +Error types carry structured data. Pattern matching works. Callers know exactly what can go wrong. + +```python +from dataclasses import dataclass +from typing import NewType + +UserId = NewType("UserId", int) + +@dataclass(frozen=True, slots=True) +class UserNotFoundError(Exception): + user_id: UserId + + def __str__(self) -> str: # REQUIRED — see note below + return f"user {self.user_id} not found" + +@dataclass(frozen=True, slots=True) +class PermissionDeniedError(Exception): + user_id: UserId + required_role: str + + def __str__(self) -> str: + return f"user {self.user_id} needs role {self.required_role}" +``` + +**`__str__` is mandatory** on dataclass exceptions. `@dataclass` replaces `Exception.__init__`, so `self.args` is always `()`. Without `__str__`, `str(e)` returns an empty string and logging/monitoring breaks. + +```python +# BAD +raise ValueError("user not found") +raise ValueError("permission denied") + +# GOOD +raise UserNotFoundError(user_id=uid) +raise PermissionDeniedError(user_id=uid, required_role="admin") +``` + +--- + +## Union returns — expected failures without exceptions + +For failures that are **expected** (not found, validation error, permission denied), return a union instead of raising. Exceptions are for **unexpected** failures (network down, OOM, corrupted data). + +### Define the outcome types + +```python +@dataclass(frozen=True, slots=True) +class User: + id: UserId + name: str + +@dataclass(frozen=True, slots=True) +class UserNotFound: + id: UserId + +@dataclass(frozen=True, slots=True) +class PermissionDenied: + id: UserId + reason: str + +type GetUserResult = User | UserNotFound | PermissionDenied +``` + +### Handle exhaustively + +```python +from typing import assert_never + +def handle_result(result: GetUserResult) -> str: + match result: + case User(name=name): + return f"Found: {name}" + case UserNotFound(id=uid): + return f"No user with id {uid}" + case PermissionDenied(reason=reason): + return f"Denied: {reason}" + case _ as unreachable: + assert_never(unreachable) +``` + +`assert_never` in the default case: if you add a new variant to `GetUserResult` without handling it here, the type checker errors. No silent fall-through. + +### When to use which + +**The heuristic**: caller is 1-2 levels away and MUST handle it → union return. Error should propagate up many layers to a boundary → exception. + +| Scenario | Pattern | Why | +|---|---|---| +| Repository → service (caller handles it) | Union return (`User \| UserNotFound`) | Caller is right there, must handle both | +| Validation at boundary (parsing input) | Exception (typed, with fields) | Propagates up to HTTP/CLI handler | +| Infrastructure failure (network, OOM) | Exception | Can't handle locally, must propagate | +| Service → service (deep internal) | Exception (typed) | Union boilerplate across many layers is worse than exceptions | +| HTTP handler → response | Catch exceptions, convert to response | Boundary code catches and translates | + +**Practical tradeoff**: union returns are safest (type checker forces handling) but create boilerplate when every caller in a chain must `match`. If the error would just propagate through 3+ layers unchanged, use a typed exception instead. + +--- + +## Exhaustive match — every match needs a default + +Every `match` statement ends with `case _: assert_never(x)`. No exceptions. + +```python +from enum import StrEnum +from typing import assert_never + +class Status(StrEnum): + PENDING = "pending" + ACTIVE = "active" + DELETED = "deleted" + +def describe(status: Status) -> str: + match status: + case Status.PENDING: + return "waiting" + case Status.ACTIVE: + return "live" + case Status.DELETED: + return "gone" + case _ as unreachable: + assert_never(unreachable) +``` + +Add a new enum member? The type checker tells you every `match` that needs updating. + +--- + +## Context managers — resource safety + +If it has `.close()`, `.shutdown()`, `.disconnect()`, or `.release()`, wrap it in `with`. + +```python +# BAD +f = open("data.txt") +data = f.read() +f.close() # forgotten? leaked + +# GOOD +from pathlib import Path + +data = Path("data.txt").read_text() +``` + +### Async resources + +```python +import httpx + +async def fetch_users() -> list[User]: + async with httpx.AsyncClient() as client: + response = await client.get("https://api.example.com/users") + response.raise_for_status() + return [User(**u) for u in response.json()] +``` + +### Custom context manager + +```python +from contextlib import asynccontextmanager +from collections.abc import AsyncIterator + +@asynccontextmanager +async def managed_connection(url: str) -> AsyncIterator[Connection]: + conn = await connect(url) + try: + yield conn + finally: + await conn.close() + +async with managed_connection("postgres://...") as conn: + await conn.execute("SELECT 1") +# conn is closed here, guaranteed +``` + +--- + +## Exception hierarchy — when you do raise + +Keep exception hierarchies shallow and specific. + +```python +class AppError(Exception): + """Base for all application errors.""" + +@dataclass(frozen=True, slots=True) +class NotFoundError(AppError): + entity: str + id: int + + def __str__(self) -> str: + return f"{self.entity} {self.id} not found" + +@dataclass(frozen=True, slots=True) +class ConflictError(AppError): + entity: str + field: str + value: str + + def __str__(self) -> str: + return f"{self.entity}.{self.field} = {self.value!r} already exists" +``` + +Callers catch `AppError` at the boundary, or specific subtypes where they can do something useful. + +--- + +## Sources + +- Python docs: [typing — assert_never](https://docs.python.org/3/library/typing.html#typing.assert_never) +- Python docs: [contextlib](https://docs.python.org/3/library/contextlib.html) +- Python docs: [match statement](https://docs.python.org/3/reference/compound_stmts.html#the-match-statement) diff --git a/packages/omo-codex/plugin/skills/programming/references/python/fastapi-stack.md b/packages/omo-codex/plugin/skills/programming/references/python/fastapi-stack.md new file mode 100644 index 000000000..d5180614f --- /dev/null +++ b/packages/omo-codex/plugin/skills/programming/references/python/fastapi-stack.md @@ -0,0 +1,316 @@ +# FastAPI + SQLAlchemy 2.x async + Postgres + Pydantic v2 + +The canonical web API stack. Async end-to-end, type-safe end-to-end, OpenAPI-generated end-to-end. + +## Project layout + +``` +myapi/ +├── pyproject.toml +├── alembic.ini +├── migrations/ +│ └── env.py +├── src/ +│ └── myapi/ +│ ├── __init__.py +│ ├── main.py # FastAPI app + lifespan +│ ├── config.py # pydantic-settings +│ ├── db.py # engine, session factory, dependency +│ ├── models.py # SQLAlchemy declarative models +│ ├── schemas.py # Pydantic request/response models +│ └── routers/ +│ ├── __init__.py +│ └── users.py +└── tests/ + ├── conftest.py + └── test_users.py +``` + +## Dependencies + +```bash +uv add fastapi 'sqlalchemy[asyncio]>=2.0' asyncpg 'pydantic[email]>=2' pydantic-settings 'uvicorn[standard]' orjson +uv add --dev httpx pytest alembic +``` + +`orjson` is mandatory: set `default_response_class=ORJSONResponse` on the FastAPI app. Pydantic-typed responses bypass it (Pydantic v2's `model_dump_json` is already Rust-backed); raw `dict` / `list` returns are accelerated. For SSE / NDJSON streams, call `orjson.dumps(...)` per chunk inside `StreamingResponse`. See `orjson-stack.md` for the decision tree, flag reference, and benchmarks. + +## Configuration (`config.py`) + +```python +from functools import lru_cache + +from pydantic import Field, PostgresDsn +from pydantic_settings import BaseSettings, SettingsConfigDict + + +class Settings(BaseSettings): + model_config = SettingsConfigDict(env_file=".env", env_prefix="MYAPI_") + + database_url: PostgresDsn + debug: bool = False + cors_origins: list[str] = Field(default_factory=list) + + +@lru_cache +def get_settings() -> Settings: + return Settings() # type: ignore[call-arg] # pydantic populates from env +``` + +Wait — that comment violates the no-excuse rule. Use proper field defaults instead. Real version: + +```python +class Settings(BaseSettings): + model_config = SettingsConfigDict(env_file=".env", env_prefix="MYAPI_") + database_url: PostgresDsn + debug: bool = False + cors_origins: list[str] = Field(default_factory=list) +``` + +Construct via `Settings(_env_file=".env")` if needed in tests; in production it reads from env. + +## Database (`db.py`) + +```python +from collections.abc import AsyncIterator +from typing import Annotated + +from fastapi import Depends +from sqlalchemy.ext.asyncio import ( + AsyncEngine, + AsyncSession, + async_sessionmaker, + create_async_engine, +) + +from myapi.config import get_settings + + +def make_engine() -> AsyncEngine: + settings = get_settings() + return create_async_engine( + str(settings.database_url), + echo=settings.debug, + pool_pre_ping=True, + ) + + +_engine = make_engine() +_SessionFactory = async_sessionmaker(_engine, expire_on_commit=False) + + +async def get_session() -> AsyncIterator[AsyncSession]: + async with _SessionFactory() as session: + yield session + + +SessionDep = Annotated[AsyncSession, Depends(get_session)] +``` + +`expire_on_commit=False` is essential for FastAPI - otherwise attribute access after commit triggers an implicit refresh and errors out under async. + +## Models (`models.py`) + +```python +from datetime import datetime, UTC +from sqlalchemy import DateTime, String, func +from sqlalchemy.orm import ( + DeclarativeBase, + Mapped, + MappedAsDataclass, + mapped_column, +) + + +class Base(MappedAsDataclass, DeclarativeBase): + pass + + +class User(Base): + __tablename__ = "users" + + id: Mapped[int] = mapped_column(primary_key=True, init=False) + email: Mapped[str] = mapped_column(String(255), unique=True, index=True) + name: Mapped[str] = mapped_column(String(100)) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + server_default=func.now(), + init=False, + ) +``` + +`MappedAsDataclass` makes `User(email=..., name=...)` work as a real dataclass constructor. `init=False` excludes the auto-generated columns (`id`, `created_at`) from `__init__`. + +## Schemas (`schemas.py`) + +```python +from datetime import datetime +from pydantic import BaseModel, ConfigDict, EmailStr + + +class UserCreate(BaseModel): + email: EmailStr + name: str + + +class UserRead(BaseModel): + model_config = ConfigDict(from_attributes=True) # SQLAlchemy → Pydantic + + id: int + email: EmailStr + name: str + created_at: datetime +``` + +Always have a separate `*Create` (input) and `*Read` (output) model. Never expose your ORM model as the API model. + +## Routers (`routers/users.py`) + +```python +from fastapi import APIRouter, HTTPException, status +from sqlalchemy import select + +from myapi.db import SessionDep +from myapi.models import User +from myapi.schemas import UserCreate, UserRead + +router = APIRouter(prefix="/users", tags=["users"]) + + +@router.post("", response_model=UserRead, status_code=status.HTTP_201_CREATED) +async def create_user(payload: UserCreate, session: SessionDep) -> User: + user = User(email=payload.email, name=payload.name) + session.add(user) + await session.commit() + await session.refresh(user) + return user + + +@router.get("/{user_id}", response_model=UserRead) +async def get_user(user_id: int, session: SessionDep) -> User: + result = await session.execute(select(User).where(User.id == user_id)) + user = result.scalar_one_or_none() + if user is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, "User not found") + return user + + +@router.get("", response_model=list[UserRead]) +async def list_users(session: SessionDep, limit: int = 100) -> list[User]: + result = await session.execute(select(User).limit(limit)) + return list(result.scalars().all()) +``` + +## Application (`main.py`) + +```python +from contextlib import asynccontextmanager +from collections.abc import AsyncIterator + +from fastapi import FastAPI + +from myapi.config import get_settings +from myapi.routers import users + + +@asynccontextmanager +async def lifespan(_: FastAPI) -> AsyncIterator[None]: + # Startup: warm up engine pool, run migrations check, etc. + yield + # Shutdown: close engine + from myapi.db import _engine + await _engine.dispose() + + +def create_app() -> FastAPI: + settings = get_settings() + app = FastAPI( + title="My API", + debug=settings.debug, + lifespan=lifespan, + ) + app.include_router(users.router) + return app + + +app = create_app() +``` + +Run with: + +```bash +uv run uvicorn myapi.main:app --host 0.0.0.0 --port 8000 --reload +``` + +## Migrations (Alembic + async) + +```bash +uv run alembic init -t async migrations +``` + +In `migrations/env.py` replace the `target_metadata` line: + +```python +from myapi.models import Base +target_metadata = Base.metadata +``` + +Set `sqlalchemy.url` in `alembic.ini` to your async URL or override via `env.py`: + +```python +from myapi.config import get_settings +config.set_main_option("sqlalchemy.url", str(get_settings().database_url)) +``` + +Generate and apply: + +```bash +uv run alembic revision --autogenerate -m "create users" +uv run alembic upgrade head +``` + +## Tests (`tests/test_users.py`) + +```python +import pytest +from httpx import ASGITransport, AsyncClient + +from myapi.main import app + + +@pytest.mark.anyio +async def test_create_and_get_user() -> None: + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + create_response = await client.post( + "/users", + json={"email": "alice@example.com", "name": "Alice"}, + ) + assert create_response.status_code == 201 + user_id = create_response.json()["id"] + + get_response = await client.get(f"/users/{user_id}") + assert get_response.status_code == 200 + assert get_response.json()["email"] == "alice@example.com" +``` + +For database-backed tests, run a Postgres container in CI (`testcontainers-python` or `docker-compose`) and apply migrations against a test schema. SQLite-as-test-db breaks once you use Postgres-specific types (`JSONB`, `tsvector`, arrays). + +## Common pitfalls + +| Pitfall | Fix | +|---|---| +| `MissingGreenlet` exception when accessing relationships after commit | `expire_on_commit=False` on the session factory | +| Connection pool exhausted under load | Set `pool_size`, `max_overflow` in `create_async_engine` | +| Pydantic v1 syntax (`from pydantic import ...; class X(BaseModel): class Config: orm_mode = True`) | v2 uses `model_config = ConfigDict(from_attributes=True)` | +| Returning ORM objects without `response_model` | FastAPI serialises with `from_attributes=True` automatically; declare `response_model` so OpenAPI is correct | +| `await session.execute(...)` returning Sequence | Wrap with `list(result.scalars().all())` to satisfy strict types | +| `func.now()` returning naive datetime | Use `DateTime(timezone=True)` and `created_at: Mapped[datetime]` with `UTC`-aware default | + +## Sources + +- FastAPI: +- SQLAlchemy 2.x async: +- SQLAlchemy MappedAsDataclass: +- asyncpg: +- Pydantic v2 migration: +- Alembic async: diff --git a/packages/omo-codex/plugin/skills/programming/references/python/httpx2-optimization.md b/packages/omo-codex/plugin/skills/programming/references/python/httpx2-optimization.md new file mode 100644 index 000000000..0f5efcbae --- /dev/null +++ b/packages/omo-codex/plugin/skills/programming/references/python/httpx2-optimization.md @@ -0,0 +1,360 @@ +# httpx2 — Production Defaults + +> **Source**: [pydantic/httpx2](https://github.com/pydantic/httpx2) — next-generation HTTP client for Python 3, continuation of HTTPX under Pydantic stewardship. +> +> **Rule**: Every network request MUST use `httpx2`. **ALL optimizations below are ON by default** — HTTP/2, brotli+zstd, tuned connection pool, fine-grained timeouts, transport retries, TCP_NODELAY. This is the baseline, not a stretch goal. A bare `httpx2.AsyncClient()` is a bug. + +--- + +## 1. Installation — all extras, always + +```toml +# pyproject.toml +dependencies = [ + "httpx2[http2,brotli,zstd]", +] +``` + +| Extra | What it enables | Why it's mandatory | +|-------|----------------|--------------------| +| `http2` | HTTP/2 multiplexing via `h2` | Single TCP connection handles concurrent requests; eliminates head-of-line blocking | +| `brotli` | Brotli content decoding (`br`) | ~20% smaller payloads than gzip for text/JSON | +| `zstd` | Zstandard content decoding | Faster decompression than brotli at similar ratios; stdlib in Python ≥ 3.14 | +| `socks` | SOCKS5 proxy support via `socksio` | Install only if you route through SOCKS proxies | + +All three core extras (`http2,brotli,zstd`) are non-negotiable. Omitting any is leaving performance on the table. + +--- + +## 2. The canonical defaults — ALL ON + +These are not "optimizations to consider". These are **the correct defaults** that every httpx2 client must use. + +```python +import socket +import httpx2 + +# ── These are the STANDARD values. Use them verbatim. ── + +LIMITS = httpx2.Limits( + max_connections=200, # library default 100 is too conservative + max_keepalive_connections=40, # library default 20 wastes reconnects + keepalive_expiry=30.0, # library default 5s kills warm connections too fast +) + +TIMEOUT = httpx2.Timeout( + connect=5.0, # TCP + TLS handshake budget + read=30.0, # time to receive a response chunk + write=10.0, # time to send a request chunk + pool=10.0, # time to acquire a connection from pool +) + +SOCKET_OPTIONS: list[tuple[int, int, int]] = [ + (socket.IPPROTO_TCP, socket.TCP_NODELAY, 1), # disable Nagle — no 40ms delay +] +``` + +### Why each knob is set this way + +| Setting | Library default | Our default | Why | +|---------|----------------|-------------|-----| +| `http2` | `False` | **`True`** | HTTP/2 multiplexing is strictly superior for any modern API | +| `max_connections` | `100` | `200` | Headroom for fan-out; prevents pool exhaustion under load | +| `max_keepalive_connections` | `20` | `40` | Keeps warm connections alive; fewer TLS handshakes | +| `keepalive_expiry` | `5.0s` | `30.0s` | 5s is too aggressive — kills connections between burst requests | +| `Timeout(5.0)` uniform | `5.0` all | Split | Uniform 5s is too tight for reads, too loose for connects | +| `read` timeout | `5.0` | `30.0` | Slow APIs and streaming need breathing room | +| `pool` timeout | `5.0` | `10.0` | Explicit — hitting this means `max_connections` needs raising | +| `TCP_NODELAY` | off | **on** | Eliminates Nagle's 40ms coalescing delay for small payloads | +| `retries` | `0` | `3` | Retries on `ConnectError`/`ConnectTimeout` only — safe and resilient | +| `follow_redirects` | `False` | **`True`** | Most APIs redirect; failing on 3xx is wrong default behavior | + +--- + +## 3. Factory functions — the ONE correct way to create clients + +Copy this into your project. This is the canonical pattern. + +```python +"""httpx2 client factory. Always use create_client() / create_async_client().""" + +from __future__ import annotations + +import socket +import typing + +import httpx2 + +_LIMITS = httpx2.Limits( + max_connections=200, + max_keepalive_connections=40, + keepalive_expiry=30.0, +) + +_TIMEOUT = httpx2.Timeout( + connect=5.0, + read=30.0, + write=10.0, + pool=10.0, +) + +_SOCKET_OPTIONS: list[tuple[int, int, int]] = [ + (socket.IPPROTO_TCP, socket.TCP_NODELAY, 1), +] + + +def create_async_client( + *, + base_url: str = "", + http2: bool = True, + retries: int = 3, + limits: httpx2.Limits = _LIMITS, + timeout: httpx2.Timeout = _TIMEOUT, + headers: dict[str, str] | None = None, + event_hooks: dict[str, list[typing.Callable[..., typing.Any]]] | None = None, + **kwargs: typing.Any, +) -> httpx2.AsyncClient: + transport = httpx2.AsyncHTTPTransport( + http2=http2, + retries=retries, + limits=limits, + socket_options=_SOCKET_OPTIONS, + ) + return httpx2.AsyncClient( + transport=transport, + timeout=timeout, + base_url=base_url, + headers=headers or {}, + event_hooks=event_hooks or {}, + follow_redirects=True, + **kwargs, + ) + + +def create_client( + *, + base_url: str = "", + http2: bool = True, + retries: int = 3, + limits: httpx2.Limits = _LIMITS, + timeout: httpx2.Timeout = _TIMEOUT, + headers: dict[str, str] | None = None, + event_hooks: dict[str, list[typing.Callable[..., typing.Any]]] | None = None, + **kwargs: typing.Any, +) -> httpx2.Client: + transport = httpx2.HTTPTransport( + http2=http2, + retries=retries, + limits=limits, + socket_options=_SOCKET_OPTIONS, + ) + return httpx2.Client( + transport=transport, + timeout=timeout, + base_url=base_url, + headers=headers or {}, + event_hooks=event_hooks or {}, + follow_redirects=True, + **kwargs, + ) +``` + +Usage: + +```python +# Async — the common case +async with create_async_client(base_url="https://api.example.com") as client: + r = await client.get("/users") + +# Sync +with create_client() as client: + r = client.get("https://api.example.com/health") +``` + +**If you are NOT using this factory pattern, you are doing it wrong.** A bare `httpx2.AsyncClient()` leaves HTTP/2 off, retries off, TCP_NODELAY off, keepalive too short, and timeouts too uniform. + +--- + +## 4. Special case overrides + +The factory defaults cover 95% of use cases. Override only when you have a specific reason: + +| Scenario | Override | +|----------|----------| +| LLM streaming endpoints | `timeout=httpx2.Timeout(connect=10.0, read=None, write=10.0, pool=10.0)` — no read timeout on streaming | +| Single-host API with low concurrency | `limits=httpx2.Limits(max_connections=50, max_keepalive_connections=20, keepalive_expiry=60.0)` | +| Ephemeral short-lived requests | `keepalive_expiry=5.0` — don't hold connections | +| Unix domain sockets | `httpx2.AsyncHTTPTransport(uds="/path/to/socket", ...)` | +| mTLS / client certs | Pass `verify=ssl_ctx` with `ctx.load_cert_chain(certfile=...)` | +| SOCKS proxy | `httpx2[socks]`, `proxy="socks5://..."` | + +--- + +## 5. Event hooks — always wire observability + +This is not optional. Every production client should log requests. + +```python +import time +import logging + +logger = logging.getLogger(__name__) + +async def log_request(request: httpx2.Request) -> None: + request.extensions["request_start"] = time.perf_counter() + +async def log_response(response: httpx2.Response) -> None: + start = response.request.extensions.get("request_start", 0) + elapsed = time.perf_counter() - start + logger.info( + "HTTP %s %s → %d (%.3fs, %s)", + response.request.method, + response.request.url, + response.status_code, + elapsed, + response.http_version, + ) + +# Sync versions for Client +def log_request_sync(request: httpx2.Request) -> None: + request.extensions["request_start"] = time.perf_counter() + +def log_response_sync(response: httpx2.Response) -> None: + start = response.request.extensions.get("request_start", 0) + elapsed = time.perf_counter() - start + logger.info( + "HTTP %s %s → %d (%.3fs, %s)", + response.request.method, + response.request.url, + response.status_code, + elapsed, + response.http_version, + ) +``` + +For auto `raise_for_status()`: + +```python +async def raise_on_error(response: httpx2.Response) -> None: + response.raise_for_status() +``` + +--- + +## 6. Verification script — confirm your setup is fully optimized + +Run this against your target endpoint to **verify** (not decide) that all optimizations are active: + +```python +"""Verify httpx2 is fully optimized against a target endpoint.""" + +from __future__ import annotations + +import socket +import time + +import anyio +import httpx2 + + +TARGET_URL = "https://api.example.com/health" +ITERATIONS = 30 + + +async def bench(label: str, client: httpx2.AsyncClient, url: str, n: int) -> float: + for _ in range(3): # warmup + await client.get(url) + start = time.perf_counter() + for _ in range(n): + r = await client.get(url) + assert r.status_code == 200 + elapsed = time.perf_counter() - start + avg_ms = (elapsed / n) * 1000 + print(f" {label}: {avg_ms:.1f}ms avg ({n} reqs in {elapsed:.2f}s)") + return avg_ms + + +async def main() -> None: + results: dict[str, float] = {} + + # BAD: bare defaults (this is what we're proving is worse) + async with httpx2.AsyncClient() as c: + results["BAD-bare-defaults"] = await bench("BAD-bare-defaults", c, TARGET_URL, ITERATIONS) + + # GOOD: full production defaults (this is what we always use) + limits = httpx2.Limits(max_connections=200, max_keepalive_connections=40, keepalive_expiry=30.0) + timeout = httpx2.Timeout(connect=5.0, read=30.0, write=10.0, pool=10.0) + transport = httpx2.AsyncHTTPTransport( + http2=True, retries=3, limits=limits, + socket_options=[(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)], + ) + async with httpx2.AsyncClient(transport=transport, timeout=timeout, follow_redirects=True) as c: + results["GOOD-full-production"] = await bench("GOOD-full-production", c, TARGET_URL, ITERATIONS) + + print("\n--- Proof ---") + baseline = results["BAD-bare-defaults"] + for label, avg in results.items(): + delta = ((avg - baseline) / baseline) * 100 + print(f" {label}: {avg:.1f}ms ({delta:+.1f}% vs bare)") + + +if __name__ == "__main__": + anyio.run(main) +``` + +--- + +## 7. Quick reference — all knobs + +### `httpx2.AsyncClient` / `httpx2.Client` + +| Parameter | Type | Library Default | **Our Default** | +|-----------|------|-----------------|-----------------| +| `http1` | `bool` | `True` | `True` | +| `http2` | `bool` | `False` | **`True`** | +| `verify` | `ssl.SSLContext \| str \| bool` | `True` | `True` | +| `cert` | `CertTypes \| None` | `None` | `None` | +| `proxy` | `str \| Proxy \| None` | `None` | `None` | +| `mounts` | `dict[str, Transport]` | `None` | `None` | +| `timeout` | `Timeout \| float \| None` | `Timeout(5.0)` | **Split: 5/30/10/10** | +| `limits` | `Limits` | `Limits(100, 20, 5.0)` | **`Limits(200, 40, 30.0)`** | +| `follow_redirects` | `bool` | `False` | **`True`** | +| `max_redirects` | `int` | `20` | `20` | +| `event_hooks` | `dict` | `{}` | **Wire logging** | +| `base_url` | `str` | `""` | Set for single-API clients | +| `trust_env` | `bool` | `True` | `True` | +| `default_encoding` | `str \| Callable` | `"utf-8"` | `"utf-8"` | + +### `httpx2.AsyncHTTPTransport` / `httpx2.HTTPTransport` + +| Parameter | Type | Library Default | **Our Default** | +|-----------|------|-----------------|-----------------| +| `http1` | `bool` | `True` | `True` | +| `http2` | `bool` | `False` | **`True`** | +| `retries` | `int` | `0` | **`3`** | +| `limits` | `Limits` | `Limits(100, 20, 5.0)` | **`Limits(200, 40, 30.0)`** | +| `uds` | `str \| None` | `None` | `None` | +| `local_address` | `str \| None` | `None` | `None` | +| `socket_options` | `Iterable[SOCKET_OPTION]` | `None` | **`[TCP_NODELAY]`** | +| `proxy` | `str \| Proxy \| None` | `None` | `None` | + +### `httpx2.Timeout` + +| Parameter | Library Default | **Our Default** | +|-----------|-----------------|-----------------| +| `connect` | `5.0` | `5.0` | +| `read` | `5.0` | **`30.0`** | +| `write` | `5.0` | **`10.0`** | +| `pool` | `5.0` | **`10.0`** | + +### `httpx2.Limits` + +| Parameter | Library Default | **Our Default** | +|-----------|-----------------|-----------------| +| `max_connections` | `100` | **`200`** | +| `max_keepalive_connections` | `20` | **`40`** | +| `keepalive_expiry` | `5.0` | **`30.0`** | + +### Async backend (httpcore2) + +httpcore2 uses `anyio` by default (works with both asyncio and trio). No extra config needed if you're already on the anyio stack. For trio, install `httpcore2[trio]`. diff --git a/packages/omo-codex/plugin/skills/programming/references/python/libraries.md b/packages/omo-codex/plugin/skills/programming/references/python/libraries.md new file mode 100644 index 000000000..f2e5e803e --- /dev/null +++ b/packages/omo-codex/plugin/skills/programming/references/python/libraries.md @@ -0,0 +1,307 @@ +# Library Defaults — Decision Tree + +For each domain, the canonical 2026 choice, why, and the canonical usage snippet. The skill enforces these unless the project's `pyproject.toml` explicitly says otherwise. + +## CLI — typer + +`typer` builds a CLI from type-annotated function signatures. argparse needs 5x the code; click ignores type annotations; fire is magic that breaks at scale. + +```python +import typer +from rich import print as rprint + +app = typer.Typer() + +@app.command() +def greet(name: str, count: int = 1, shout: bool = False) -> None: + """Print a greeting `count` times.""" + message = f"Hello, {name}!" if not shout else f"HELLO, {name.upper()}!" + for _ in range(count): + rprint(message) + +if __name__ == "__main__": + app() +``` + +For a single-function script, `typer.run(main)` skips the `Typer()` boilerplate. Subcommands use `@app.command()`. + +## Terminal output — rich + +`rich` produces tables, progress bars, syntax highlighting, traceback rendering. Use it for any structured output. Plain `print` is acceptable for non-interactive log lines (and even those are usually better via `rich.console.Console(stderr=True).log(...)`). + +```python +from rich.console import Console +from rich.table import Table + +console = Console() + +table = Table(title="Users") +table.add_column("ID", style="cyan") +table.add_column("Name", style="magenta") +table.add_row("1", "Alice") +console.print(table) + +# Rich tracebacks (call once at process start) +from rich.traceback import install +install(show_locals=True) +``` + +## HTTP client — [httpx2](https://github.com/pydantic/httpx2) + +Next-generation HTTP client under Pydantic stewardship. Sync and async in one library, HTTP/2 native, brotli + zstd content decoding, real type stubs. Replaces `requests` (sync only), `aiohttp` (async only), and the original `httpx`. + +**Install**: `httpx2[http2,brotli,zstd]` — always include all three extras, no exceptions. + +**A bare `httpx2.AsyncClient()` / `httpx2.Client()` is a bug.** Always use the factory pattern from `references/httpx2-optimization.md` with ALL optimizations enabled by default: + +```python +import socket +import httpx2 + +# ── Production defaults — ALL ON, always. ── +_LIMITS = httpx2.Limits(max_connections=200, max_keepalive_connections=40, keepalive_expiry=30.0) +_TIMEOUT = httpx2.Timeout(connect=5.0, read=30.0, write=10.0, pool=10.0) +_SOCKET_OPTS: list[tuple[int, int, int]] = [(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)] + +# Async (the common case) +transport = httpx2.AsyncHTTPTransport(http2=True, retries=3, limits=_LIMITS, socket_options=_SOCKET_OPTS) +async with httpx2.AsyncClient(transport=transport, timeout=_TIMEOUT, follow_redirects=True) as client: + response = await client.get("https://api.example.com/users") + response.raise_for_status() + users = response.json() + +# Sync +transport = httpx2.HTTPTransport(http2=True, retries=3, limits=_LIMITS, socket_options=_SOCKET_OPTS) +with httpx2.Client(transport=transport, timeout=_TIMEOUT, follow_redirects=True) as client: + response = client.get("https://api.example.com/users") + response.raise_for_status() + users = response.json() +``` + +See `references/httpx2-optimization.md` for the full factory functions (`create_client()` / `create_async_client()`), event hooks, and the rationale behind every setting. **Load that reference whenever you write ANY network code.** + +## JSON — stdlib `json` (default) or `orjson` (hot paths) + +Stdlib `json` is fine for cold paths and configs. **Reach for `orjson` when JSON is in the hot path** — cache layers, queue payloads, streaming responses, structured logs, FastAPI endpoints returning raw `dict` / `list`. + +```python +import orjson + +# orjson.dumps returns bytes, not str +raw: bytes = orjson.dumps( + payload, + option=orjson.OPT_NAIVE_UTC | orjson.OPT_UTC_Z | orjson.OPT_SERIALIZE_DATACLASS, +) +``` + +**Critical 2026 fact**: with Pydantic v2, `model.model_dump_json()` is backed by pydantic-core (Rust) and is faster than `orjson + default=` bridge for Pydantic-shaped responses. **Use `model_dump_json()` for Pydantic; orjson for everything else.** + +For FastAPI: `app = FastAPI(default_response_class=ORJSONResponse)`. Pydantic-typed responses bypass it (and that's correct — Pydantic's path is faster). Raw `dict`/`list` returns go through orjson. + +See `references/orjson-stack.md` for the full decision tree, option flag reference, FastAPI integration, Redis/queue/logging patterns, and the `model_dump_json()` vs orjson benchmark. + +## Validation — pydantic v2 + +Pydantic v2's core is in Rust (~10x faster than v1). It is the de-facto boundary validator. Use it for: + +- HTTP request/response models (FastAPI uses pydantic natively) +- Config files (env vars via `pydantic-settings`) +- Anything entering the program from outside + +```python +from pydantic import BaseModel, Field, EmailStr, field_validator + +class User(BaseModel): + id: int = Field(ge=1) + email: EmailStr + name: str = Field(min_length=1, max_length=100) + age: int | None = Field(default=None, ge=0, le=150) + + @field_validator("name") + @classmethod + def name_no_digits(cls, v: str) -> str: + if any(c.isdigit() for c in v): + raise ValueError("name cannot contain digits") + return v + +# Inside the program, use the validated instance with confidence +user = User.model_validate({"id": 1, "email": "a@b.com", "name": "Alice"}) +print(user.model_dump_json(indent=2)) +``` + +`@dataclass` is fine for purely internal records (no validation needed). For anything crossing a process boundary, use Pydantic. + +## Async — anyio + +Full reference: [async-anyio.md](async-anyio.md). The summary: + +```python +import anyio + +async def fetch(url: str) -> str: + await anyio.sleep(0.1) + return url + +async def main() -> None: + async with anyio.create_task_group() as tg: + for url in ["a", "b", "c"]: + tg.start_soon(fetch, url) + +anyio.run(main) +``` + +Never `import asyncio` directly. The third-party libraries you call are free to use asyncio internally. + +## Web framework — fastapi + +Type-hint-driven HTTP framework. Pydantic models become OpenAPI schemas automatically. + +```python +from fastapi import FastAPI +from pydantic import BaseModel + +app = FastAPI() + +class CreateUser(BaseModel): + name: str + email: str + +class User(BaseModel): + id: int + name: str + email: str + +@app.post("/users", response_model=User) +async def create_user(payload: CreateUser) -> User: + return User(id=1, **payload.model_dump()) +``` + +Full stack with database: [fastapi-stack.md](fastapi-stack.md). + +## ORM — sqlalchemy 2.x async + +SQLAlchemy 2.x finally has a real async API. Use the modern declarative `MappedAsDataclass` style with type annotations. + +```python +from sqlalchemy import String +from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker +from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, MappedAsDataclass + +class Base(MappedAsDataclass, DeclarativeBase): + pass + +class User(Base): + __tablename__ = "users" + id: Mapped[int] = mapped_column(primary_key=True, init=False) + name: Mapped[str] = mapped_column(String(100)) + email: Mapped[str] = mapped_column(String(255), unique=True) + +engine = create_async_engine("postgresql+asyncpg://localhost/myapp") +SessionFactory = async_sessionmaker(engine, expire_on_commit=False) +``` + +Full pattern with FastAPI integration: [fastapi-stack.md](fastapi-stack.md). + +## Database — postgres + asyncpg + +For new applications, default to Postgres. SQLite for tests is fine; SQLite for production is not. + +asyncpg is the fastest Python Postgres driver, native to SQLAlchemy 2.x async, native to FastAPI's lifespan model. URL: `postgresql+asyncpg://user:pass@host:5432/db`. + +For migrations, use Alembic with `[alembic.context]` configured to use the async engine. Single-step: + +```bash +uv add alembic +uv run alembic init -t async migrations +``` + +## TUI — textual + +Textual builds rich, mouse-aware, mobile-style TUIs on the rich rendering engine. See [textual-tui.md](textual-tui.md). + +## AI agents — pydantic-ai + +The agent framework from the Pydantic team. Type-strict, structured outputs are first-class, model-agnostic. See [pydantic-ai.md](pydantic-ai.md). + +## DataFrames — polars + numpy + +Polars is 10-50x faster than pandas, has a real type system, and supports lazy evaluation. Numpy stays in the toolbox for arrays. See [data-processing.md](data-processing.md). + +## OLAP / SQL — duckdb + +DuckDB is the SQL engine for analytical workloads. Query CSV/Parquet/JSON files directly without loading into memory; perform joins and aggregations 3-4x faster than Polars; zero-copy interchange with Polars via Arrow. See [data-processing.md](data-processing.md). + +## Tests — pytest + +Plain `unittest` is fine for stdlib; everything else uses pytest. Conventions: + +- File names `test_*.py`, function names `test_*`. +- Fixtures via `@pytest.fixture`. Async fixtures are anyio-aware (`@pytest.fixture` on an async function works under `pytest-anyio` which is bundled with anyio). +- Parametrise with `@pytest.mark.parametrize`. +- Mark async tests with `@pytest.mark.anyio` (provided by anyio's pytest plugin). + +```python +import pytest +import anyio + +@pytest.fixture +def sample_user() -> dict[str, str]: + return {"name": "Alice", "email": "a@b.com"} + +@pytest.mark.parametrize("count,expected", [(1, "Hello"), (2, "Hello, Hello")]) +def test_greet(count: int, expected: str) -> None: + result = ", ".join(["Hello"] * count) + assert result == expected + +@pytest.mark.anyio +async def test_async_fetch() -> None: + await anyio.sleep(0) + assert True +``` + +`pyproject.toml`: + +```toml +[tool.pytest.ini_options] +minversion = "8.0" +testpaths = ["tests"] +addopts = ["-ra", "--strict-config", "--strict-markers"] +``` + +## Settings / config — pydantic-settings + +Loads env vars and `.env` files into a Pydantic model. Replaces ad-hoc `os.environ.get(...)` everywhere. + +```python +from pydantic import Field +from pydantic_settings import BaseSettings, SettingsConfigDict + +class Settings(BaseSettings): + model_config = SettingsConfigDict(env_file=".env", env_prefix="MYAPP_") + + database_url: str + api_key: str = Field(min_length=1) + debug: bool = False + +settings = Settings() # loads at import time; raises if any required var is missing +``` + +## Logging — stdlib logging + rich handler + +Stdlib `logging` is fine; it gets a face-lift from `rich.logging.RichHandler`. + +```python +import logging +from rich.logging import RichHandler + +logging.basicConfig( + level=logging.INFO, + format="%(message)s", + datefmt="[%X]", + handlers=[RichHandler(rich_tracebacks=True, show_path=False)], +) +log = logging.getLogger(__name__) +log.info("ready") +``` + +For structured logging in production, swap to `structlog` (separate dep). Don't roll your own. diff --git a/packages/omo-codex/plugin/skills/programming/references/python/one-liners.md b/packages/omo-codex/plugin/skills/programming/references/python/one-liners.md new file mode 100644 index 000000000..de1ddbc2e --- /dev/null +++ b/packages/omo-codex/plugin/skills/programming/references/python/one-liners.md @@ -0,0 +1,268 @@ +# One-liner Scripts (PEP 723 + uv) + +Self-contained Python scripts with declared dependencies, run with no environment setup. The combination eliminates the historical reason to write small tools in Go or Bash. + +**Rule: EVERY `.py` script — even throwaway — MUST use PEP 723 inline metadata with the usage comment block.** No venv, no requirements.txt, no setup.py. The script IS the environment spec. + +## The two patterns + +### Pattern 1: inline `uv run` invocation + +```bash +uv run --with httpx2 --with rich python -c " +import httpx2 +from rich import print +print(httpx2.get('https://api.github.com').json()) +" +``` + +Use for terminal one-shots that you don't want to save. `--with PKG` may be repeated. + +### Pattern 2: PEP 723 script with shebang (THE CANONICAL PATTERN) + +A regular `.py` file with metadata in a comment block. uv reads the metadata, materialises a disposable venv (cached), and runs the script. + +```python +#!/usr/bin/env -S uv run --script +# /// script +# requires-python = ">=3.13" +# dependencies = [ +# "httpx2[http2,brotli,zstd]", +# "rich", +# ] +# /// + +# ─── How to run ─── +# 1. Install uv (if not installed): +# curl -LsSf https://astral.sh/uv/install.sh | sh +# 2. Run directly (no venv, no pip install needed): +# uv run my_script.py +# 3. Or make executable and run: +# chmod +x my_script.py && ./my_script.py +# ────────────────── + +from __future__ import annotations + +import httpx2 +from rich import print as rprint + + +def main() -> None: + with httpx2.Client(http2=True, follow_redirects=True) as client: + resp = client.get("https://api.github.com") + resp.raise_for_status() + rprint(resp.json()) + + +if __name__ == "__main__": + main() +``` + +### Mandatory elements + +Every PEP 723 script MUST include these, in order: + +1. **Shebang**: `#!/usr/bin/env -S uv run --script` +2. **PEP 723 metadata block**: `# /// script` ... `# ///` with `requires-python` and `dependencies` +3. **Usage comment block**: How to install uv + how to run the script. Copy the template above verbatim. +4. **`from __future__ import annotations`**: Always first import. +5. **`if __name__ == "__main__": main()`**: Entry point guard. + +### The usage comment block (NON-NEGOTIABLE) + +```python +# ─── How to run ─── +# 1. Install uv (if not installed): +# curl -LsSf https://astral.sh/uv/install.sh | sh +# 2. Run directly (no venv, no pip install needed): +# uv run .py [ARGS] +# 3. Or make executable and run: +# chmod +x .py && ./.py +# ────────────────── +``` + +Replace `` with the actual filename. Add argument descriptions if the script takes CLI args. This block goes immediately after the `# ///` closing line, before any imports. + +**Why mandatory**: Anyone who receives this script — colleague, CI, future you — must know how to run it without reading docs. The comment IS the docs. + +## Template generator + +Use `scripts/new-script.py` to scaffold a new PEP 723 script with all boilerplate pre-filled: + +```bash +# Generate to temp directory (default) +uv run scripts/new-script.py my_tool + +# Generate to specific path +uv run scripts/new-script.py my_tool --output ./scripts/my_tool.py + +# With extra dependencies +uv run scripts/new-script.py my_tool --deps "polars" "duckdb" "rich" +``` + +## Common dependency sets + +| Use case | Dependencies line | +|---|---| +| API client | `"httpx2[http2,brotli,zstd]"` | +| Data processing | `"polars"`, `"duckdb"` | +| CLI tool | `"typer"`, `"rich"` | +| Web scraping | `"httpx2[http2,brotli,zstd]"`, `"selectolax"` | +| File watcher | `"watchfiles"` | +| JSON pretty | `"rich"` | +| AI / LLM | `"pydantic-ai"`, `"httpx2[http2,brotli,zstd]"` | + +## Real-world examples + +### Fetch + print JSON + +```python +#!/usr/bin/env -S uv run --script +# /// script +# requires-python = ">=3.13" +# dependencies = [ +# "httpx2[http2,brotli,zstd]", +# "rich", +# ] +# /// + +# ─── How to run ─── +# 1. Install uv: curl -LsSf https://astral.sh/uv/install.sh | sh +# 2. Run: uv run fetch_json.py https://api.github.com/repos/pydantic/httpx2 +# ────────────────── + +from __future__ import annotations + +import sys + +import httpx2 +from rich import print as rprint + + +def main() -> None: + url = sys.argv[1] if len(sys.argv) > 1 else "https://api.github.com" + with httpx2.Client(http2=True, follow_redirects=True) as client: + resp = client.get(url) + resp.raise_for_status() + rprint(resp.json()) + + +if __name__ == "__main__": + main() +``` + +### CSV → Parquet conversion + +```python +#!/usr/bin/env -S uv run --script +# /// script +# requires-python = ">=3.13" +# dependencies = [ +# "polars", +# "typer", +# "rich", +# ] +# /// + +# ─── How to run ─── +# 1. Install uv: curl -LsSf https://astral.sh/uv/install.sh | sh +# 2. Run: uv run csv2parquet.py input.csv output.parquet +# ────────────────── + +from __future__ import annotations + +from pathlib import Path + +import polars as pl +import typer +from rich import print as rprint + + +def main(input_path: Path, output_path: Path | None = None) -> None: + """Convert CSV to Parquet.""" + out = output_path or input_path.with_suffix(".parquet") + df = pl.read_csv(input_path) + df.write_parquet(out) + rprint(f"[green]✓[/green] {input_path} → {out} ({len(df)} rows)") + + +if __name__ == "__main__": + typer.run(main) +``` + +### Quick benchmark + +```python +#!/usr/bin/env -S uv run --script +# /// script +# requires-python = ">=3.13" +# dependencies = [ +# "httpx2[http2,brotli,zstd]", +# "rich", +# "anyio", +# ] +# /// + +# ─── How to run ─── +# 1. Install uv: curl -LsSf https://astral.sh/uv/install.sh | sh +# 2. Run: uv run bench.py https://api.example.com/health 50 +# ────────────────── + +from __future__ import annotations + +import socket +import sys +import time + +import anyio +import httpx2 +from rich import print as rprint + + +async def main() -> None: + url = sys.argv[1] if len(sys.argv) > 1 else "https://api.github.com" + n = int(sys.argv[2]) if len(sys.argv) > 2 else 20 + + limits = httpx2.Limits(max_connections=200, max_keepalive_connections=40, keepalive_expiry=30.0) + timeout = httpx2.Timeout(connect=5.0, read=30.0, write=10.0, pool=10.0) + transport = httpx2.AsyncHTTPTransport( + http2=True, retries=3, limits=limits, + socket_options=[(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)], + ) + + async with httpx2.AsyncClient(transport=transport, timeout=timeout, follow_redirects=True) as client: + # warmup + for _ in range(3): + await client.get(url) + + start = time.perf_counter() + for _ in range(n): + r = await client.get(url) + assert r.status_code == 200 + elapsed = time.perf_counter() - start + + avg_ms = (elapsed / n) * 1000 + rprint(f"[bold]{url}[/bold]: {avg_ms:.1f}ms avg over {n} requests ({elapsed:.2f}s total, {r.http_version})") + + +if __name__ == "__main__": + anyio.run(main) +``` + +## Anti-patterns + +| ❌ Don't | ✅ Do | +|---|---| +| `pip install httpx2 && python script.py` | `uv run script.py` | +| `requirements.txt` alongside script | PEP 723 inline metadata | +| `python -m venv .venv && ...` | `uv run --script` handles it | +| Script without usage comment | Always include the "How to run" block | +| `import asyncio; asyncio.run(main())` | `import anyio; anyio.run(main)` | +| Bare `httpx2.AsyncClient()` | Full production defaults (see `references/httpx2-optimization.md`) | + +## Sources + +- PEP 723 - Inline script metadata: +- uv `run --script` docs: +- Original article: +- Simon Willison on one-shot Python tools: diff --git a/packages/omo-codex/plugin/skills/programming/references/python/orjson-stack.md b/packages/omo-codex/plugin/skills/programming/references/python/orjson-stack.md new file mode 100644 index 000000000..9c3ca0c65 --- /dev/null +++ b/packages/omo-codex/plugin/skills/programming/references/python/orjson-stack.md @@ -0,0 +1,378 @@ +# orjson — When to Use, How to Integrate + +`orjson` is the fastest JSON library on PyPI — written in Rust, 6–11× faster than stdlib `json` on serialization, 1.5–4× faster on deserialization. It also supports types the stdlib refuses to serialize: `datetime`, `date`, `UUID`, `numpy` arrays, `dataclass`, Pydantic models (via a small bridge). + +This document covers the production patterns. **Not every project needs orjson.** The decision tree is in §1. + +--- + +## 1. Decision tree — should you adopt orjson? + +``` +Are you serializing/deserializing JSON in a hot path? + ├─ NO → stdlib `json` is fine. Stop here. + └─ YES ↓ + +Is the project FastAPI? + ├─ YES ↓ + │ + │ Is your response body fully described by a Pydantic v2 model? + │ ├─ YES → Use FastAPI's default JSON response (uses Pydantic's + │ │ Rust-backed serializer; orjson saves nothing in this path). + │ │ Adopt orjson only for *non-Pydantic* responses below. + │ └─ NO → Use `ORJSONResponse` for endpoints that return dicts, + │ lists, or arbitrary structures. + │ + └─ NOT FastAPI ↓ + +Are you serializing Pydantic v2 models repeatedly? + ├─ YES → Use `model.model_dump_json()` directly — backed by pydantic-core + │ (Rust), within ~10% of orjson on the same payload, and respects + │ every Pydantic feature (computed fields, aliases, validators). + └─ NO ↓ + +Are you serializing dicts / lists / dataclasses / datetime / UUID? + ├─ YES → orjson is the right answer. + └─ NO → stdlib `json`. +``` + +**The crucial 2026 fact**: with Pydantic v2's `model_dump_json()`, **Pydantic-shaped responses no longer need orjson**. Adopt orjson where you are still going through `dict` / `list` / `dataclass`. + +--- + +## 2. Install + +```toml +# pyproject.toml +dependencies = [ + "orjson>=3.10", +] +``` + +orjson wheels are published for every major CPython version and platform (macOS, Linux glibc/musl, Windows, ARM64). No compilation step on install. + +--- + +## 3. Basic usage + +```python +import orjson + +# Serialization — returns bytes, not str +raw: bytes = orjson.dumps({"hello": "world", "ts": datetime.now(UTC)}) + +# Deserialization +data = orjson.loads(raw) +``` + +Two things to internalize: + +1. **`orjson.dumps` returns `bytes`**, not `str`. Stdlib `json.dumps` returns `str`. This is by design — most JSON destinations (sockets, files in binary mode, HTTP bodies) want bytes anyway, and skipping the encode/decode round trip is part of the speedup. +2. **No `indent` arg.** orjson supports `OPT_INDENT_2` (and only 2-space indent) via flags. If you need other indentation, use stdlib `json`. + +--- + +## 4. The option flags you actually use + +```python +import orjson + +orjson.dumps( + payload, + option=( + orjson.OPT_NAIVE_UTC # treat naive datetimes as UTC (recommended) + | orjson.OPT_UTC_Z # render UTC as "...Z" instead of "+00:00" + | orjson.OPT_SERIALIZE_NUMPY # serialize numpy arrays natively + | orjson.OPT_SERIALIZE_DATACLASS # serialize @dataclass instances + | orjson.OPT_NON_STR_KEYS # allow int / UUID / datetime dict keys + # | orjson.OPT_SORT_KEYS # only when you need deterministic output + # | orjson.OPT_INDENT_2 # only for human-readable output (slower) + ), +) +``` + +Each flag is opt-in for a reason — orjson defaults to spec-strict JSON. + +The flag combination above is a sensible "production default" for application code. The `OPT_NAIVE_UTC | OPT_UTC_Z` pair is especially important: it produces RFC 3339 timestamps that every parser on earth accepts. + +--- + +## 5. orjson + FastAPI + +### 5.1 The legacy pattern: `ORJSONResponse` + +```python +from fastapi import FastAPI +from fastapi.responses import ORJSONResponse + +app = FastAPI(default_response_class=ORJSONResponse) + +@app.get("/items") +async def get_items() -> dict[str, list[dict[str, int]]]: + return {"items": [{"id": i, "qty": i * 2} for i in range(1000)]} +``` + +`default_response_class=ORJSONResponse` swaps the global JSON encoder for orjson. **This affects only the response body serialization**, not request parsing — for request parsing, FastAPI still uses Pydantic. + +### 5.2 The 2026 reality — Pydantic v2 vs orjson + +With FastAPI 0.100+ on Pydantic v2: + +- If your response is annotated with a Pydantic model, FastAPI calls `model_dump_json()` directly. **orjson is bypassed** even with `default_response_class=ORJSONResponse`, because the Pydantic serializer is already Rust-backed. +- If your response is a raw `dict` / `list` / Python object, `ORJSONResponse` does kick in and saves real time. + +The benchmark in `tiangolo/fastapi#11728` (Apr 2024) showed `model_dump_json()` is ~10–15% faster than `ORJSONResponse + model_dump()` for Pydantic-shaped responses. The shape of the data matters; on mixed-shape APIs, keep `ORJSONResponse` as the default and trust Pydantic's path for typed responses. + +### 5.3 Recommended setup + +```python +from fastapi import FastAPI +from fastapi.responses import ORJSONResponse + +app = FastAPI( + default_response_class=ORJSONResponse, # benefits dict/list returns + # Pydantic-typed returns automatically use pydantic-core serialization +) +``` + +**Do NOT** wrap Pydantic models manually: + +```python +# BAD — defeats Pydantic's optimized path +@app.get("/users/{id}", response_class=ORJSONResponse) +async def get_user(id: int) -> ORJSONResponse: + user = await fetch_user(id) + return ORJSONResponse(content=user.model_dump()) # extra dict trip + +# GOOD — let FastAPI serialize the model +@app.get("/users/{id}") +async def get_user(id: int) -> User: + return await fetch_user(id) +``` + +### 5.4 Streaming responses + +`ORJSONResponse` does not stream — it buffers the whole response. For SSE, NDJSON, or chunked JSON, use `StreamingResponse` and call `orjson.dumps` per chunk: + +```python +from fastapi.responses import StreamingResponse +import orjson + +async def ndjson_stream(): + async for row in fetch_rows(): + yield orjson.dumps(row) + b"\n" + +@app.get("/export") +async def export(): + return StreamingResponse(ndjson_stream(), media_type="application/x-ndjson") +``` + +This is where orjson shines — per-chunk serialization in a tight loop, zero buffering. + +--- + +## 6. orjson + Pydantic v2 (no FastAPI) + +When you have a Pydantic model and want orjson's output for non-FastAPI contexts: + +```python +from pydantic import BaseModel +import orjson + +class User(BaseModel): + id: int + email: str + created: datetime + +user = User(id=1, email="a@b.com", created=datetime.now(UTC)) + +# Option A — Pydantic's built-in Rust serializer (USE THIS by default) +raw: bytes = user.model_dump_json().encode() +# 2026: ~1.2× faster than orjson on the same payload, supports +# every Pydantic feature (aliases, computed fields, json_schema_extra, etc.) + +# Option B — orjson bridge for cases Pydantic does not cover +raw: bytes = orjson.dumps( + user, + default=lambda obj: obj.model_dump() if isinstance(obj, BaseModel) else None, +) +# Useful when serializing nested non-Pydantic structures that contain +# BaseModels — e.g. a list of dicts that each may contain a BaseModel. +``` + +For routine "serialize one Pydantic model to JSON", `model_dump_json()` wins on speed AND feature parity. Reach for orjson only at the *container* level (a dict of mixed types). + +### Custom `default=` callback — the universal extension point + +```python +import orjson +from decimal import Decimal +from pydantic import BaseModel + +def _default(obj): + if isinstance(obj, BaseModel): + return obj.model_dump() + if isinstance(obj, Decimal): + return str(obj) + if isinstance(obj, set): + return list(obj) + raise TypeError(f"orjson: cannot serialize {type(obj).__name__}") + +orjson.dumps(payload, default=_default, option=orjson.OPT_NAIVE_UTC | orjson.OPT_UTC_Z) +``` + +The `default=` callback runs once per unrecognized type, then orjson caches the path. Performance impact on subsequent calls is negligible. + +--- + +## 7. Caching, queues, logging — the prime orjson use cases + +These are where orjson pays off most clearly because there is no Pydantic in the loop: + +### Redis cache + +```python +import orjson +import redis.asyncio as redis + +r = redis.from_url("redis://localhost") + +async def set_cache(key: str, value: dict) -> None: + await r.set(key, orjson.dumps(value), ex=3600) + +async def get_cache(key: str) -> dict | None: + raw = await r.get(key) + return orjson.loads(raw) if raw else None +``` + +`orjson` over stdlib `json` here saves ~5–10× on the serialize step for typical cache payloads. Multiply by request rate. + +### Task queue payloads (Celery, RQ, dramatiq) + +```python +# Celery custom serializer +from kombu.serialization import register +import orjson + +def _orjson_dumps(obj): + return orjson.dumps(obj, option=orjson.OPT_NAIVE_UTC | orjson.OPT_UTC_Z).decode() + +def _orjson_loads(s): + return orjson.loads(s) + +register("orjson", _orjson_dumps, _orjson_loads, + content_type="application/x-orjson", + content_encoding="utf-8") +``` + +Same speedup, applied to every task payload encode/decode. + +### Structured logging (structlog, custom slog) + +```python +import structlog +import orjson + +structlog.configure( + processors=[ + structlog.processors.TimeStamper(fmt="iso"), + structlog.processors.add_log_level, + structlog.processors.JSONRenderer(serializer=orjson.dumps), + ], +) +``` + +structlog's `JSONRenderer` accepts any callable; orjson is the obvious default. Logging hot paths benefit dramatically — every log line at info level becomes ~5× cheaper to render. + +--- + +## 8. Gotchas + +### `orjson.dumps` returns bytes, not str + +```python +# BAD — concatenating bytes and str +log.info("payload: " + orjson.dumps(data)) # TypeError + +# GOOD +log.info("payload: %s", orjson.dumps(data).decode()) +# or +log.info("payload: %s", orjson.dumps(data)) # let the formatter handle it +``` + +### No `cls=` argument for custom encoders + +orjson uses `default=` only. If you have a custom `JSONEncoder` subclass from stdlib `json`, port its `default()` method to a `default=` callable. + +### Subclasses of `dict` / `list` are NOT serialized as their parent + +```python +class StrictDict(dict): ... +d = StrictDict({"k": "v"}) + +import json +json.dumps(d) # OK — stdlib walks subclasses +orjson.dumps(d) # TypeError — orjson is strict by design +orjson.dumps(d, option=orjson.OPT_PASSTHROUGH_SUBCLASS) # then route via default= +``` + +Set `OPT_PASSTHROUGH_SUBCLASS` and handle the subclass in `default=`. The design discourages accidental subclass usage that breaks elsewhere. + +### `int` overflow + +orjson refuses to encode integers larger than 2⁵³ - 1 by default (the IEEE-754 double-precision safe-integer limit — what JavaScript can round-trip). For larger ints, opt in: + +```python +orjson.dumps(huge_int, option=orjson.OPT_STRICT_INTEGER) # error +orjson.dumps(huge_int) # default — int is encoded as JSON number +# JavaScript clients lose precision past 2^53; consider sending as string +``` + +This is more spec-strict than stdlib `json`, which silently emits ints of any size. + +### Timezone-naive datetimes + +By default, orjson treats naive `datetime` as the system local timezone — almost never what you want. **Always set `OPT_NAIVE_UTC`** to treat naive datetimes as UTC, or use timezone-aware datetimes (which is the better long-term habit). + +--- + +## 9. Benchmark — should I actually adopt this? + +The numbers below are 2024–2026 averages from `tiangolo/fastapi#11728` and orjson's own benchmark suite, on Python 3.13, modern x86_64: + +| Payload | stdlib `json` | `orjson` | `model_dump_json()` (Pydantic v2) | +|---|---|---|---| +| Small dict (100 fields) | 1.0× | **8×** | n/a | +| List of 10k dicts | 1.0× | **11×** | n/a | +| Pydantic model with 20 fields | 1.0× (after `model_dump()`) | 5× (with `default=` bridge) | **6×** | +| Datetime-heavy payload | 1.0× (after manual ISO conv) | **9×** | 6× | +| numpy array (1M floats) | impossible without manual conv | **20×** vs json+tolist | n/a | + +The takeaways: + +- For raw dict/list/datetime, **orjson is dramatically faster**. +- For Pydantic models, **`model_dump_json()` is already faster than orjson+bridge**. +- For numpy, orjson is the only sane choice. + +In production, the actual measured win on a FastAPI app with mixed payloads is typically 5–15% reduction in p99 latency. Worth the one-line `default_response_class=ORJSONResponse` switch. + +--- + +## 10. When NOT to adopt orjson + +- The codebase is small, JSON is not a bottleneck, and you have no measured perf concern. +- You depend on stdlib `json`'s `cls=` arg or its lax tolerance for non-spec input (NaN, Infinity, comments). +- You need pretty-printed JSON with custom indent — orjson only supports 2-space indent via the flag. +- You need pure-Python portability (e.g., MicroPython, no-wheel platforms) — orjson is a compiled Rust extension. + +If the choice is "add a dependency that does 5–10× the speed on serialization for free", the answer is almost always yes. The "almost" is in the bullets above. + +--- + +## Sources + +- orjson: https://github.com/ijl/orjson +- Pydantic v2 `model_dump_json`: https://docs.pydantic.dev/latest/concepts/serialization/#modelmodel_dump_json +- FastAPI `ORJSONResponse`: https://fastapi.tiangolo.com/advanced/custom-response/#use-orjsonresponse +- "FastAPI + orjson vs Pydantic v2" benchmark: https://github.com/fastapi/fastapi/discussions/11728 +- structlog JSON rendering: https://www.structlog.org/en/stable/api.html#structlog.processors.JSONRenderer diff --git a/packages/omo-codex/plugin/skills/programming/references/python/pydantic-ai.md b/packages/omo-codex/plugin/skills/programming/references/python/pydantic-ai.md new file mode 100644 index 000000000..3d8f41108 --- /dev/null +++ b/packages/omo-codex/plugin/skills/programming/references/python/pydantic-ai.md @@ -0,0 +1,285 @@ +# PydanticAI Reference (v1.x, 2026) + +> Canonical patterns for wiring PydanticAI agents. Target: production usage, late-2025 / 2026. +> Source: [ai.pydantic.dev](https://ai.pydantic.dev) and [pydantic/pydantic-ai@`cad9569`](https://github.com/pydantic/pydantic-ai/blob/cad956910079737ea0886b50cef15777208f92e6). + +--- + +## 1. Agent Constructor + +```python +from pydantic_ai import Agent + +agent = Agent( + 'openai:gpt-5.2', # model (str | Model | None) + output_type=MyOutputModel, # structured output type; default=str + instructions='You are a...', # static or callable instructions + system_prompt='Be concise.', # static system prompt(s) + deps_type=MyDeps, # dependency type for type-checking only + name='my-agent', # optional, inferred from var name if omitted + retries=1, # default retries for tools + output validation + output_retries=None, # override retries for output validation only + tools=[my_tool], # list of Tool objects or plain functions + defer_model_check=False, # set True to skip env-var check at init time + end_strategy='early', # 'early' | 'graceful' | 'exhaustive' +) +``` + +**Breaking change (v1.88.0)**: `result_type` was renamed to `output_type`. Use `output_type`. + +--- + +## 2. Model Strings + +Format: `provider:model-name`. The framework infers the provider from the prefix. + +| Provider prefix | Example | +|---|---| +| `openai:` | `'openai:gpt-5.2'`, `'openai:gpt-4o'` | +| `anthropic:` | `'anthropic:claude-sonnet-4-6'`, `'anthropic:claude-opus-4-1'` | +| `google-gla:` | `'google-gla:gemini-3-flash-preview'` | +| `google-vertex:` | `'google-vertex:gemini-3-pro-preview'` | +| `bedrock:` | `'bedrock:anthropic.claude-sonnet-4-6'` | +| `xai:` / `grok:` | `'xai:grok-3'`, `'grok:grok-3-fast'` | +| `deepseek:` | `'deepseek:deepseek-chat'` | +| `cohere:` | `'cohere:command-r-08-2024'` | +| `gateway/...` | `'gateway/openai:gpt-5.2'` (PydanticAI Gateway) | + +Model can also be omitted at construction and passed per-run: `agent.run(prompt, model='openai:gpt-5.2')`. + +--- + +## 3. Tools + +### Decorator syntax + +```python +from pydantic_ai import Agent, RunContext + +agent = Agent('openai:gpt-5.2', deps_type=str) + +@agent.tool # default: receives RunContext as first arg +async def greet(ctx: RunContext[str], name: str) -> str: + return f"Hello {ctx.deps}, {name}!" + +@agent.tool_plain # no context needed +async def roll_dice(sides: int) -> int: + import random + return random.randint(1, sides) +``` + +### `RunContext[Deps]` + +First parameter of `@agent.tool` functions. Carries: + +- `ctx.deps` — the dependency instance +- `ctx.model` — the model being used +- `ctx.usage` — token usage so far +- `ctx.messages` — conversation history +- `ctx.retry` / `ctx.max_retries` — current retry count +- `ctx.agent` — the running agent instance + +Use `@agent.tool_plain` when the tool does **not** need any of the above. + +--- + +## 4. Structured Output + +Pass a Pydantic `BaseModel` (or `bool`, `int`, `list[str]`, etc.) as `output_type`. The result is accessed via `.output`. + +```python +from pydantic import BaseModel +from pydantic_ai import Agent + +class City(BaseModel): + name: str + country: str + population_millions: float + +agent = Agent('openai:gpt-5.2', output_type=City) +result = agent.run_sync('Tell me about Tokyo') +print(result.output) # City(name='Tokyo', country='Japan', ...) +print(result.output.name) # 'Tokyo' +``` + +**Note**: `result.data` was renamed; the canonical accessor is `result.output`. + +--- + +## 5. Async vs Sync + +| Method | Mode | Returns | +|---|---|---| +| `await agent.run(prompt, ...)` | async | `AgentRunResult[OutputDataT]` | +| `agent.run_sync(prompt, ...)` | sync | `AgentRunResult[OutputDataT]` | +| `async with agent.run_stream(prompt, ...) as response:` | async streaming | `StreamedRunResult` | + +```python +# Sync +result = agent.run_sync('What is the capital of Italy?') +print(result.output) + +# Async +result = await agent.run('What is the capital of France?') +print(result.output) + +# Streaming +async with agent.run_stream('What is the capital of the UK?') as response: + async for text in response.stream_text(): + print(text, end='') + # After streaming finishes: + print(response.output) +``` + +`run_sync()` is a convenience wrapper over `loop.run_until_complete(self.run(...))`. Do not use it inside an active async context. + +--- + +## 6. Dependencies + +Use a `@dataclass` container, pass the **type** to `deps_type`, and pass an **instance** to `deps` at run time. + +```python +from dataclasses import dataclass +import httpx +from pydantic_ai import Agent, RunContext + +@dataclass +class Deps: + api_key: str + http_client: httpx.AsyncClient + +agent = Agent( + 'openai:gpt-5.2', + deps_type=Deps, +) + +@agent.tool +async def fetch_data(ctx: RunContext[Deps], endpoint: str) -> str: + r = await ctx.deps.http_client.get( + endpoint, + headers={'Authorization': f'Bearer {ctx.deps.api_key}'}, + ) + r.raise_for_status() + return r.text + +async def main(): + async with httpx.AsyncClient() as client: + deps = Deps(api_key='sk-...', http_client=client) + result = await agent.run('Get /users', deps=deps) + print(result.output) +``` + +--- + +## 7. Error Types & Retrying from a Tool + +```python +from pydantic_ai import Agent, ModelRetry, UnexpectedModelBehavior, capture_run_messages + +agent = Agent('openai:gpt-5.2', retries=3) + +@agent.tool_plain +def calc_volume(size: int) -> int: + if size == 42: + return size ** 3 + raise ModelRetry('Please try again with size 42.') + +with capture_run_messages() as messages: + try: + result = agent.run_sync('Get the volume of a box with size 6.') + except UnexpectedModelBehavior as e: + print('Error:', e) # "Tool 'calc_volume' exceeded max retries count of 3" + print('Cause:', e.__cause__) # ModelRetry('Please try again...') + print('Messages:', messages) +``` + +- **`ModelRetry`** — raise from a tool, output validator, or capability hook to ask the model to retry. +- **`UnexpectedModelBehavior`** — raised when the retry limit is exceeded or the model API returns an unrecoverable error. +- **`capture_run_messages()`** — context manager that records all messages exchanged during a run for debugging. + +--- + +## 8. Logfire Integration + +One-line setup if the `logfire` extra is installed (included in the default `pydantic-ai` package): + +```python +import logfire + +logfire.configure() # reads token from .logfire directory +logfire.instrument_pydantic_ai() # auto-traces all agent runs +``` + +Alternatively, set `instrument=True` on the agent: + +```python +agent = Agent('openai:gpt-5.2', instrument=True) +``` + +--- + +## 9. Minimal Complete Snippets + +### (a) Basic agent with structured output + +```python +from pydantic import BaseModel +from pydantic_ai import Agent + +class City(BaseModel): + name: str + country: str + +agent = Agent('openai:gpt-5.2', output_type=City) +result = agent.run_sync('Tell me about Paris') +print(result.output) # City(name='Paris', country='France') +``` + +### (b) Agent with tools and dependencies + +```python +from dataclasses import dataclass +from pydantic_ai import Agent, RunContext + +@dataclass +class Deps: + api_key: str + +agent = Agent('openai:gpt-5.2', deps_type=Deps) + +@agent.tool +async def get_secret(ctx: RunContext[Deps], code: str) -> str: + if code == '1234': + return f'secret-for-{ctx.deps.api_key}' + return 'wrong code' + +result = agent.run_sync('My code is 1234', deps=Deps(api_key='sk-abc')) +print(result.output) +``` + +### (c) Async streaming + +```python +import anyio +from pydantic_ai import Agent + +agent = Agent('openai:gpt-5.2') + +async def main() -> None: + async with agent.run_stream('Write a haiku about Python') as response: + async for text in response.stream_text(): + print(text, end='') + print('\n---') + print('Final:', response.output) + +anyio.run(main) +``` + +--- + +## Version Notes + +- **V1** reached API stability in September 2025. Breaking changes are reserved for V2 (earliest April 2026). +- **v1.88.0** renamed `result_type` → `output_type` and `result_tool_name` / `result_tool_description` were removed. Use `output_type`. +- The canonical accessor for run results is `result.output` (not `result.data`). diff --git a/packages/omo-codex/plugin/skills/programming/references/python/pyproject-strict.md b/packages/omo-codex/plugin/skills/programming/references/python/pyproject-strict.md new file mode 100644 index 000000000..7bbf96601 --- /dev/null +++ b/packages/omo-codex/plugin/skills/programming/references/python/pyproject-strict.md @@ -0,0 +1,232 @@ +# Strict pyproject.toml (basedpyright + ruff + uv) + +The canonical "super strict but sane" config for modern Python projects. Copy-paste, then add your own dependencies. + +## Bootstrap + +```bash +# Application +uv init --app myproject +cd myproject + +# Library (publishable to PyPI) +uv init --lib mylibrary +cd mylibrary + +# Add dev tools +uv add --dev basedpyright ruff pytest +``` + +`uv init` creates `pyproject.toml`, `.python-version`, and `src/` layout. Replace its `pyproject.toml` `[tool.*]` sections with the block below. + +## The full pyproject.toml + +```toml +[project] +name = "myproject" +version = "0.1.0" +description = "..." +readme = "README.md" +requires-python = ">=3.13" +dependencies = [] + +[dependency-groups] +dev = [ + "basedpyright>=1.21", + "ruff>=0.8", + "pytest>=8", + "pytest-cov>=5", +] + +# ───────────────────────────────────────────────────────────────── +# basedpyright - typeCheckingMode = "all" sets every report flag to error +# Source: https://docs.basedpyright.com/latest/configuration/config-files/ +# ───────────────────────────────────────────────────────────────── +[tool.basedpyright] +typeCheckingMode = "all" +pythonVersion = "3.13" +pythonPlatform = "All" # default in basedpyright; explicit for clarity +include = ["src", "tests"] +exclude = ["**/__pycache__", "**/.venv", "**/build", "**/dist"] + +# Strict enforcement extras (most are already "error" under "all" mode, +# but listing them explicitly documents the intent) +reportUnusedCallResult = "warning" # flag ignored return values +reportUnnecessaryTypeIgnoreComment = "error" # stale type: ignore comments must die +reportUnusedVariable = "error" # unused variables are errors +reportMissingParameterType = "error" # every parameter must have a type +reportMissingReturnType = "error" # every function must declare its return type +reportPrivateUsage = "error" # respect _private convention + +# Optional: gradual adoption baseline +# baselineFile = "./.basedpyright/baseline.json" + +# ───────────────────────────────────────────────────────────────── +# ruff - select = ["ALL"] enables every rule, then we ignore the +# small set that conflicts with the formatter or is not useful. +# Source: https://docs.astral.sh/ruff/linter/#rule-selection +# ───────────────────────────────────────────────────────────────── +[tool.ruff] +target-version = "py313" +line-length = 88 # ruff/black default; 100 or 120 also fine +src = ["src", "tests"] + +[tool.ruff.lint] +select = ["ALL"] +ignore = [ + # Formatter conflicts (ruff itself tells you to ignore these) + "COM812", # missing trailing comma + "ISC001", # implicit string concat + # Docstyle conflicts (pick D211 over D203, D212 over D213) + "D203", + "D213", + # Project-specific noise + "CPY001", # missing copyright notice + "FBT001", # boolean positional arg in def + "FBT002", # boolean positional default in def + "TD002", # missing TODO author + "TD003", # missing TODO link + "FIX002", # line contains TODO (TODOs are allowed) +] +fixable = ["ALL"] +unfixable = [] + +[tool.ruff.lint.per-file-ignores] +"tests/**/*.py" = [ + "S101", # `assert` is the entire point of pytest + "ARG", # unused args (fixtures appear unused) + "PLR2004", # magic numbers in test data + "SLF001", # tests need access to private members + "D", # docstrings not required in tests +] +"scripts/**/*.py" = [ + "T201", # `print` allowed in scripts + "INP001", # implicit namespace package +] + +[tool.ruff.lint.pydocstyle] +convention = "google" # or "numpy" / "pep257" + +[tool.ruff.lint.flake8-bugbear] +# typer / fastapi rely on call-as-default for parameter metadata. +# Without this, ruff B008 ("function call in default") fires on every typer/fastapi route. +extend-immutable-calls = [ + "typer.Argument", + "typer.Option", + "fastapi.Depends", + "fastapi.Query", + "fastapi.Path", + "fastapi.Body", + "fastapi.Header", + "fastapi.Cookie", + "fastapi.File", + "fastapi.Form", +] + +[tool.ruff.format] +quote-style = "double" +indent-style = "space" +docstring-code-format = true +docstring-code-line-length = "dynamic" + +# ───────────────────────────────────────────────────────────────── +# pytest +# ───────────────────────────────────────────────────────────────── +[tool.pytest.ini_options] +minversion = "8.0" +testpaths = ["tests"] +addopts = [ + "-ra", + "--strict-config", + "--strict-markers", +] +filterwarnings = ["error"] + +# ───────────────────────────────────────────────────────────────── +# coverage +# ───────────────────────────────────────────────────────────────── +[tool.coverage.run] +source = ["src"] +branch = true + +[tool.coverage.report] +exclude_lines = [ + "pragma: no cover", + "if TYPE_CHECKING:", + "if typing.TYPE_CHECKING:", + "raise NotImplementedError", + "@(abc\\.)?abstractmethod", +] +``` + +## Why these settings + +### basedpyright `typeCheckingMode = "all"` + +basedpyright's modes, strictest first: + +| Mode | Behavior | +|---|---| +| `"all"` | Every diagnostic at `error` | +| `"recommended"` | Same rules; less severe ones at `warning`; `failOnWarnings = true` makes CI still fail | +| `"strict"` | pyright's strict mode | +| `"standard"` | Default | +| `"basic"` / `"off"` | Loose / disabled | + +`"all"` enables basedpyright-exclusive rules pyright lacks: `reportImplicitOverride`, `reportImplicitStringConcatenation`, `reportIncompatibleUnannotatedOverride`, `reportUnannotatedClassAttribute`. No need to opt-in to additional flags. + +`pythonPlatform = "All"` is basedpyright's default (better than pyright's host-OS default) - it errors on platform-specific imports that fail on other OSes. + +### ruff `select = ["ALL"]` + +The official docs say *"Use ALL with discretion. Enabling ALL will implicitly enable new rules whenever you upgrade."* For a strict skill that is the intended behavior - every new ruff rule should be considered an error until you justify ignoring it. + +The minimal ignore set: + +| Rule | Reason | +|---|---| +| `COM812`, `ISC001` | Conflict with `ruff format` (ruff itself documents this) | +| `D203` vs `D211`, `D213` vs `D212` | Mutually-exclusive docstring conventions; pick the modern one | +| `CPY001` | Most projects don't need a copyright header on every file | +| `FBT001`, `FBT002` | Boolean flags are ergonomic for CLI/typer; ban makes typer awkward | +| `TD002`, `TD003`, `FIX002` | TODOs without a JIRA link are fine in solo / internal code | + +`ANN101` and `ANN102` were **removed in ruff 0.8.0** (Nov 2024). Do NOT include them in `ignore` - ruff errors on unknown rule codes. + +`per-file-ignores` for `tests/**` is the standard pattern from real-world repos like `community-of-python/auto-typing-final` and `Preston-Landers/concurrent-log-handler`. + +## CI gate + +```bash +# In CI, fail on any violation: +uv run basedpyright +uv run ruff check +uv run ruff format --check +uv run pytest +``` + +A single `make ci` target combining the four works fine. + +## Enforcement summary + +The config above, combined with `scripts/check-no-excuse-rules.py`, enforces: + +| What | How | +|---|---| +| Exhaustive match | basedpyright `all` mode + `assert_never` | +| No `Any` | basedpyright `all` mode + script `cast-any` rule | +| Ignored return values | `reportUnusedCallResult = "warning"` | +| Immutable default | Script `mutable-dataclass` + `missing-slots` rules | +| No null surprise | basedpyright strict `None` analysis | +| Constants are const | basedpyright catches `Final` reassignment | +| Unused variables | `reportUnusedVariable = "error"` | + +## Sources + +- basedpyright modes: +- basedpyright `"all"` vs `"recommended"`: +- basedpyright better defaults: +- ruff rule selection: +- ruff ANN101/ANN102 removed: +- Real-world ALL config: +- PEP 735 dependency-groups: diff --git a/packages/omo-codex/plugin/skills/programming/references/python/textual-tui.md b/packages/omo-codex/plugin/skills/programming/references/python/textual-tui.md new file mode 100644 index 000000000..27f027de2 --- /dev/null +++ b/packages/omo-codex/plugin/skills/programming/references/python/textual-tui.md @@ -0,0 +1,201 @@ +# Textual TUI + +Textual builds rich, mouse-aware, scrollable, mobile-style TUIs on top of `rich`. Replaces curses, urwid, blessed. + +## Install + +```bash +uv add textual +uv add --dev textual-dev # textual console + run --dev for hot reload +``` + +## Minimal app + +```python +from textual.app import App, ComposeResult +from textual.widgets import Header, Footer, Button, Label +from textual.containers import Vertical + + +class CounterApp(App[None]): + """A trivial counter app.""" + + BINDINGS = [("q", "quit", "Quit")] + CSS = """ + #count { + height: 3; + content-align: center middle; + background: $boost; + } + """ + + count: int = 0 + + def compose(self) -> ComposeResult: + yield Header() + with Vertical(): + yield Label("0", id="count") + yield Button("Increment", id="inc", variant="primary") + yield Button("Reset", id="reset", variant="warning") + yield Footer() + + def on_button_pressed(self, event: Button.Pressed) -> None: + if event.button.id == "inc": + self.count += 1 + elif event.button.id == "reset": + self.count = 0 + self.query_one("#count", Label).update(str(self.count)) + + +if __name__ == "__main__": + CounterApp().run() +``` + +Run: + +```bash +uv run python counter.py +``` + +For hot reload during development: + +```bash +uv run textual run --dev counter.py +``` + +## Reactive attributes + +Textual's `reactive()` descriptor turns a class attribute into something that watches assignments and re-renders automatically. Replaces the manual `query_one` + `update` dance. + +```python +from textual.app import App, ComposeResult +from textual.reactive import reactive +from textual.widgets import Label + + +class CountWidget(Label): + count: reactive[int] = reactive(0) + + def render(self) -> str: + return f"Count: {self.count}" + + +class CounterApp(App[None]): + def compose(self) -> ComposeResult: + yield CountWidget() + + def on_key(self, event) -> None: + if event.key == "space": + self.query_one(CountWidget).count += 1 +``` + +`reactive()` triggers `render()` (or `watch_` and `validate_` callbacks if defined). Use `recompose=True` if you need to call `compose()` again on change. + +## Async work — workers + +NEVER block the event loop. For network/disk/CPU work, use `@work` (creates a worker) or `run_worker`. + +```python +import httpx +from textual.app import App, ComposeResult +from textual.widgets import Input, Static +from textual.work import work + + +class FetchApp(App[None]): + def compose(self) -> ComposeResult: + yield Input(placeholder="URL", id="url") + yield Static(id="result") + + @work(exclusive=True) + async def fetch(self, url: str) -> None: + async with httpx.AsyncClient(timeout=10.0) as client: + response = await client.get(url) + self.query_one("#result", Static).update(f"{response.status_code} - {len(response.text)} bytes") + + def on_input_submitted(self, event: Input.Submitted) -> None: + self.fetch(event.value) +``` + +`exclusive=True` cancels the previous worker if the user submits a new URL before the first finishes. Workers integrate with Textual's lifecycle - they're cancelled when the app exits. + +`@work` is asyncio-flavoured under the hood. That is fine - it does not violate the no-asyncio rule because you are calling Textual's API, not importing asyncio yourself. Inside the worker body, use `httpx.AsyncClient` and other anyio-friendly libraries. + +## Action handlers + +Bind keys to method calls via `BINDINGS` and `action_*` methods. + +```python +class App(App): + BINDINGS = [ + ("ctrl+s", "save", "Save"), + ("ctrl+r", "reload", "Reload"), + ] + + def action_save(self) -> None: + # Called on ctrl+s + ... + + def action_reload(self) -> None: + ... +``` + +Bindings can also include the `priority=True` flag to fire before children get a chance. + +## CSS + +Textual's CSS supports selectors, variables (`$primary`, `$boost`), animations. Inline via `CSS = "..."` or external via `CSS_PATH = "app.tcss"`. + +```css +Screen { + background: $surface; + color: $text; + layout: vertical; +} + +#sidebar { + width: 30; + background: $boost; +} + +Button.danger { + background: $error; +} +``` + +Reload with `r` in dev mode (`textual run --dev`). + +## Testing + +```python +import pytest +from myapp import CounterApp + + +@pytest.mark.anyio +async def test_counter_increments() -> None: + app = CounterApp() + async with app.run_test() as pilot: + await pilot.click("#inc") + await pilot.click("#inc") + assert app.count == 2 +``` + +`pilot.click(selector)`, `pilot.press("q")`, `pilot.pause()` for waiting on the next frame. + +## When NOT to use Textual + +| Need | Use | +|---|---| +| One-off CLI with structured output | typer + rich | +| Progress bar in a script | rich.progress | +| Tabular display of query results | rich.table | +| Full-screen app with state, input, mouse | Textual | + +A pretty CLI is not a TUI. Reach for Textual when the user expects to navigate a UI, not when you want colours. + +## Sources + +- Textual docs: +- Textual tutorial: +- API reference: diff --git a/packages/omo-codex/plugin/skills/programming/references/python/type-patterns.md b/packages/omo-codex/plugin/skills/programming/references/python/type-patterns.md new file mode 100644 index 000000000..2811813d0 --- /dev/null +++ b/packages/omo-codex/plugin/skills/programming/references/python/type-patterns.md @@ -0,0 +1,176 @@ +# Type Patterns + +How to use Python's type system to catch bugs at check time, not runtime. + +--- + +## NewType — distinct primitives + +Same runtime type, different meaning. The type checker prevents mixing. + +```python +from typing import NewType + +UserId = NewType("UserId", int) +MovieId = NewType("MovieId", int) +Email = NewType("Email", str) +Seconds = NewType("Seconds", float) +Milliseconds = NewType("Milliseconds", float) + +def get_user(user_id: UserId) -> User: ... +def get_movie(movie_id: MovieId) -> Movie: ... +def sleep(duration: Seconds) -> None: ... + +uid = UserId(42) +mid = MovieId(42) + +get_user(uid) # OK +get_user(mid) # type error: MovieId is not UserId +get_user(42) # type error: int is not UserId +sleep(Milliseconds(100.0)) # type error +``` + +**Use when**: IDs, indices, keys, units of measurement — any pair where swapping is a bug. +**Skip when**: ephemeral local math where branding adds noise with zero safety gain. + +--- + +## Final — constants are const + +Module-level constants declare their intent. Reassignment is a type error. + +```python +from typing import Final + +MAX_RETRIES: Final = 3 +API_BASE_URL: Final = "https://api.example.com" +DEFAULT_TIMEOUT: Final = 30.0 + +MAX_RETRIES = 5 # type error: cannot assign to Final +``` + +If it changes at runtime, it's not a constant — make it a function parameter or config field. + +--- + +## TypeAlias — name complex types + +If a union or generic appears more than once, give it a name. + +```python +# Python 3.12+ +type JsonValue = str | int | float | bool | None | list["JsonValue"] | dict[str, "JsonValue"] +type Headers = dict[str, str] +type Middleware = Callable[[Request], Awaitable[Response]] + +# Pre-3.12 +from typing import TypeAlias + +JsonValue: TypeAlias = str | int | float | bool | None | list["JsonValue"] | dict[str, "JsonValue"] +``` + +--- + +## StrEnum / IntEnum — closed sets + +Any fixed set of known values. No string literals scattered through code. + +```python +from enum import StrEnum, IntEnum, unique + +@unique +class Role(StrEnum): + ADMIN = "admin" + USER = "user" + GUEST = "guest" + +@unique +class HttpStatus(IntEnum): + OK = 200 + NOT_FOUND = 404 + INTERNAL_ERROR = 500 + +# BAD +def check_role(role: str) -> bool: ... + +# GOOD +def check_role(role: Role) -> bool: ... +``` + +`StrEnum` when values serialize as strings (API, DB). `IntEnum` for numeric codes. Plain `Enum` for pure labels. + +--- + +## Type narrowing — let the checker follow your logic + +`isinstance`, `is None`, and `match` narrow types automatically. Use them instead of `cast`. + +```python +def process(value: str | int | None) -> str: + if value is None: + return "nothing" + # checker knows: str | int + + if isinstance(value, str): + return value.upper() + # checker knows: int + + return str(value * 2) +``` + +### TypeGuard for custom narrowing + +```python +from typing import TypeGuard + +def is_valid_email(value: str) -> TypeGuard[Email]: + return "@" in value and "." in value.split("@")[1] + +def send(addr: str) -> None: + if not is_valid_email(addr): + raise ValueError(addr) + # checker knows: addr is Email + deliver(addr) +``` + +### TypeIs (Python 3.13+) — the strict version + +`TypeIs` is stricter than `TypeGuard` — it narrows in both `if` and `else` branches. + +```python +from typing import TypeIs + +def is_str(value: str | int) -> TypeIs[str]: + return isinstance(value, str) + +def handle(v: str | int) -> None: + if is_str(v): + print(v.upper()) # checker knows: str + else: + print(v + 1) # checker knows: int +``` + +--- + +## Union syntax + +Always `X | Y`. Never `Union[X, Y]` or `Optional[X]`. + +```python +# BAD +from typing import Union, Optional +def f(x: Optional[int]) -> Union[str, int]: ... + +# GOOD +def f(x: int | None) -> str | int: ... +``` + +--- + +## Sources + +- Python docs: [typing — NewType](https://docs.python.org/3/library/typing.html#newtype) +- Python docs: [typing — Final](https://docs.python.org/3/library/typing.html#typing.Final) +- Python docs: [typing — TypeGuard](https://docs.python.org/3/library/typing.html#typing.TypeGuard) +- PEP 604: [Union syntax X | Y](https://peps.python.org/pep-0604/) +- PEP 742: [TypeIs](https://peps.python.org/pep-0742/)