generating-production-code-recipe

Production Code Standards (Python)

SCOPE: Production code under src/ only. Test code has separate standards.

Quick Reference

Protocol-based service:

from dataclasses import dataclass
from typing import Protocol

class UserServiceProtocol(Protocol):
    async def create(self, dto: CreateUserDTO) -> User: ...

@dataclass(frozen=True)
class UserService(UserServiceProtocol):
    repository: UserRepositoryProtocol
    config: AppConfig

    async def create(self, dto: CreateUserDTO) -> User:
        return await self._validate(dto).then(self._save)

FastAPI handler:

from fastapi import APIRouter, Depends, HTTPException

@router.post("/users")
async def create_user(
    dto: CreateUserDTO,
    service: Annotated[UserServiceProtocol, Depends(get_user_service)],
) -> UserResponse:
    try:
        user = await service.create(dto)
        return UserResponse.from_entity(user)
    except ValidationError as e:
        raise HTTPException(400, e.user_friendly_message())

Domain entity (Pydantic):

from pydantic import BaseModel, Field

class User(BaseModel):
    id: UserId
    email: EmailStr = Field(description="User email")
    created_at: datetime

    class Config:
        frozen = True

When to Use

Code generation | service implementation | business logic | API endpoints | error handling

Tooling (MANDATORY)

Tool	Purpose	Command
ruff	Lint + format (replaces Black/flake8/isort)	ruff check src/ && ruff format src/
mypy	Static type checking	mypy src/
uv	Package management (no pip)	uv add pkg / uv run pytest
pyproject.toml	Single config file for all tools	no pytest.ini, setup.cfg, .flake8

See tooling.md for full configuration.

Architecture (HEXAGONAL)

Layer structure:

Layer	Responsibility	Dependencies
Domain	Business logic, entities	None (pure)
Service	Orchestration, workflows	Domain + protocols
API (Inbound)	Routes, DTOs, validation	Service protocols
Infra (Outbound)	DB, HTTP, external systems	Domain protocols

Dependency rule: Domain ← Service ← API/Infra

# user/domain/entities/user.py         ← feature-by-package (preferred)
class User(BaseModel):
    id: UserId
    email: EmailStr

# user/domain/interfaces/user_protocol.py
class UserServiceProtocol(Protocol):
    async def create(self, dto: CreateUserDTO) -> User: ...

# user/service/user_service.py
@dataclass(frozen=True)
class UserService(UserServiceProtocol):
    repository: UserRepositoryProtocol  # Protocol, not concrete

# user/api/routes.py
@router.post("/users")
async def create(
    dto: CreateUserDTO,
    service: Annotated[UserServiceProtocol, Depends(get_service)],
): ...

Immutability (ABSOLUTE)

Pattern	Status
@dataclass(frozen=True)	✓ Services, value objects
Pydantic BaseModel	✓ Entities, DTOs (frozen via Config)
list.append(), dict[key] = val	✗ Mutation
[...list, item], {**dict, key: val}	✓ Immutable operations
class variables	✗ Global state
@functools.cached_property	✓ Lazy immutable computed

Service pattern:

@dataclass(frozen=True)
class AgentService(AgentProtocol):
    llm_conductor: LLMConductorProtocol
    event_logging: EventLoggingProtocol
    tools: ToolsProtocol
    # All dependencies frozen

    async def execute(self, request: AgentRequest) -> AgentResponse:
        return await self._validate(request).then(self._process)

Entity pattern:

class AgentConfig(BaseModel):
    agent_name: str = Field(description="Agent name")
    timeout: int = Field(default=5000)

    class Config:
        frozen = True  # Immutable Pydantic model

Absolute Prohibitions

Never Allowed	Use Instead
# comments, """docstrings""" in logic	Self-documenting code
type: ignore, noqa, pylint: disable	Fix type errors properly
dict/list/tuple mutability	@frozen dataclass or Pydantic frozen
Any type hint	Union type or specific type
Global state, class variables	@dataclass(frozen=True) + DI
Optional[T]	T | None (PEP 604 syntax)
Union[A, B]	A | B (PEP 604 syntax)
Mutable default arguments	None with initialization
all exports	Named exports
from x import *	Explicit imports
Dict/List without typing	dict[Key, Value], list[Item]
Concrete dependencies	Protocol-based dependencies
Multiple exception catches	Hierarchical exception design
dict for structured data	Pydantic model, dataclass, or NamedTuple
In-function imports	Top-of-file; order: stdlib → third-party → local

Error Handling (MANDATORY)

