In this repository

• Scope: FastAPI app lives in backend/. Python also in services/; same style applies. After edits run make format and make lint from repo root.
• Rules: .cursor/rules/fastapi-python-best-practices.md, .cursor/rules/fastapi-patterns.md, .cursor/rules/standards.md, .cursor/rules/project.md.
• Full context: .cursor/skills/README.md.

FastAPI best practices (team standards)

Standards derived from this project’s implementation. Follow these so the codebase stays consistent and maintainable.

1. Project structure and layers

• Endpoints (controllers): backend/app/api/v1/endpoints/ — one file per domain (e.g. data.py, database.py, metrics.py). Only HTTP concerns: parse request, call service/repo via Depends, return response or raise HTTPException. Use command service for write operations (POST/PUT/PATCH/DELETE) and query service for read operations (GET). See .cursor/rules/cqrs.md.
• Business logic: backend/app/services/ — use command services for writes and query services for reads (CQRS). Services receive repositories via constructor; use AsyncSession for DB when in the API path. No raw request/response objects.
• Data access: backend/app/repositories/ — use write repositories (create/update/delete) and read repositories (get/list) per domain. Repositories receive AsyncSession; perform queries and commits. No business rules.
• Shared dependencies: backend/app/api/v1/deps.py — define get_* functions that return service or repository instances (injecting get_async_session). Use these in route handlers via Depends(...).
• Models: backend/app/models/database.py (SQLModel table models), backend/app/models/schemas.py (Pydantic request/response). DB name is fastapi_db.

Do not: Put SQL or business logic in endpoint files; put HTTP logic in services or repositories.

2. Routers and API design

• Use APIRouter with a prefix and tags for OpenAPI grouping:

  python
  1router = APIRouter(prefix="/data", tags=["data"])

• Mount routers in backend/app/api/v1/__init__.py on the v1 router (prefix="/api/v1").
• Prefer explicit response_model and status_code on route decorators:

  python
  1@router.post("/process", response_model=TaskResult, status_code=200)
  2async def process_data(request: DataRequest, service: DataService = Depends(get_data_service)) -> TaskResult:

• Use async def for handlers that do I/O (DB, external calls). Use def only for handlers that do no I/O (e.g. simple metrics that read in-memory state).
• Document with short docstrings on each route; they appear in OpenAPI.

3. Dependencies (Depends)

• Session: Use get_async_session from app.core.database for any route or dependency that touches the DB. Never create a raw engine or session in an endpoint.
• Services and repositories: Define factory functions in app.api.v1.deps that depend on get_async_session and return a service or repository instance. Use them in endpoints via Depends(get_data_service), Depends(get_task_repository), etc.

Example (from deps.py):

python
1def get_data_service(session: AsyncSession = Depends(get_async_session)) -> DataService:
2    """Dependency that returns a DataService instance (async session)."""
3    return DataService(session)

• In endpoints, inject the service (or repository when the endpoint only needs data access) and call methods; do not instantiate services or repos inside the route.

4. Request and response models

• Request bodies: Use Pydantic models from app.models.schemas (e.g. DataRequest). Declare them as the route parameter; FastAPI validates and parses.
• Responses: Use Pydantic response models (e.g. TaskResult, RecordResponse, TaskLogResponse) and set response_model= on the decorator. Never return a SQLModel/ORM instance directly; map to a schema and return that (e.g. RecordResponse.model_validate(r) or RecordResponse(**record)).
• Prefer model_validate(orm_instance) (Pydantic v2) when building response models from ORM objects.

5. Async in the request path

• Use async def for route handlers that perform I/O (DB, HTTP, etc.). Use AsyncSession and async repository methods so the event loop is not blocked.
• If you must call a blocking operation (e.g. sync RabbitMQ RPC), run it in a thread pool with await asyncio.to_thread(blocking_func, ...) and keep the rest of the flow async (e.g. DB write). See DataService.process_data_sync_and_save in app.services.data_service.
• Sync engine/session are only for Celery tasks and startup (e.g. init_db). Do not use them in request handlers.

6. Error handling

• API boundary: Use HTTPException with the appropriate status code (400, 404, 503, etc.) and a clear detail. Do not let uncaught exceptions bubble to the client without mapping.
• Services/repositories: Raise domain-friendly exceptions or return None/result; let the endpoint translate to HTTP (e.g. if not record: raise HTTPException(status_code=404, detail="Record not found")).
• External calls: When a dependency (e.g. call_service_via_rabbitmq) raises ValueError, catch it in the endpoint and map to HTTPException(status_code=503, detail=str(e)) (or another suitable code).
• Handle errors and edge cases first (early returns, guard clauses); keep the happy path last. See .cursor/rules/error-handling.md.

7. Middleware and app setup

• CORS: Configure in main.py (e.g. CORSMiddleware). Keep origins/methods/headers explicit for production.
• Idempotency: Optional Idempotency-Key header on POST endpoints that support it; implemented in middleware (IdempotencyMiddleware). Do not duplicate idempotency logic in services. Paths are listed in the middleware (e.g. /api/v1/data/process, /api/v1/data/process-async).
• Startup: For simple one-off init (e.g. init_db()), @app.on_event("startup") is acceptable. For multi-step startup/shutdown, prefer a lifespan context manager.

8. Database (PostgreSQL, SQLModel, Alembic)

• In routes: Always use async session via get_async_session and pass it to services/repositories. Use connection pooling (already configured in app.core.database).
• Repositories: Accept AsyncSession in the constructor; use await session.commit(), await session.refresh(record) after writes; use select(Model).where(...) and await session.execute() / session.get() for reads. Avoid N+1; batch or eager load when needed.
• Migrations: Add or change models in app.models.database; then from backend/ run alembic revision --autogenerate -m "description" and alembic upgrade head. Do not change the database name from fastapi_db.

9. Naming and style

• Files and modules: Lowercase with underscores (e.g. data_service.py, task_repository.py, data.py for the data router).
• Routers: Name the router variable router. Group related routes in one file (e.g. all /data/* in data.py).
• Schemas: Clear names for request (e.g. DataRequest) and response (e.g. TaskResult, RecordResponse). Use type hints and optional fields where appropriate.
• Docstrings: Short module docstring and route docstrings so OpenAPI and readers understand purpose. Document parameters and return values in Pydantic models if non-obvious.

10. Team checklist (before merging)

• New routes live under backend/app/api/v1/endpoints/ and are mounted in api/v1/__init__.py.
• Handlers use Depends(get_async_session) or Depends(get_*_service) / Depends(get_*_repository) from deps.py; no ad-hoc session or service creation in routes.
• Request/response use Pydantic models from schemas.py; responses use response_model= and never return raw ORM instances.
• I/O in the request path is async (AsyncSession, async repo methods); any blocking call is wrapped in asyncio.to_thread.
• Errors are mapped to HTTPException in endpoints; services/repos do not raise HTTP-specific errors.
• New or changed DB models have an Alembic migration; DB name remains fastapi_db.
• make format and make lint have been run from repo root and pass.

Quick reference (this project)

For more detail, see .cursor/rules/fastapi-python-best-practices.md and .cursor/rules/standards.md.