{"id":"15840320-0b9d-4c6a-b99d-8af86c99e167","entityType":"agent","slug":"clawhub-skills-1kalin-afrexai-fastapi-production","name":"afrexai-fastapi-production","canonicalUrl":"https://www.xpersona.co/agent/clawhub-skills-1kalin-afrexai-fastapi-production","canonicalPath":"/agent/clawhub-skills-1kalin-afrexai-fastapi-production","generatedAt":"2026-10-09T21:57:49.520Z","source":"CLAWHUB","claimStatus":"UNCLAIMED","verificationTier":"NONE","summary":{"evidence":{"source":"editorial-content","verified":true,"confidence":"high","updatedAt":"2026-04-15T00:45:39.800Z","emptyReason":null},"description":"FastAPI Production Engineering FastAPI Production Engineering Complete methodology for building, deploying, and scaling production FastAPI applications. Not a tutorial — a production operating system. Quick Health Check (/16) Score 2 points each. Total < 8 = critical work needed. | Signal | Healthy | Unhealthy | |--------|---------|-----------| | Type safety | Pydantic v2 models everywhere | dict returns, no validation | | Error handling | Structu","descriptionLabel":"Technical summary","evidenceSummary":"Capability contract not published. No trust telemetry is available yet. Last updated 4/15/2026.","installCommand":"clawhub skill install skills:1kalin:afrexai-fastapi-production","sourceUrl":"https://github.com/openclaw/skills/tree/main/skills/1kalin/afrexai-fastapi-production","homepage":null,"primaryLinks":[{"label":"View on ClawHub","url":"https://github.com/openclaw/skills/tree/main/skills/1kalin/afrexai-fastapi-production","kind":"source"}],"safetyScore":84,"overallRank":62,"popularityScore":50,"trustScore":null,"claimedByName":null,"isOwner":false,"seoDescription":"FastAPI Production Engineering FastAPI Production Engineering Complete methodology for building, deploying, and scaling production FastAPI applications. Not a t"},"coverage":{"evidence":{"source":"public-profile","verified":false,"confidence":"medium","updatedAt":"2026-04-15T00:45:39.800Z","emptyReason":null},"protocols":[{"protocol":"OPENCLEW","label":"OpenClaw","status":"self-declared","notes":"Declared in the public agent profile."}],"capabilities":[{"label":"we","status":"self-declared"},{"label":"7","status":"self-declared"}],"verifiedCount":0,"selfDeclaredCount":3,"capabilityMatrix":{"rows":[{"key":"OPENCLEW","type":"protocol","support":"unknown","confidenceSource":"profile","notes":"Listed on profile"},{"key":"we","type":"capability","support":"supported","confidenceSource":"profile","notes":"Declared in agent profile metadata"},{"key":"7","type":"capability","support":"supported","confidenceSource":"profile","notes":"Declared in agent profile metadata"}],"flattenedTokens":"protocol:OPENCLEW|unknown|profile capability:we|supported|profile capability:7|supported|profile"}},"adoption":{"evidence":{"source":"no-adoption-signals","verified":false,"confidence":"low","updatedAt":"2026-04-15T00:45:39.800Z","emptyReason":"No source adoption metrics were available."},"stars":null,"forks":null,"downloads":null,"packageName":null,"latestVersion":null,"tractionLabel":null},"release":{"evidence":{"source":"agent-index","verified":false,"confidence":"medium","updatedAt":"2026-02-25T06:17:26.837Z","emptyReason":null},"lastUpdatedAt":"2026-04-15T00:45:39.800Z","lastCrawledAt":"2026-02-25T06:17:26.837Z","lastIndexedAt":null,"nextCrawlAt":"2026-02-26T06:17:26.837Z","lastVerifiedAt":null,"highlights":[]},"execution":{"evidence":{"source":"CLAWHUB","verified":false,"confidence":"low","updatedAt":null,"emptyReason":"No published capability contract is available yet."},"installCommand":"clawhub skill install skills:1kalin:afrexai-fastapi-production","setupComplexity":"low","setupSteps":["Setup complexity is LOW. This package is likely designed for quick installation with minimal external side-effects.","Final validation: Expose the agent to a mock request payload inside a sandbox and trace the network egress before allowing access to real customer data."],"contract":{"contractStatus":"missing","authModes":[],"requires":[],"forbidden":[],"supportsMcp":false,"supportsA2a":false,"supportsStreaming":false,"inputSchemaRef":null,"outputSchemaRef":null,"dataRegion":null,"contractUpdatedAt":null,"sourceUpdatedAt":null,"freshnessSeconds":null},"invocationGuide":{"preferredApi":{"snapshotUrl":"https://www.xpersona.co/api/v1/agents/clawhub-skills-1kalin-afrexai-fastapi-production/snapshot","contractUrl":"https://www.xpersona.co/api/v1/agents/clawhub-skills-1kalin-afrexai-fastapi-production/contract","trustUrl":"https://www.xpersona.co/api/v1/agents/clawhub-skills-1kalin-afrexai-fastapi-production/trust"},"curlExamples":["curl -s \"https://www.xpersona.co/api/v1/agents/clawhub-skills-1kalin-afrexai-fastapi-production/snapshot\"","curl -s \"https://www.xpersona.co/api/v1/agents/clawhub-skills-1kalin-afrexai-fastapi-production/contract\"","curl -s \"https://www.xpersona.co/api/v1/agents/clawhub-skills-1kalin-afrexai-fastapi-production/trust\""],"jsonRequestTemplate":{"query":"summarize this repo","constraints":{"maxLatencyMs":2000,"protocolPreference":["OPENCLEW"]}},"jsonResponseTemplate":{"ok":true,"result":{"summary":"...","confidence":0.9},"meta":{"source":"CLAWHUB","generatedAt":"2026-10-09T21:57:49.519Z"}},"retryPolicy":{"maxAttempts":3,"backoffMs":[500,1500,3500],"retryableConditions":["HTTP_429","HTTP_503","NETWORK_TIMEOUT"]}},"endpoints":{"dossierUrl":"https://www.xpersona.co/api/v1/agents/clawhub-skills-1kalin-afrexai-fastapi-production/dossier","snapshotUrl":"https://www.xpersona.co/api/v1/agents/clawhub-skills-1kalin-afrexai-fastapi-production/snapshot","contractUrl":"https://www.xpersona.co/api/v1/agents/clawhub-skills-1kalin-afrexai-fastapi-production/contract","trustUrl":"https://www.xpersona.co/api/v1/agents/clawhub-skills-1kalin-afrexai-fastapi-production/trust"}},"reliability":{"evidence":{"source":"runtime-metrics","verified":false,"confidence":"low","updatedAt":null,"emptyReason":"No trust, reliability, or runtime telemetry is available."},"trust":{"status":"unavailable","handshakeStatus":"UNKNOWN","verificationFreshnessHours":null,"reputationScore":null,"p95LatencyMs":null,"successRate30d":null,"fallbackRate":null,"attempts30d":null,"trustUpdatedAt":null,"trustConfidence":"unknown","sourceUpdatedAt":null,"freshnessSeconds":null},"decisionGuardrails":{"doNotUseIf":["Contract metadata is missing or unavailable for deterministic execution."],"safeUseWhen":[],"riskFlags":["missing_or_unavailable_contract","trust_data_unavailable","schema_references_missing"],"operationalConfidence":"low"},"executionMetrics":{"observedLatencyMsP50":null,"observedLatencyMsP95":null,"estimatedCostUsd":null,"uptime30d":null,"rateLimitRpm":null,"rateLimitBurst":null,"lastVerifiedAt":null,"verificationSource":null},"runtimeMetrics":{"successRate":null,"avgLatencyMs":null,"avgCostUsd":null,"hallucinationRate":null,"retryRate":null,"disputeRate":null,"p50Latency":null,"p95Latency":null,"lastUpdated":null}},"benchmarks":{"evidence":{"source":"no-benchmark-data","verified":false,"confidence":"low","updatedAt":null,"emptyReason":"No benchmark suites or observed failure patterns are available."},"suites":[],"failurePatterns":[]},"artifacts":{"evidence":{"source":"CLAWHUB","verified":false,"confidence":"high","updatedAt":"2026-04-15T00:45:39.800Z","emptyReason":null},"readme":"# FastAPI Production Engineering\n\nComplete methodology for building, deploying, and scaling production FastAPI applications. Not a tutorial — a production operating system.\n\n## Quick Health Check (/16)\n\nScore 2 points each. Total < 8 = critical work needed.\n\n| Signal | Healthy | Unhealthy |\n|--------|---------|-----------|\n| Type safety | Pydantic v2 models everywhere | `dict` returns, no validation |\n| Error handling | Structured error hierarchy | Bare `HTTPException` strings |\n| Auth | JWT + dependency injection | Manual token parsing |\n| Testing | 80%+ coverage, async tests | No tests or sync-only |\n| Database | Async ORM, migrations | Raw SQL, no migrations |\n| Observability | Structured logging + tracing | `print()` debugging |\n| Deployment | Multi-stage Docker, health checks | `uvicorn main:app` on bare metal |\n| Documentation | Auto-generated, accurate OpenAPI | Default `/docs` untouched |\n\n## Phase 1: Project Architecture\n\n### Recommended Structure\n\n```\nsrc/\n├── app/\n│   ├── __init__.py\n│   ├── main.py              # App factory\n│   ├── config.py             # Pydantic Settings\n│   ├── dependencies.py       # Shared DI\n│   ├── middleware.py          # Custom middleware\n│   ├── features/\n│   │   ├── users/\n│   │   │   ├── __init__.py\n│   │   │   ├── router.py     # Endpoints\n│   │   │   ├── schemas.py    # Pydantic models\n│   │   │   ├── service.py    # Business logic\n│   │   │   ├── repository.py # Data access\n│   │   │   ├── models.py     # SQLAlchemy/SQLModel\n│   │   │   ├── dependencies.py\n│   │   │   └── exceptions.py\n│   │   ├── auth/\n│   │   ├── orders/\n│   │   └── ...\n│   ├── core/\n│   │   ├── database.py       # Engine, session factory\n│   │   ├── security.py       # JWT, hashing\n│   │   ├── errors.py         # Error hierarchy\n│   │   └── logging.py        # Structlog config\n│   └── shared/\n│       ├── pagination.py\n│       ├── filters.py\n│       └── responses.py\n├── migrations/               # Alembic\n├── tests/\n│   ├── conftest.py\n│   ├── unit/\n│   ├── integration/\n│   └── e2e/\n├── pyproject.toml\n├── Dockerfile\n└── docker-compose.yml\n```\n\n### 7 Architecture Rules\n\n1. **Feature-based modules** — group by domain, not by layer\n2. **Router → Service → Repository** — strict layering, no skipping\n3. **Dependency injection everywhere** — use `Depends()` for testability\n4. **Pydantic models at boundaries** — validate all input AND output\n5. **No business logic in routers** — routers are thin, services are thick\n6. **Config via environment** — Pydantic Settings with `.env` support\n7. **Async by default** — use async def for all I/O-bound operations\n\n### Framework Selection Context\n\n```yaml\n# When to choose FastAPI over alternatives\nfastapi_is_best_when:\n  - \"You need auto-generated OpenAPI docs\"\n  - \"Team knows Python type hints\"\n  - \"API-first (no server-rendered HTML as primary)\"\n  - \"High concurrency with async I/O\"\n  - \"Microservice or API gateway\"\n\nconsider_alternatives:\n  django: \"Full-featured web app with admin, ORM, auth batteries\"\n  flask: \"Simple app, team prefers explicit over magic\"\n  litestar: \"Need WebSocket-heavy or more opinionated framework\"\n  hono_or_express: \"Team prefers TypeScript\"\n```\n\n## Phase 2: Configuration & Environment\n\n### Pydantic Settings Pattern\n\n```python\nfrom pydantic_settings import BaseSettings\nfrom pydantic import SecretStr, field_validator\nfrom functools import lru_cache\n\nclass Settings(BaseSettings):\n    # App\n    app_name: str = \"MyAPI\"\n    debug: bool = False\n    environment: str = \"production\"  # development | staging | production\n    \n    # Server\n    host: str = \"0.0.0.0\"\n    port: int = 8000\n    workers: int = 4\n    \n    # Database\n    database_url: SecretStr  # Required — no default\n    db_pool_size: int = 20\n    db_max_overflow: int = 10\n    db_pool_timeout: int = 30\n    \n    # Auth\n    jwt_secret: SecretStr  # Required\n    jwt_algorithm: str = \"HS256\"\n    jwt_expire_minutes: int = 30\n    \n    # Redis\n    redis_url: str = \"redis://localhost:6379/0\"\n    \n    # CORS\n    cors_origins: list[str] = [\"http://localhost:3000\"]\n    \n    @field_validator(\"environment\")\n    @classmethod\n    def validate_environment(cls, v: str) -> str:\n        allowed = {\"development\", \"staging\", \"production\"}\n        if v not in allowed:\n            raise ValueError(f\"environment must be one of {allowed}\")\n        return v\n    \n    model_config = {\"env_file\": \".env\", \"env_file_encoding\": \"utf-8\"}\n\n@lru_cache\ndef get_settings() -> Settings:\n    return Settings()\n```\n\n### 5 Configuration Rules\n\n1. **Never hardcode secrets** — use `SecretStr` for sensitive values\n2. **Fail fast** — required fields have no defaults; app won't start without them\n3. **Validate at startup** — use `@field_validator` for constraint checking\n4. **Cache settings** — `@lru_cache` ensures single parse\n5. **Type everything** — no `str` for structured values; use enums, Literal types\n\n## Phase 3: Pydantic v2 Mastery\n\n### Schema Design Patterns\n\n```python\nfrom pydantic import BaseModel, Field, ConfigDict\nfrom datetime import datetime\nfrom uuid import UUID\n\n# Base with common config\nclass AppSchema(BaseModel):\n    model_config = ConfigDict(\n        from_attributes=True,      # ORM mode\n        str_strip_whitespace=True,  # Auto-strip\n        validate_default=True,      # Validate defaults too\n    )\n\n# Input schemas (what the API accepts)\nclass UserCreate(AppSchema):\n    email: str = Field(..., pattern=r\"^[\\w\\.-]+@[\\w\\.-]+\\.\\w+$\")\n    name: str = Field(..., min_length=1, max_length=100)\n    password: str = Field(..., min_length=8, max_length=128)\n\nclass UserUpdate(AppSchema):\n    name: str | None = Field(None, min_length=1, max_length=100)\n    email: str | None = Field(None, pattern=r\"^[\\w\\.-]+@[\\w\\.-]+\\.\\w+$\")\n\n# Output schemas (what the API returns)\nclass UserResponse(AppSchema):\n    id: UUID\n    email: str\n    name: str\n    created_at: datetime\n    # Note: password is NEVER in response schema\n\n# List response with pagination\nclass PaginatedResponse[T](AppSchema):\n    items: list[T]\n    total: int\n    page: int\n    page_size: int\n    has_next: bool\n```\n\n### 8 Pydantic Rules\n\n1. **Separate Create/Update/Response schemas** — never reuse input as output\n2. **Never expose internal fields** — no passwords, internal IDs, or debug info in responses\n3. **Use Field() for constraints** — min/max length, regex patterns, gt/lt for numbers\n4. **Enable `from_attributes=True`** — for ORM model → schema conversion\n5. **Use generics for wrappers** — `PaginatedResponse[T]`, `ApiResponse[T]`\n6. **Validate at boundaries** — request body, query params, path params, headers\n7. **Use computed fields** — `@computed_field` for derived values\n8. **Document with examples** — `model_config = {\"json_schema_extra\": {\"examples\": [...]}}`\n\n## Phase 4: Error Handling Architecture\n\n### Structured Error Hierarchy\n\n```python\nfrom fastapi import Request\nfrom fastapi.responses import JSONResponse\nfrom starlette.status import (\n    HTTP_400_BAD_REQUEST, HTTP_401_UNAUTHORIZED,\n    HTTP_403_FORBIDDEN, HTTP_404_NOT_FOUND,\n    HTTP_409_CONFLICT, HTTP_422_UNPROCESSABLE_ENTITY,\n    HTTP_429_TOO_MANY_REQUESTS, HTTP_500_INTERNAL_SERVER_ERROR,\n)\n\nclass AppError(Exception):\n    \"\"\"Base application error.\"\"\"\n    def __init__(\n        self,\n        message: str,\n        code: str,\n        status_code: int = HTTP_500_INTERNAL_SERVER_ERROR,\n        details: dict | None = None,\n    ):\n        self.message = message\n        self.code = code\n        self.status_code = status_code\n        self.details = details or {}\n        super().__init__(message)\n\nclass NotFoundError(AppError):\n    def __init__(self, resource: str, identifier: str | int):\n        super().__init__(\n            message=f\"{resource} not found: {identifier}\",\n            code=\"NOT_FOUND\",\n            status_code=HTTP_404_NOT_FOUND,\n            details={\"resource\": resource, \"identifier\": str(identifier)},\n        )\n\nclass ConflictError(AppError):\n    def __init__(self, message: str, field: str | None = None):\n        super().__init__(\n            message=message, code=\"CONFLICT\",\n            status_code=HTTP_409_CONFLICT,\n            details={\"field\": field} if field else {},\n        )\n\nclass AuthenticationError(AppError):\n    def __init__(self, message: str = \"Invalid credentials\"):\n        super().__init__(message=message, code=\"UNAUTHORIZED\", status_code=HTTP_401_UNAUTHORIZED)\n\nclass AuthorizationError(AppError):\n    def __init__(self, message: str = \"Insufficient permissions\"):\n        super().__init__(message=message, code=\"FORBIDDEN\", status_code=HTTP_403_FORBIDDEN)\n\nclass ValidationError(AppError):\n    def __init__(self, message: str, errors: list[dict] | None = None):\n        super().__init__(\n            message=message, code=\"VALIDATION_ERROR\",\n            status_code=HTTP_422_UNPROCESSABLE_ENTITY,\n            details={\"errors\": errors or []},\n        )\n\nclass RateLimitError(AppError):\n    def __init__(self, retry_after: int = 60):\n        super().__init__(\n            message=\"Rate limit exceeded\", code=\"RATE_LIMITED\",\n            status_code=HTTP_429_TOO_MANY_REQUESTS,\n            details={\"retry_after\": retry_after},\n        )\n\n# Global error handler\nasync def app_error_handler(request: Request, exc: AppError) -> JSONResponse:\n    return JSONResponse(\n        status_code=exc.status_code,\n        content={\n            \"error\": {\n                \"code\": exc.code,\n                \"message\": exc.message,\n                \"details\": exc.details,\n            }\n        },\n    )\n\n# Register in app factory\n# app.add_exception_handler(AppError, app_error_handler)\n```\n\n### 6 Error Handling Rules\n\n1. **Never return bare strings** — always structured `{\"error\": {\"code\", \"message\", \"details\"}}`\n2. **Use domain-specific errors** — `NotFoundError(\"User\", user_id)` not `HTTPException(404)`\n3. **Global handler catches all** — register `AppError` handler in app factory\n4. **Log server errors, don't expose** — 5xx returns generic message, logs full traceback\n5. **Include actionable details** — which field failed, what's allowed, retry-after for rate limits\n6. **Never leak internals** — no stack traces, SQL queries, or file paths in responses\n\n## Phase 5: Authentication & Authorization\n\n### JWT + Dependency Injection Pattern\n\n```python\nfrom fastapi import Depends, Security\nfrom fastapi.security import HTTPBearer, HTTPAuthorizationCredentials\nfrom jose import jwt, JWTError\nfrom datetime import datetime, timedelta, timezone\n\nsecurity = HTTPBearer()\n\ndef create_access_token(user_id: str, roles: list[str], settings: Settings) -> str:\n    expire = datetime.now(timezone.utc) + timedelta(minutes=settings.jwt_expire_minutes)\n    payload = {\n        \"sub\": user_id,\n        \"roles\": roles,\n        \"exp\": expire,\n        \"iat\": datetime.now(timezone.utc),\n    }\n    return jwt.encode(payload, settings.jwt_secret.get_secret_value(), algorithm=settings.jwt_algorithm)\n\nasync def get_current_user(\n    credentials: HTTPAuthorizationCredentials = Security(security),\n    settings: Settings = Depends(get_settings),\n    db: AsyncSession = Depends(get_db),\n) -> User:\n    try:\n        payload = jwt.decode(\n            credentials.credentials,\n            settings.jwt_secret.get_secret_value(),\n            algorithms=[settings.jwt_algorithm],\n        )\n        user_id = payload.get(\"sub\")\n        if not user_id:\n            raise AuthenticationError(\"Invalid token payload\")\n    except JWTError:\n        raise AuthenticationError(\"Invalid or expired token\")\n    \n    user = await db.get(User, user_id)\n    if not user:\n        raise AuthenticationError(\"User not found\")\n    return user\n\n# Role-based authorization\ndef require_role(*roles: str):\n    async def checker(user: User = Depends(get_current_user)) -> User:\n        if not any(r in user.roles for r in roles):\n            raise AuthorizationError(f\"Requires one of: {', '.join(roles)}\")\n        return user\n    return checker\n\n# Usage in router\n@router.get(\"/admin/users\")\nasync def list_users(\n    admin: User = Depends(require_role(\"admin\", \"superadmin\")),\n    service: UserService = Depends(get_user_service),\n):\n    return await service.list_all()\n```\n\n### 10-Point Security Checklist\n\n| # | Check | Priority |\n|---|-------|----------|\n| 1 | JWT secret ≥ 256 bits, from env | P0 |\n| 2 | Token expiry ≤ 30 min for access, ≤ 7 days refresh | P0 |\n| 3 | Password hashed with bcrypt/argon2 | P0 |\n| 4 | CORS configured per environment | P0 |\n| 5 | Rate limiting on auth endpoints | P0 |\n| 6 | HTTPS enforced (redirect HTTP) | P0 |\n| 7 | Security headers (HSTS, CSP, X-Frame) | P1 |\n| 8 | Input validation on ALL endpoints | P1 |\n| 9 | SQL injection prevented (parameterized queries) | P0 |\n| 10 | Dependency scanning (safety/pip-audit) | P1 |\n\n## Phase 6: Database Patterns\n\n### Async SQLAlchemy + Repository Pattern\n\n```python\nfrom sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker\nfrom sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column\nfrom sqlalchemy import select, func\nfrom uuid import uuid4, UUID\nfrom datetime import datetime, timezone\n\n# Engine setup\nengine = create_async_engine(\n    settings.database_url.get_secret_value(),\n    pool_size=settings.db_pool_size,\n    max_overflow=settings.db_max_overflow,\n    pool_timeout=settings.db_pool_timeout,\n    pool_pre_ping=True,  # Check connection health\n    echo=settings.debug,\n)\n\nSessionFactory = async_sessionmaker(engine, expire_on_commit=False)\n\nasync def get_db() -> AsyncGenerator[AsyncSession, None]:\n    async with SessionFactory() as session:\n        try:\n            yield session\n            await session.commit()\n        except Exception:\n            await session.rollback()\n            raise\n\n# Base model with common fields\nclass Base(DeclarativeBase):\n    pass\n\nclass TimestampMixin:\n    created_at: Mapped[datetime] = mapped_column(default=lambda: datetime.now(timezone.utc))\n    updated_at: Mapped[datetime] = mapped_column(\n        default=lambda: datetime.now(timezone.utc),\n        onupdate=lambda: datetime.now(timezone.utc),\n    )\n\n# Repository pattern\nclass BaseRepository[T]:\n    def __init__(self, session: AsyncSession, model: type[T]):\n        self.session = session\n        self.model = model\n    \n    async def get_by_id(self, id: UUID) -> T | None:\n        return await self.session.get(self.model, id)\n    \n    async def get_or_raise(self, id: UUID) -> T:\n        entity = await self.get_by_id(id)\n        if not entity:\n            raise NotFoundError(self.model.__name__, str(id))\n        return entity\n    \n    async def list(\n        self, *, offset: int = 0, limit: int = 20, **filters\n    ) -> tuple[list[T], int]:\n        query = select(self.model)\n        count_query = select(func.count()).select_from(self.model)\n        \n        for field, value in filters.items():\n            if value is not None:\n                query = query.where(getattr(self.model, field) == value)\n                count_query = count_query.where(getattr(self.model, field) == value)\n        \n        total = await self.session.scalar(count_query) or 0\n        result = await self.session.execute(\n            query.offset(offset).limit(limit).order_by(self.model.created_at.desc())\n        )\n        return list(result.scalars().all()), total\n    \n    async def create(self, entity: T) -> T:\n        self.session.add(entity)\n        await self.session.flush()\n        return entity\n    \n    async def delete(self, entity: T) -> None:\n        await self.session.delete(entity)\n```\n\n### ORM Selection Guide\n\n| ORM | Best For | Async | Type Safety | Learning Curve |\n|-----|----------|-------|-------------|----------------|\n| **SQLAlchemy 2.0** | Complex queries, enterprise | ✅ | ✅ Mapped[] | Medium |\n| **SQLModel** | Simple CRUD, Pydantic sync | ✅ | ✅ | Low |\n| **Tortoise** | Django-like feel | ✅ | Partial | Low |\n| **Piccolo** | Modern, migrations built-in | ✅ | ✅ | Low |\n\n**Recommendation:** SQLAlchemy 2.0 for production. SQLModel for prototypes.\n\n### Migration Strategy (Alembic)\n\n```bash\n# Setup\nalembic init migrations\n# Edit alembic.ini: sqlalchemy.url = from env\n\n# Generate migration\nalembic revision --autogenerate -m \"add users table\"\n\n# Apply\nalembic upgrade head\n\n# Rollback\nalembic downgrade -1\n```\n\n**Migration Rules:**\n1. Always review autogenerated migrations before applying\n2. Never edit applied migrations — create new ones\n3. Test migrations in staging before production\n4. Include `downgrade()` for every `upgrade()`\n5. Use `batch_alter_table` for SQLite compatibility\n\n## Phase 7: Testing Strategy\n\n### Test Pyramid\n\n| Level | Coverage Target | Tools | Focus |\n|-------|----------------|-------|-------|\n| Unit | 80%+ | pytest, unittest.mock | Service logic, validators |\n| Integration | Key paths | pytest-asyncio, testcontainers | DB queries, external APIs |\n| E2E | Critical flows | httpx.AsyncClient | Full request→response |\n| Contract | API boundaries | schemathesis | OpenAPI compliance |\n\n### Test Patterns\n\n```python\nimport pytest\nfrom httpx import AsyncClient, ASGITransport\nfrom app.main import create_app\n\n@pytest.fixture\nasync def app():\n    app = create_app()\n    yield app\n\n@pytest.fixture\nasync def client(app):\n    transport = ASGITransport(app=app)\n    async with AsyncClient(transport=transport, base_url=\"http://test\") as ac:\n        yield ac\n\n@pytest.fixture\nasync def auth_client(client, test_user):\n    token = create_access_token(test_user.id, test_user.roles)\n    client.headers[\"Authorization\"] = f\"Bearer {token}\"\n    return client\n\n# E2E test\n@pytest.mark.asyncio\nasync def test_create_user(client: AsyncClient):\n    response = await client.post(\"/api/users\", json={\n        \"email\": \"test@example.com\",\n        \"name\": \"Test User\",\n        \"password\": \"securepass123\",\n    })\n    assert response.status_code == 201\n    data = response.json()\n    assert data[\"email\"] == \"test@example.com\"\n    assert \"password\" not in data  # Never expose\n\n# Unit test (service layer)\n@pytest.mark.asyncio\nasync def test_user_service_duplicate_email(user_service, mock_repo):\n    mock_repo.get_by_email.return_value = existing_user\n    with pytest.raises(ConflictError, match=\"Email already registered\"):\n        await user_service.create(UserCreate(email=\"taken@example.com\", ...))\n\n# Parametrized validation\n@pytest.mark.parametrize(\"email,expected\", [\n    (\"valid@example.com\", True),\n    (\"invalid\", False),\n    (\"\", False),\n    (\"a@b.c\", True),\n])\ndef test_email_validation(email, expected):\n    if expected:\n        UserCreate(email=email, name=\"Test\", password=\"12345678\")\n    else:\n        with pytest.raises(ValidationError):\n            UserCreate(email=email, name=\"Test\", password=\"12345678\")\n```\n\n### 7 Testing Rules\n\n1. **Test services, not routers** — business logic lives in services\n2. **Use fixtures for DI override** — swap real DB with test DB via `app.dependency_overrides`\n3. **One assertion per test** — clear what broke when it fails\n4. **Test error paths** — 40% of tests should be sad-path\n5. **Use factories for test data** — `UserFactory.create()` not manual dict construction\n6. **Async tests need `@pytest.mark.asyncio`** — or set `asyncio_mode = \"auto\"` in config\n7. **Run tests in CI** — block merge if tests fail\n\n## Phase 8: Structured Logging & Observability\n\n### Structlog Setup\n\n```python\nimport structlog\nfrom uuid import uuid4\nfrom starlette.middleware.base import BaseHTTPMiddleware\n\nstructlog.configure(\n    processors=[\n        structlog.contextvars.merge_contextvars,\n        structlog.stdlib.add_log_level,\n        structlog.stdlib.add_logger_name,\n        structlog.processors.TimeStamper(fmt=\"iso\"),\n        structlog.processors.StackInfoRenderer(),\n        structlog.processors.format_exc_info,\n        structlog.processors.JSONRenderer(),\n    ],\n    logger_factory=structlog.stdlib.LoggerFactory(),\n)\n\nlogger = structlog.get_logger()\n\n# Request ID middleware\nclass RequestIDMiddleware(BaseHTTPMiddleware):\n    async def dispatch(self, request, call_next):\n        request_id = request.headers.get(\"X-Request-ID\", str(uuid4()))\n        structlog.contextvars.clear_contextvars()\n        structlog.contextvars.bind_contextvars(\n            request_id=request_id,\n            method=request.method,\n            path=request.url.path,\n        )\n        \n        response = await call_next(request)\n        response.headers[\"X-Request-ID\"] = request_id\n        \n        logger.info(\n            \"request_completed\",\n            status_code=response.status_code,\n        )\n        return response\n```\n\n### Health Check Endpoints\n\n```python\n@router.get(\"/health\")\nasync def health():\n    \"\"\"Liveness probe — is the process running?\"\"\"\n    return {\"status\": \"ok\"}\n\n@router.get(\"/ready\")\nasync def ready(db: AsyncSession = Depends(get_db)):\n    \"\"\"Readiness probe — can we serve traffic?\"\"\"\n    checks = {}\n    try:\n        await db.execute(text(\"SELECT 1\"))\n        checks[\"database\"] = \"ok\"\n    except Exception:\n        checks[\"database\"] = \"error\"\n    \n    all_ok = all(v == \"ok\" for v in checks.values())\n    return JSONResponse(\n        status_code=200 if all_ok else 503,\n        content={\"status\": \"ok\" if all_ok else \"degraded\", \"checks\": checks},\n    )\n```\n\n## Phase 9: Performance Optimization\n\n### Priority Stack\n\n| # | Technique | Impact | Effort |\n|---|-----------|--------|--------|\n| 1 | Async database queries | High | Low |\n| 2 | Connection pooling (tuned) | High | Low |\n| 3 | Response caching (Redis) | High | Medium |\n| 4 | Background tasks for heavy work | High | Low |\n| 5 | Pagination on all list endpoints | Medium | Low |\n| 6 | Select only needed columns | Medium | Low |\n| 7 | Eager loading (joinedload) | Medium | Medium |\n| 8 | Rate limiting | Medium | Low |\n\n### Background Tasks\n\n```python\nfrom fastapi import BackgroundTasks\n\n@router.post(\"/users\", status_code=201)\nasync def create_user(\n    user_in: UserCreate,\n    background_tasks: BackgroundTasks,\n    service: UserService = Depends(get_user_service),\n):\n    user = await service.create(user_in)\n    background_tasks.add_task(send_welcome_email, user.email, user.name)\n    return user\n```\n\n### Caching Pattern\n\n```python\nfrom redis.asyncio import Redis\nimport json\n\nclass CacheService:\n    def __init__(self, redis: Redis):\n        self.redis = redis\n    \n    async def get_or_set(self, key: str, factory, ttl: int = 300):\n        cached = await self.redis.get(key)\n        if cached:\n            return json.loads(cached)\n        result = await factory()\n        await self.redis.setex(key, ttl, json.dumps(result, default=str))\n        return result\n    \n    async def invalidate(self, pattern: str):\n        keys = await self.redis.keys(pattern)\n        if keys:\n            await self.redis.delete(*keys)\n```\n\n## Phase 10: Production Deployment\n\n### Multi-Stage Dockerfile\n\n```dockerfile\n# Build stage\nFROM python:3.12-slim AS builder\nWORKDIR /app\n\nRUN pip install --no-cache-dir uv\nCOPY pyproject.toml uv.lock ./\nRUN uv sync --frozen --no-dev --no-editable\n\n# Production stage\nFROM python:3.12-slim\nWORKDIR /app\n\nRUN adduser --disabled-password --no-create-home appuser\n\nCOPY --from=builder /app/.venv /app/.venv\nCOPY src/ ./src/\nCOPY migrations/ ./migrations/\nCOPY alembic.ini ./\n\nENV PATH=\"/app/.venv/bin:$PATH\"\nENV PYTHONUNBUFFERED=1\nENV PYTHONDONTWRITEBYTECODE=1\n\nUSER appuser\nEXPOSE 8000\n\nHEALTHCHECK --interval=30s --timeout=5s --retries=3 \\\n    CMD [\"python\", \"-c\", \"import httpx; httpx.get('http://localhost:8000/health').raise_for_status()\"]\n\nCMD [\"uvicorn\", \"src.app.main:app\", \"--host\", \"0.0.0.0\", \"--port\", \"8000\", \"--workers\", \"4\"]\n```\n\n### App Factory Pattern\n\n```python\nfrom fastapi import FastAPI\nfrom contextlib import asynccontextmanager\n\n@asynccontextmanager\nasync def lifespan(app: FastAPI):\n    # Startup\n    logger.info(\"starting_up\", environment=settings.environment)\n    await init_db()\n    yield\n    # Shutdown\n    logger.info(\"shutting_down\")\n    await engine.dispose()\n\ndef create_app() -> FastAPI:\n    settings = get_settings()\n    \n    app = FastAPI(\n        title=settings.app_name,\n        lifespan=lifespan,\n        docs_url=\"/docs\" if settings.debug else None,\n        redoc_url=None,\n    )\n    \n    # Middleware (order matters — last added = first executed)\n    app.add_middleware(\n        CORSMiddleware,\n        allow_origins=settings.cors_origins,\n        allow_credentials=True,\n        allow_methods=[\"*\"],\n        allow_headers=[\"*\"],\n    )\n    app.add_middleware(RequestIDMiddleware)\n    \n    # Error handlers\n    app.add_exception_handler(AppError, app_error_handler)\n    \n    # Routers\n    app.include_router(auth_router, prefix=\"/api/auth\", tags=[\"auth\"])\n    app.include_router(users_router, prefix=\"/api/users\", tags=[\"users\"])\n    app.include_router(health_router, tags=[\"health\"])\n    \n    return app\n\napp = create_app()\n```\n\n### GitHub Actions CI\n\n```yaml\nname: CI\non: [push, pull_request]\n\njobs:\n  test:\n    runs-on: ubuntu-latest\n    services:\n      postgres:\n        image: postgres:16\n        env:\n          POSTGRES_PASSWORD: test\n          POSTGRES_DB: testdb\n        ports: [\"5432:5432\"]\n        options: >-\n          --health-cmd pg_isready\n          --health-interval 10s\n          --health-timeout 5s\n          --health-retries 5\n    \n    steps:\n      - uses: actions/checkout@v4\n      - uses: actions/setup-python@v5\n        with: { python-version: \"3.12\" }\n      - run: pip install uv && uv sync\n      - run: uv run ruff check .\n      - run: uv run mypy src/\n      - run: uv run pytest --cov=src --cov-report=xml -x\n        env:\n          DATABASE_URL: postgresql+asyncpg://postgres:test@localhost:5432/testdb\n          JWT_SECRET: test-secret-key-at-least-32-chars\n```\n\n### Production Checklist\n\n**P0 — Mandatory:**\n- [ ] All secrets from environment variables (SecretStr)\n- [ ] HTTPS enforced\n- [ ] CORS configured per environment\n- [ ] Rate limiting on auth endpoints\n- [ ] Input validation on all endpoints\n- [ ] Structured error responses (no stack traces)\n- [ ] Health + readiness endpoints\n- [ ] Database connection pooling\n- [ ] Migrations run before deploy\n- [ ] Structured logging (JSON)\n- [ ] Tests passing in CI\n\n**P1 — Recommended:**\n- [ ] OpenTelemetry tracing\n- [ ] Prometheus metrics endpoint\n- [ ] Background task queue (Celery/ARQ)\n- [ ] Redis caching layer\n- [ ] API versioning strategy\n- [ ] Request/response logging\n- [ ] Dependency security scanning\n- [ ] Performance benchmarks established\n\n## Phase 11: Advanced Patterns\n\n### Middleware Stack Order\n\n```python\n# Applied bottom-to-top (last added = first executed)\napp.add_middleware(GZipMiddleware, minimum_size=1000)    # 5. Compress\napp.add_middleware(CORSMiddleware, ...)                  # 4. CORS\napp.add_middleware(RequestIDMiddleware)                   # 3. Request ID\napp.add_middleware(RateLimitMiddleware)                   # 2. Rate limit\napp.add_middleware(TrustedHostMiddleware, allowed=[\"*\"])  # 1. Host check\n```\n\n### Pagination with Cursor-Based Option\n\n```python\nfrom fastapi import Query\n\nclass PaginationParams:\n    def __init__(\n        self,\n        page: int = Query(1, ge=1, description=\"Page number\"),\n        page_size: int = Query(20, ge=1, le=100, description=\"Items per page\"),\n    ):\n        self.offset = (page - 1) * page_size\n        self.limit = page_size\n        self.page = page\n        self.page_size = page_size\n\n@router.get(\"/users\", response_model=PaginatedResponse[UserResponse])\nasync def list_users(\n    pagination: PaginationParams = Depends(),\n    service: UserService = Depends(get_user_service),\n):\n    items, total = await service.list(\n        offset=pagination.offset, limit=pagination.limit\n    )\n    return PaginatedResponse(\n        items=items, total=total,\n        page=pagination.page, page_size=pagination.page_size,\n        has_next=(pagination.offset + pagination.limit) < total,\n    )\n```\n\n### WebSocket Pattern\n\n```python\nfrom fastapi import WebSocket, WebSocketDisconnect\n\nclass ConnectionManager:\n    def __init__(self):\n        self.connections: dict[str, WebSocket] = {}\n    \n    async def connect(self, user_id: str, ws: WebSocket):\n        await ws.accept()\n        self.connections[user_id] = ws\n    \n    def disconnect(self, user_id: str):\n        self.connections.pop(user_id, None)\n    \n    async def send(self, user_id: str, message: dict):\n        if ws := self.connections.get(user_id):\n            await ws.send_json(message)\n\nmanager = ConnectionManager()\n\n@router.websocket(\"/ws/{user_id}\")\nasync def websocket_endpoint(websocket: WebSocket, user_id: str):\n    await manager.connect(user_id, websocket)\n    try:\n        while True:\n            data = await websocket.receive_json()\n            # Process message\n    except WebSocketDisconnect:\n        manager.disconnect(user_id)\n```\n\n### File Upload Pattern\n\n```python\nfrom fastapi import UploadFile, File\n\n@router.post(\"/upload\")\nasync def upload_file(\n    file: UploadFile = File(..., description=\"File to upload\"),\n    user: User = Depends(get_current_user),\n):\n    # Validate\n    if file.size and file.size > 10 * 1024 * 1024:  # 10MB\n        raise ValidationError(\"File too large (max 10MB)\")\n    \n    allowed_types = {\"image/jpeg\", \"image/png\", \"application/pdf\"}\n    if file.content_type not in allowed_types:\n        raise ValidationError(f\"File type not allowed: {file.content_type}\")\n    \n    # Save\n    contents = await file.read()\n    path = f\"uploads/{user.id}/{file.filename}\"\n    # Save to S3/local storage...\n    \n    return {\"filename\": file.filename, \"size\": len(contents)}\n```\n\n## Phase 12: Common Mistakes\n\n| # | Mistake | Fix |\n|---|---------|-----|\n| 1 | Sync database calls in async app | Use async SQLAlchemy/databases |\n| 2 | Business logic in route handlers | Move to service layer |\n| 3 | No input validation | Pydantic models on every endpoint |\n| 4 | Returning ORM models directly | Use response schemas (from_attributes) |\n| 5 | Hardcoded config values | Pydantic Settings + env vars |\n| 6 | No error handling strategy | Custom exception hierarchy + global handler |\n| 7 | Missing health checks | /health + /ready endpoints |\n| 8 | `print()` for logging | structlog with JSON output |\n| 9 | No pagination on list endpoints | Default limit, max cap (100) |\n| 10 | Testing against production DB | Test fixtures with separate DB |\n\n## Quality Scoring (0–100)\n\n| Dimension | Weight | 0–25 | 50 | 75 | 100 |\n|-----------|--------|------|----|----|-----|\n| Type Safety | 15% | No types | Partial Pydantic | Full schemas | Strict mypy pass |\n| Error Handling | 15% | Bare HTTPException | Custom errors | Full hierarchy | + monitoring |\n| Testing | 15% | None | Happy path | 80%+ coverage | + contract tests |\n| Security | 15% | No auth | Basic JWT | + RBAC + rate limit | + scanning + audit |\n| Performance | 10% | Sync everything | Async DB | + caching | + profiling |\n| Observability | 10% | print() | Structured logs | + tracing | + metrics + alerts |\n| Database | 10% | Raw SQL | ORM + migrations | + repository pattern | + connection tuning |\n| Deployment | 10% | Manual | Dockerfile | + CI/CD | + health + rollback |\n\n**Scoring:** Your Score = Σ (dimension score × weight). **< 40 = critical, 40–60 = needs work, 60–80 = solid, 80+ = production-grade.**\n\n## 10 Commandments of FastAPI Production\n\n1. **Pydantic models at every boundary** — request, response, config\n2. **Async all the way down** — one sync call blocks the event loop\n3. **Services own business logic** — routers are thin wrappers\n4. **Dependency injection for testability** — `Depends()` is your best friend\n5. **Structured errors, structured logs** — JSON everything\n6. **Health checks are non-negotiable** — liveness + readiness\n7. **Test the sad paths** — 40% of tests should be error cases\n8. **Migrations before deployment** — never modify schema manually\n9. **Secrets in environment, never in code** — `SecretStr` enforces this\n10. **Profile before optimizing** — measure, don't guess\n\n## Natural Language Commands\n\n- `audit my FastAPI project` → Run health check, identify gaps\n- `set up a new FastAPI project` → Generate project structure + config\n- `add authentication to my API` → JWT + RBAC dependency pattern\n- `create a CRUD feature for [resource]` → Full router/service/repo/schemas\n- `optimize my database queries` → Connection pooling + async + N+1 prevention\n- `add structured logging` → Structlog + request ID middleware\n- `write tests for [feature]` → Async test patterns + fixtures\n- `prepare for production deployment` → Dockerfile + CI + checklist\n- `add caching to my API` → Redis caching pattern\n- `set up error handling` → Custom exception hierarchy + global handler\n- `add WebSocket support` → Connection manager pattern\n- `review my API security` → 10-point security checklist audit\n\n---\n\n⚡ **Level up your FastAPI APIs** → Get the [AfrexAI SaaS Context Pack ($47)](https://afrexai-cto.github.io/context-packs/) for complete SaaS architecture, pricing strategies, and go-to-market playbooks.\n\n🔗 **More free skills by AfrexAI:**\n- [afrexai-python-production](https://clawhub.com/skills/afrexai-python-production) — Python production engineering\n- [afrexai-api-architecture](https://clawhub.com/skills/afrexai-api-architecture) — API design methodology\n- [afrexai-database-engineering](https://clawhub.com/skills/afrexai-database-engineering) — Database patterns\n- [afrexai-test-automation-engineering](https://clawhub.com/skills/afrexai-test-automation-engineering) — Testing strategy\n- [afrexai-cicd-engineering](https://clawhub.com/skills/afrexai-cicd-engineering) — CI/CD pipeline design\n\n🛒 Browse all packs → [AfrexAI Storefront](https://afrexai-cto.github.io/context-packs/)\n","readmeExcerpt":"FastAPI Production Engineering Complete methodology for building, deploying, and scaling production FastAPI applications. Not a tutorial — a production operating system. Quick Health Check (/16) Score 2 points each. Total < 8 = critical work needed. | Signal | Healthy | Unhealthy | |--------|---------|-----------| | Type safety | Pydantic v2 models everywhere | dict returns, no validation | | Error handling | Structu","codeSnippets":[],"executableExamples":[{"language":"text","snippet":"src/\n├── app/\n│   ├── __init__.py\n│   ├── main.py              # App factory\n│   ├── config.py             # Pydantic Settings\n│   ├── dependencies.py       # Shared DI\n│   ├── middleware.py          # Custom middleware\n│   ├── features/\n│   │   ├── users/\n│   │   │   ├── __init__.py\n│   │   │   ├── router.py     # Endpoints\n│   │   │   ├── schemas.py    # Pydantic models\n│   │   │   ├── service.py    # Business logic\n│   │   │   ├── repository.py # Data access\n│   │   │   ├── models.py     # SQLAlchemy/SQLModel\n│   │   │   ├── dependencies.py\n│   │   │   └── exceptions.py\n│   │   ├── auth/\n│   │   ├── orders/\n│   │   └── ...\n│   ├── core/\n│   │   ├── database.py       # Engine, session factory\n│   │   ├── security.py       # JWT, hashing\n│   │   ├── errors.py         # Error hierarchy\n│   │   └── logging.py        # Structlog config\n│   └── shared/\n│       ├── pagination.py\n│       ├── filters.py\n│       └── responses.py\n├── migrations/               # Alembic\n├── tests/\n│   ├── conftest.py\n│   ├── unit/\n│   ├── integration/\n│   └── e2e/\n├── pyproject.toml\n├── Dockerfile\n└── docker-compose.yml"},{"language":"yaml","snippet":"# When to choose FastAPI over alternatives\nfastapi_is_best_when:\n  - \"You need auto-generated OpenAPI docs\"\n  - \"Team knows Python type hints\"\n  - \"API-first (no server-rendered HTML as primary)\"\n  - \"High concurrency with async I/O\"\n  - \"Microservice or API gateway\"\n\nconsider_alternatives:\n  django: \"Full-featured web app with admin, ORM, auth batteries\"\n  flask: \"Simple app, team prefers explicit over magic\"\n  litestar: \"Need WebSocket-heavy or more opinionated framework\"\n  hono_or_express: \"Team prefers TypeScript\""},{"language":"python","snippet":"from pydantic_settings import BaseSettings\nfrom pydantic import SecretStr, field_validator\nfrom functools import lru_cache\n\nclass Settings(BaseSettings):\n    # App\n    app_name: str = \"MyAPI\"\n    debug: bool = False\n    environment: str = \"production\"  # development | staging | production\n    \n    # Server\n    host: str = \"0.0.0.0\"\n    port: int = 8000\n    workers: int = 4\n    \n    # Database\n    database_url: SecretStr  # Required — no default\n    db_pool_size: int = 20\n    db_max_overflow: int = 10\n    db_pool_timeout: int = 30\n    \n    # Auth\n    jwt_secret: SecretStr  # Required\n    jwt_algorithm: str = \"HS256\"\n    jwt_expire_minutes: int = 30\n    \n    # Redis\n    redis_url: str = \"redis://localhost:6379/0\"\n    \n    # CORS\n    cors_origins: list[str] = [\"http://localhost:3000\"]\n    \n    @field_validator(\"environment\")\n    @classmethod\n    def validate_environment(cls, v: str) -> str:\n        allowed = {\"development\", \"staging\", \"production\"}\n        if v not in allowed:\n            raise ValueError(f\"environment must be one of {allowed}\")\n        return v\n    \n    model_config = {\"env_file\": \".env\", \"env_file_encoding\": \"utf-8\"}\n\n@lru_cache\ndef get_settings() -> Settings:\n    return Settings()"},{"language":"python","snippet":"from pydantic import BaseModel, Field, ConfigDict\nfrom datetime import datetime\nfrom uuid import UUID\n\n# Base with common config\nclass AppSchema(BaseModel):\n    model_config = ConfigDict(\n        from_attributes=True,      # ORM mode\n        str_strip_whitespace=True,  # Auto-strip\n        validate_default=True,      # Validate defaults too\n    )\n\n# Input schemas (what the API accepts)\nclass UserCreate(AppSchema):\n    email: str = Field(..., pattern=r\"^[\\w\\.-]+@[\\w\\.-]+\\.\\w+$\")\n    name: str = Field(..., min_length=1, max_length=100)\n    password: str = Field(..., min_length=8, max_length=128)\n\nclass UserUpdate(AppSchema):\n    name: str | None = Field(None, min_length=1, max_length=100)\n    email: str | None = Field(None, pattern=r\"^[\\w\\.-]+@[\\w\\.-]+\\.\\w+$\")\n\n# Output schemas (what the API returns)\nclass UserResponse(AppSchema):\n    id: UUID\n    email: str\n    name: str\n    created_at: datetime\n    # Note: password is NEVER in response schema\n\n# List response with pagination\nclass PaginatedResponse[T](AppSchema):\n    items: list[T]\n    total: int\n    page: int\n    page_size: int\n    has_next: bool"},{"language":"python","snippet":"from fastapi import Request\nfrom fastapi.responses import JSONResponse\nfrom starlette.status import (\n    HTTP_400_BAD_REQUEST, HTTP_401_UNAUTHORIZED,\n    HTTP_403_FORBIDDEN, HTTP_404_NOT_FOUND,\n    HTTP_409_CONFLICT, HTTP_422_UNPROCESSABLE_ENTITY,\n    HTTP_429_TOO_MANY_REQUESTS, HTTP_500_INTERNAL_SERVER_ERROR,\n)\n\nclass AppError(Exception):\n    \"\"\"Base application error.\"\"\"\n    def __init__(\n        self,\n        message: str,\n        code: str,\n        status_code: int = HTTP_500_INTERNAL_SERVER_ERROR,\n        details: dict | None = None,\n    ):\n        self.message = message\n        self.code = code\n        self.status_code = status_code\n        self.details = details or {}\n        super().__init__(message)\n\nclass NotFoundError(AppError):\n    def __init__(self, resource: str, identifier: str | int):\n        super().__init__(\n            message=f\"{resource} not found: {identifier}\",\n            code=\"NOT_FOUND\",\n            status_code=HTTP_404_NOT_FOUND,\n            details={\"resource\": resource, \"identifier\": str(identifier)},\n        )\n\nclass ConflictError(AppError):\n    def __init__(self, message: str, field: str | None = None):\n        super().__init__(\n            message=message, code=\"CONFLICT\",\n            status_code=HTTP_409_CONFLICT,\n            details={\"field\": field} if field else {},\n        )\n\nclass AuthenticationError(AppError):\n    def __init__(self, message: str = \"Invalid credentials\"):\n        super().__init__(message=message, code=\"UNAUTHORIZED\", status_code=HTTP_401_UNAUTHORIZED)\n\nclass AuthorizationError(AppError):\n    def __init__(self, message: str = \"Insufficient permissions\"):\n        super().__init__(message=message, code=\"FORBIDDEN\", status_code=HTTP_403_FORBIDDEN)\n\nclass ValidationError(AppError):\n    def __init__(self, message: str, errors: list[dict] | None = None):\n        super().__init__(\n            message=message, code=\"VALIDATION_ERROR\",\n            status_code=HTTP_422_UNPROCESSABLE_ENTITY,\n            details={\"errors\": e"},{"language":"python","snippet":"from fastapi import Depends, Security\nfrom fastapi.security import HTTPBearer, HTTPAuthorizationCredentials\nfrom jose import jwt, JWTError\nfrom datetime import datetime, timedelta, timezone\n\nsecurity = HTTPBearer()\n\ndef create_access_token(user_id: str, roles: list[str], settings: Settings) -> str:\n    expire = datetime.now(timezone.utc) + timedelta(minutes=settings.jwt_expire_minutes)\n    payload = {\n        \"sub\": user_id,\n        \"roles\": roles,\n        \"exp\": expire,\n        \"iat\": datetime.now(timezone.utc),\n    }\n    return jwt.encode(payload, settings.jwt_secret.get_secret_value(), algorithm=settings.jwt_algorithm)\n\nasync def get_current_user(\n    credentials: HTTPAuthorizationCredentials = Security(security),\n    settings: Settings = Depends(get_settings),\n    db: AsyncSession = Depends(get_db),\n) -> User:\n    try:\n        payload = jwt.decode(\n            credentials.credentials,\n            settings.jwt_secret.get_secret_value(),\n            algorithms=[settings.jwt_algorithm],\n        )\n        user_id = payload.get(\"sub\")\n        if not user_id:\n            raise AuthenticationError(\"Invalid token payload\")\n    except JWTError:\n        raise AuthenticationError(\"Invalid or expired token\")\n    \n    user = await db.get(User, user_id)\n    if not user:\n        raise AuthenticationError(\"User not found\")\n    return user\n\n# Role-based authorization\ndef require_role(*roles: str):\n    async def checker(user: User = Depends(get_current_user)) -> User:\n        if not any(r in user.roles for r in roles):\n            raise AuthorizationError(f\"Requires one of: {', '.join(roles)}\")\n        return user\n    return checker\n\n# Usage in router\n@router.get(\"/admin/users\")\nasync def list_users(\n    admin: User = Depends(require_role(\"admin\", \"superadmin\")),\n    service: UserService = Depends(get_user_service),\n):\n    return await service.list_all()"}],"parameters":{},"dependencies":[],"permissions":[],"extractedFiles":[],"languages":["typescript"],"docsSourceLabel":"CLAWHUB","editorialOverview":"FastAPI Production Engineering FastAPI Production Engineering Complete methodology for building, deploying, and scaling production FastAPI applications. Not a tutorial — a production operating system. Quick Health Check (/16) Score 2 points each. Total < 8 = critical work needed. | Signal | Healthy | Unhealthy | |--------|---------|-----------| | Type safety | Pydantic v2 models everywhere | dict returns, no validation | | Error handling | Structu","editorialQuality":{"score":100,"threshold":65,"status":"ready","wordCount":358,"uniquenessScore":69,"reasons":[]}},"media":{"evidence":{"source":"no-media","verified":false,"confidence":"low","updatedAt":"2026-04-15T00:45:39.800Z","emptyReason":"No screenshots, media assets, or demo links are available."},"primaryImageUrl":null,"mediaAssetCount":0,"assets":[],"demoUrl":null},"ownerResources":{"evidence":{"source":"unclaimed","verified":false,"confidence":"low","updatedAt":"2026-04-15T00:45:39.800Z","emptyReason":"This page has not been claimed by the agent owner."},"hasCustomPage":false,"customPageUpdatedAt":null,"customLinks":[],"structuredLinks":{"docsUrl":null,"demoUrl":null,"supportUrl":null,"pricingUrl":null,"statusUrl":null},"customPage":null},"relatedAgents":{"evidence":{"source":"protocol-neighbors","verified":false,"confidence":"medium","updatedAt":"2026-10-09T21:57:49.520Z","emptyReason":null},"items":[{"id":"8ebccd8e-3863-4187-8355-c3f14e1f9edf","entityType":"agent","canonicalPath":"/agent/iofficeai-aionui","slug":"iofficeai-aionui","name":"AionUi","description":"Free, local, open-source 24/7 Cowork app and OpenClaw for Gemini CLI, Claude Code, Codex, OpenCode, Qwen Code, Goose CLI, Auggie, and more | 🌟 Star if you like it!","url":"https://github.com/iOfficeAI/AionUi","homepage":"https://www.aionui.com","source":"GITHUB_REPOS","protocols":["MCP","OPENCLAW"],"capabilities":[],"safetyScore":100,"overallRank":70,"updatedAt":"2026-10-09T19:11:12.944Z","createdAt":"2026-02-25T03:38:16.584Z","downloads":null},{"id":"b917f68a-ebff-438e-84f8-3f4b2494c0bc","entityType":"agent","canonicalPath":"/agent/activepieces-activepieces","slug":"activepieces-activepieces","name":"activepieces","description":"AI Agents & MCPs & AI Workflow Automation • (~400 MCP servers for AI agents) • AI Automation / AI Agent with MCPs • AI Workflows & AI Agents • MCPs for AI Agents","url":"https://github.com/activepieces/activepieces","homepage":"https://www.activepieces.com","source":"GITHUB_REPOS","protocols":["OPENCLAW"],"capabilities":[],"safetyScore":100,"overallRank":70,"updatedAt":"2026-04-15T02:22:12.426Z","createdAt":"2026-02-25T03:38:12.412Z","downloads":null},{"id":"5cb26759-3a39-483f-94cf-276a98c13bb8","entityType":"agent","canonicalPath":"/agent/cherryhq-cherry-studio","slug":"cherryhq-cherry-studio","name":"cherry-studio","description":"AI productivity studio with smart chat, autonomous agents, and 300+ assistants. Unified access to frontier LLMs","url":"https://github.com/CherryHQ/cherry-studio","homepage":"https://cherry-ai.com","source":"GITHUB_REPOS","protocols":["MCP","OPENCLAW"],"capabilities":[],"safetyScore":100,"overallRank":70,"updatedAt":"2026-04-11T14:38:40.986Z","createdAt":"2026-02-25T03:38:19.379Z","downloads":null},{"id":"6f6582d0-5d76-4f0f-b81d-86520247950b","entityType":"agent","canonicalPath":"/agent/copilotkit-copilotkit","slug":"copilotkit-copilotkit","name":"CopilotKit","description":"The Frontend for Agents & Generative UI. React + Angular","url":"https://github.com/CopilotKit/CopilotKit","homepage":"https://docs.copilotkit.ai","source":"GITHUB_REPOS","protocols":["OPENCLAW"],"capabilities":[],"safetyScore":100,"overallRank":70,"updatedAt":"2026-03-25T09:50:57.846Z","createdAt":"2026-02-25T03:39:14.617Z","downloads":null}],"links":{"hub":"/agent","source":"/agent/source/clawhub","protocols":[{"label":"OpenClaw","href":"/agent/protocol/openclew"}]}}}