Python 3.10+ is required. Use modern syntax:
X | Yunion types instead ofOptional[X]/Union[X, Y]where possiblematchstatements for structural pattern matchinglist[X],dict[K, V]instead ofList[X],Dict[K, V](no import needed in 3.10+)
- All data models use Pydantic v2 (
from pydantic import BaseModel). - Use
Field(default_factory=list)for mutable defaults, not[]. - Serialize with
model.model_dump(mode="json")before passing tojson.dumps— this convertsdatetimeand other types to JSON-compatible forms. - Never pass raw dicts across service boundaries; always use typed models.
- Invoke with
asyncio.create_subprocess_exec— nevershell=True. - Always capture both
stdoutandstderr(asyncio.subprocess.PIPE). - Decode output with
stdout.decode(errors="replace")to handle encoding issues. - Parse output defensively — malformed or empty lines must never crash the process.
- Return empty/partial models on failure, log the error to
sys.stderr.
- Service functions that invoke subprocesses must be
async. - Use
asyncio.gatherto run independent operations in parallel. - Use
asyncio.Semaphoreto cap concurrency when launching many subprocesses (default limit: 50).
- All public functions and methods must have full type annotations (parameters + return type).
- Use
Optional[X]for fields that may beNone(orX | Nonewithfrom __future__ import annotations).
- Services must not raise exceptions that propagate to the CLI layer.
- On failure: log to
sys.stderr, return an empty or partial model. - The CLI layer is the only place that exits with a non-zero status code.
- Standard library imports first, then third-party, then local (
lib.*). - Local imports use absolute paths from the project root (e.g.
from lib.common.entities import Domain).