Exception hierarchy with dataclasses:

from dataclasses import dataclass
from abc import ABC, abstractmethod

@dataclass
class DomainError(ABC, Exception):
    @abstractmethod
    def user_friendly_message(self) -> str: ...

    @property
    def error_type(self) -> str:
        return self.__class__.__name__

@dataclass
class ValidationError(DomainError):
    field: str
    message: str

    def user_friendly_message(self) -> str:
        return f"Invalid {self.field}: {self.message}"

@dataclass
class NotFoundError(DomainError):
    resource: str
    id: str

    def user_friendly_message(self) -> str:
        return f"{self.resource} with id {self.id} not found"

Service raises, handler catches:

# Service layer
@dataclass(frozen=True)
class UserService:
    async def create(self, dto: CreateUserDTO) -> User:
        if await self.repository.exists(dto.email):
            raise DuplicateError("email", dto.email)
        return await self.repository.save(User(**dto.dict()))

# API layer
@router.post("/users")
async def create_user(dto: CreateUserDTO, service: UserServiceProtocol):
    try:
        user = await service.create(dto)
        return UserResponse.from_entity(user)
    except ValidationError as e:
        raise HTTPException(400, e.user_friendly_message())
    except DuplicateError as e:
        raise HTTPException(409, e.user_friendly_message())
    except DomainError as e:
        raise HTTPException(500, "Internal error")

Pattern matching (Python 3.10+):

async def process(self, request: AgentRequest) -> AgentResponse:
    match request:
        case HumanInputRequest():
            return await self._handle_human(request)
        case ToolResponseRequest():
            return await self._handle_tool(request)
        case _:
            raise NotImplementedError(f"Unknown: {type(request)}")

Type System (MODERN PYTHON)

Pattern	Status
Untyped parameters	✗ Use type hints
Any type hint	✗ Use union or specific
Optional[T]	✗ Use T | None (PEP 604)
Union[A, B]	✗ Use A | B (PEP 604)
dict/list without params	✗ Use dict[K, V], list[T]
Mutable dataclass	✗ Use @dataclass(frozen=True)
Untyped functions	✗ Annotate params and return
Protocol without runtime_checkable	✓ Use @runtime_checkable if needed
Simple immutable lightweight struct	✓ typing.NamedTuple acceptable

Modern type syntax:

from dataclasses import dataclass
from typing import NewType, Protocol, Annotated
from pydantic import BaseModel, Field

UserId = NewType('UserId', str)

class User(BaseModel):
    id: UserId
    email: str | None  # PEP 604 instead of Optional[str]
    roles: list[str]  # Built-in generics (Python 3.9+)

    class Config:
        frozen = True

class UserRepositoryProtocol(Protocol):
    async def find(self, id: UserId) -> User | None: ...
    async def save(self, user: User) -> User: ...

## Naming Conventions

### Structure

**Feature-by-package (preferred):**

src/ ├── order/ │ ├── domain/entities/ # Pydantic models │ ├── domain/exceptions/ # Domain errors │ ├── domain/interfaces/ # Service + repo protocols │ ├── service/ # Service implementations │ ├── api/ # Routes + deps │ └── infra/ # DB + HTTP clients └── user/ └── ... # same structure

**Layered (small services only):** `domain/` | `api/` | `infra/`

### Naming Patterns

| Component | Pattern | Example |
|-----------|---------|---------|
| Service interface | {Domain}Protocol | UserServiceProtocol |
| Service impl | {Domain}Service | UserService |
| Repository interface | {Entity}RepositoryProtocol | UserRepositoryProtocol |
| Repository impl | {Entity}Repository | UserRepository |
| Entities | {Domain} | User, AgentConfig |
| Exceptions | {Type}Error | ValidationError, NotFoundError |
| Private methods | _method_name | _validate, _persist |

### Code Style

