Typed pydantic-settings config, structlog JSON logging with request/trace id propagation and redaction of phone numbers, tokens and face keys, the unified response envelope and error-code table from 架构文档 §9.2, async SQLAlchemy session management, ORM models for the eight foundation tables, and the OpenTelemetry/Prometheus wiring. packages/domain stays pure logic -- snowflake ids, the quota three-phase protocol and the tk_task state machine -- so it can be tested without a database and is held to mypy strict.
27 lines
868 B
Python
Executable File
27 lines
868 B
Python
Executable File
"""Pure-domain exceptions.
|
|
|
|
This layer must not import :mod:`packages.shared` (enforced by import-linter),
|
|
so it defines its own error types. Adapters translate them into
|
|
:class:`packages.shared.errors.AppError` at the boundary.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
|
|
class DomainError(Exception):
|
|
"""Base class for all pure-domain rule violations."""
|
|
|
|
|
|
class QuotaExceededError(DomainError):
|
|
def __init__(self, *, required: int, remaining: int) -> None:
|
|
self.required = required
|
|
self.remaining = remaining
|
|
super().__init__(f"quota exceeded: required={required} remaining={remaining}")
|
|
|
|
|
|
class InvalidTaskTransitionError(DomainError):
|
|
def __init__(self, *, current: str, target: str) -> None:
|
|
self.current = current
|
|
self.target = target
|
|
super().__init__(f"invalid task transition: {current} -> {target}")
|