```python
# Functions: snake_case
async def validate_email(email: str) -> bool:
    return EMAIL_REGEX.match(email) is not None

# Classes: PascalCase
class UserService(UserServiceProtocol):
    ...

# Constants: UPPER_SNAKE_CASE
MAX_RETRIES = 3
DEFAULT_TIMEOUT_MS = 5000

# Type aliases: PascalCase
UserId = NewType('UserId', str)

Dependency Injection (FASTAPI)

FastAPI Depends pattern:

# api/deps.py
from functools import lru_cache

@lru_cache
def get_user_repository() -> UserRepositoryProtocol:
    return UserRepository(db_url=settings.DATABASE_URL)

@lru_cache
def get_user_service(
    repository: Annotated[UserRepositoryProtocol, Depends(get_user_repository)],
) -> UserServiceProtocol:
    return UserService(repository=repository, config=settings)

# api/users.py
@router.post("/users")
async def create_user(
    dto: CreateUserDTO,
    service: Annotated[UserServiceProtocol, Depends(get_user_service)],
) -> UserResponse:
    user = await service.create(dto)
    return UserResponse.from_entity(user)

Lazy initialization with cached_property:

import functools

@dataclass(frozen=True)
class HttpClientService:
    config: HttpConfig

    @functools.cached_property
    def default_client(self) -> httpx.AsyncClient:
        return httpx.AsyncClient(
            limits=httpx.Limits(max_connections=4096),
        )

Configuration (PYDANTIC SETTINGS)

Pydantic BaseSettings:

from pydantic import BaseSettings, Field

class Settings(BaseSettings):
    API_PREFIX: str = "/api"
    DATABASE_URL: str = Field(..., env="DATABASE_URL")
    API_KEY: str = Field(..., env="API_KEY")
    TIMEOUT_MS: int = Field(default=5000)

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

settings = Settings()

Input Validation (PYDANTIC)

Auto-validation with Pydantic:

from pydantic import BaseModel, EmailStr, Field, validator

class CreateUserDTO(BaseModel):
    name: str = Field(min_length=3, max_length=100)
    email: EmailStr = Field(description="User email address")
    age: int = Field(ge=0, le=150)
    tags: list[str] = Field(default_factory=list)  # mutable default → default_factory

    @validator('name')
    def name_must_not_contain_spaces(cls, v: str) -> str:
        if ' ' in v:
            raise ValueError('Name cannot contain spaces')
        return v

# FastAPI auto-validates
@router.post("/users")
async def create_user(dto: CreateUserDTO, service: UserServiceProtocol):
    # dto is already validated by Pydantic
    user = await service.create(dto)
    return UserResponse.from_entity(user)

Verification Protocol

CHECK:
1. Hexagonal architecture → Domain/Service/API/Infra separation
2. Protocol-based design → All dependencies use protocols
3. Immutability → @dataclass(frozen=True) for services, Pydantic frozen for entities
4. Modern type hints → T | None (PEP 604), dict[K,V], list[T]
5. Exception hierarchy → Dataclass errors with user_friendly_message()
6. Service raises, handler catches → Try/except only in API layer
7. Dependency injection → FastAPI Depends with @lru_cache
8. Pydantic validation → DTOs auto-validated by FastAPI
9. Async/await → All I/O operations are async
10. Pattern matching → Use match/case for request types (Python 3.10+)
11. Type hints → ALL parameters and returns typed
12. Naming conventions → Protocol suffix, Service suffix, {module}Error
13. Configuration → Pydantic BaseSettings with .env
14. NO mutable defaults → None with initialization
15. NO concrete dependencies → Only protocols
16. Lazy loading → @functools.cached_property for expensive init
17. Structure → domain/api/infra folders with proper separation
18. Prohibitions → Check ALL absolute prohibitions table
19. Tooling → ruff + mypy pass; uv used; pyproject.toml as single config
20. No dict for structured data → Pydantic / dataclass / NamedTuple used
21. Imports → top-of-file only; stdlib → third-party → local order
22. Mutable Pydantic defaults → Field(default_factory=...) not Field(default=[])
23. All pass → Generate code

ANY fail → REJECT with violation

Validation Workflow

Phase	Action
1. SCOPE	Extract from context: files, mode (informative | executive), sessionId
2. VERIFY	Run ALL 19 checks on code
3. VIOLATIONS	Collect violations with file:line, check name, severity
4. REPORT	✓ Pass → Proceed | ✗ Fail → Return violations
5. METRICS	Call mcp__agent-orchestrator__store-skill-metrics
6. OUTPUT	Return JSON: violations[], fixRate, finalViolations

Metrics Structure:

{
  "sessionId": str,
  "skill": "python-services:production-code-recipe",
  "initialViolations": int,
  "iterations": int,
  "fixesApplied": int,
  "finalViolations": int,
  "mode": "informative" | "executive",
  "duration": float
}

Output Format:

{
  "status": "success" | "failed",
  "violations": [
    {
      "file": "src/domain/user/services/user_service.py",
      "line": 42,
      "check": "Protocol-based design",
      "violation": "Concrete dependency instead of protocol",
      "severity": "critical"
    }
  ],
  "metrics": {
    "initialViolations": 5,
    "finalViolations": 0,
    "fixRate": 1.0
  }
}