Skip to content

API Refactor Plan — ArjanCodes Best Practices

Archived: fully implemented as of 2026-05-05. See PRs #370 (Phase 1), #375 (Phase 2), #377 (Phase 3), #379 (Phase 4), #380 (Phase 5), #381 (Phase 6).

Executive Summary

The portfolio FastAPI backend has accreted into a 2,404-line god module (api/main.py) holding configuration, response schemas, ~1,000 lines of hardcoded seed data, every GET handler, inline HTML email templates, subprocess CV orchestration, and the OpenAPI customizer. Routers admin.py and uigen.py work around the resulting circular import by doing from api.main import settings inside function bodies. Every read endpoint repeats the same if AsyncSessionLocal is None: return STATIC_LIST branch and hand-rolled _orm_to_pydantic conversion. Seed content lives in two places that can drift (Python lists in main.py and 1,103 lines of SQL inserts in alembic migration 0002).

This refactor restructures the backend into a layered FastAPI application — config / schemas / repositories / services / routers — using ArjanCodes' patterns from his examples repo (notably 2024/build_deploy_host_backend/skypulse, 2024/repository, 2026/ports, and 2026/clean). After the refactor:

  • api/main.py shrinks to a thin app factory (~150 lines) that registers routers, middleware, and exception handlers.
  • Pydantic v2 from_attributes=True schemas replace all _x_to_y helpers.
  • Per-domain routers and repositories eliminate cross-cutting duplication.
  • One seed.json is the single source of truth for static content; alembic and the fallback path both read from it.
  • Domain exceptions are translated to HTTP at the edge — handlers focus on the happy path.
  • Email and CV generation become async services (no blocking subprocess on the event loop).

The plan is ordered so each phase is a safe, independently shippable PR. Phase 1 introduces zero behavior change — only file layout. Phases 2–4 progressively strip the god module. Phases 5–6 fix deeper architectural issues (seed-data drift, blocking subprocess, in-process token store, structured logging).

What We're NOT Doing (avoid over-engineering)

Per ArjanCodes' 2025/anti/ cautions and the project's actual scale (single repo, single team, ~9 entities):

  1. No Repository[T] ABC with generics. Use plain async def functions per aggregate. Generic base classes only pay off with 3+ implementations of the same interface — we have one (PostgreSQL via SQLAlchemy).
  2. No DI container (no dependency-injector, punq, kink, etc.). FastAPI's Depends covers every injection need we have.
  3. No CQRS, no event bus, no domain events. This is a content-serving API, not an aggregate-with-invariants domain.
  4. No hexagonal "ports & adapters" full split. ArjanCodes' 2026/ports example is instructive but heavyweight; we adopt the idea (translate domain exceptions at the edge) without creating a separate domain/ package and adapter wiring layer.
  5. No premature service interfaces. EmailService and CvService are concrete classes/modules. We do not create IEmailService Protocols until there's a second implementation.
  6. No DTO-per-layer ceremony. ORM models flow into Pydantic response schemas via model_validate(orm_obj). We don't introduce a third "domain entity" layer.
  7. No rewrite of api/feature_flags.py, api/db/engine.py, or api/db/models.py. Per the brief, these work — we extend models.py (Phase 5 timeline split) but don't restructure them.
  8. No swap of slowapi, no swap of Alembic, no SQLModel migration. Stay on SQLAlchemy 2.0 Mapped[...].
  9. No abstraction over the DB-or-fallback pattern into a base class. A single fallback per repository function handles it cleanly without parameterized strategies.
  10. No HTML-template engine for OpenAPI customization — keep the existing custom hook, just relocate it.

Phase 1: Foundation — File Layout, No Behavior Change

Goal: break the circular import, give every concept a home, ship with zero functional diff. This phase alone unblocks everything downstream.

1.1 Extract Settings to api/core/config.py

Problem. Settings lives in api/main.py, so routers that need it (admin.py, uigen.py) trigger a circular import and resort to from api.main import settings inside function bodies.

ArjanCodes pattern. 2024/build_deploy_host_backend/skypulse/app/core/config.py — Settings class derived from pydantic-settings, single settings = Settings() module-level singleton imported anywhere.

Files to create.

  • api/core/__init__.py
  • api/core/config.py

Files to modify.

  • api/main.py — replace inline Settings with from api.core.config import settings.
  • api/routers/admin.py — replace from api.main import settings (inside functions) with module-level from api.core.config import settings.
  • api/routers/uigen.py — same.

Steps.

  1. Copy the Settings class verbatim from main.py into api/core/config.py. Keep the same field names, env var bindings, and defaults.
  2. Export settings = Settings() at module bottom.
  3. In main.py, delete the class and replace settings = Settings() with the import.
  4. In each router, hoist the import to the top of the file and remove the inside-function workaround.
  5. Run the API locally and pytest api/ — no diff in behavior.

Acceptance. grep -r "from api.main import settings" api/ returns nothing.

1.2 Carve out package skeleton

Problem. There's no place to put services, repositories, response schemas, exception types.

Files to create (empty __init__.py + placeholder modules).

  • api/repositories/__init__.py
  • api/services/__init__.py
  • api/schemas/responses.py (placeholder, populated in Phase 2)
  • api/core/exceptions.py (placeholder, populated in Phase 3)
  • api/core/logging.py (placeholder, populated in Phase 6)
  • api/core/openapi.py (placeholder, populated in 1.4)
  • api/data/ directory (populated in Phase 5)

Steps. Create empty modules with docstring stubs only. No imports yet. This is purely scaffolding so subsequent phases land cleanly.

1.3 Move OpenAPI customizer to api/core/openapi.py

Problem. ~50–80 lines of custom OpenAPI hook bloat main.py.

ArjanCodes pattern. 2024/build_deploy_host_backend/skypulse/app/main.py keeps the FastAPI factory minimal; cross-cutting setup lives in app/core/.

Files to create. api/core/openapi.py exposing def custom_openapi(app: FastAPI) -> Callable[[], dict].

Files to modify. api/main.py — app.openapi = custom_openapi(app).

Acceptance. /openapi.json byte-identical to before.

1.4 Move inline HTML email templates to files

Problem. Two ~130-line HTML strings live in route handlers.

Files to create.

  • api/templates/__init__.py
  • api/templates/contact_admin.html
  • api/templates/contact_user.html

Steps.

  1. Cut each HTML block into its own .html file. Replace string interpolation with Jinja2 placeholders ({{ name }}, {{ message }}).
  2. In main.py (still — service extraction comes in Phase 4), load via jinja2.Environment(loader=PackageLoader("api", "templates")). Render into the existing send-email call.
  3. Add jinja2 to requirements.txt if not present.

Acceptance. Sent test email visually identical to before.

Phase 2: Response Schemas with from_attributes=True

Goal: kill all _x_to_y hand-converters; let Pydantic v2 do the work.

2.1 Define response schemas in api/schemas/responses.py

Problem. Every route does _project_to_card(orm_project) etc. — 200+ lines of manual field copying. ORM and response shapes drift silently.

ArjanCodes pattern. 2024/build_deploy_host_backend/skypulse/app/schemas/city.py — Pydantic v2 model with model_config = ConfigDict(from_attributes=True, populate_by_name=True) and Field(alias="camelCase") for JSON field names while attributes stay snake_case.

Files to create. api/schemas/responses.py containing:

  • ProjectCard, ProjectDetail, ProjectPhase
  • WorkEntry, Skill, SkillCategory
  • Award, Certification
  • ContactResponse, HealthResponse, FeatureFlagsResponse
  • A shared BaseResponse with model_config = ConfigDict(from_attributes=True, populate_by_name=True, alias_generator=to_camel).

Files to modify. api/main.py — delete each old response model class as it's superseded. Replace _project_to_card(p) with ProjectCard.model_validate(p). Replace _project_to_detail(p) with ProjectDetail.model_validate(p). Same for work, skills, awards, certifications.

Steps.

  1. Create BaseResponse with shared config in api/schemas/responses.py.
  2. For each ORM model in api/db/models.py, define its response schema mirroring exposed fields. Use Field(alias=...) only where the JSON key differs from the attribute.
  3. For nested data (project phases, skills inside categories), use list types — Pydantic walks relationship() collections automatically when from_attributes=True.
  4. One domain at a time: replace handlers' _x_to_y(obj) calls with Schema.model_validate(obj). Run tests after each.
  5. Delete the now-unused converter helpers.

Acceptance. grep -n "_project_to_\|_work_entry_to_\|_skill_to_" api/main.py returns nothing. Response payloads byte-identical (verify via snapshot test on /api/projects etc.).

2.2 Keep request schemas where they are

api/schemas/projects.py, experience.py, skills.py already hold the create/update DTOs. Leave them. Optionally move them next to the response schemas for symmetry — but only if free; not load-bearing.

Phase 3: Per-Domain Routers + Domain Exceptions

Goal: dissolve the god module's route section into one router per aggregate; centralize error → HTTP translation.

3.1 Domain exception hierarchy in api/core/exceptions.py

ArjanCodes pattern. 2026/ports/domain/errors.py (defines DomainError, NotFound, Conflict as plain exceptions) + 2026/ports/api.py (FastAPI app.add_exception_handler(NotFound, ...) translates at the edge). Also 2024/build_deploy_host_backend/skypulse/app/exceptions/ exceptions.py for the factory pattern (create_exception_handler(status_code, message)).

Files to create. api/core/exceptions.py:

python
class PortfolioError(Exception): ...
class NotFoundError(PortfolioError): ...
class ConflictError(PortfolioError): ...
class ValidationError(PortfolioError): ...
class DatabaseUnavailableError(PortfolioError): ...

Plus a register_exception_handlers(app: FastAPI) -> None function using a create_exception_handler(status_code) factory.

Files to modify. api/main.py — call register_exception_handlers(app) once during app construction. Remove ad-hoc HTTPException(404, ...) from handlers as they're refactored (repositories raise NotFoundError, handlers don't catch).

Acceptance. A repository raising NotFoundError("project", slug) results in a 404 response with consistent JSON shape.

3.2 Per-domain GET routers

ArjanCodes pattern. 2024/build_deploy_host_backend/skypulse/app/api/routes/cities.py — one router file per aggregate, each declaring its own APIRouter(prefix=..., tags=[...]). 2023/fastapi-router/api/routers/items.py for the basic shape.

Files to create.

  • api/routers/projects.py — GET /api/projects, GET /api/projects/{slug}
  • api/routers/experience.py — GET /api/experience
  • api/routers/skills.py — GET /api/skills
  • api/routers/awards.py — GET /api/awards
  • api/routers/certifications.py — GET /api/certifications
  • api/routers/contact.py — POST /api/contact
  • api/routers/cv.py — CV download/status/generate routes
  • api/routers/health.py — GET /health, GET /api/health
  • api/routers/feature_flags.py — GET /api/feature-flags, PATCH /api/feature-flags

Files to modify. api/main.py — replace inline route handlers with app.include_router(projects.router) etc. Order routers under a def create_app() -> FastAPI factory.

Steps (one router at a time).

  1. Move the handler function and decorator into the new router file.
  2. Update imports: settings from api.core.config, response schemas from api.schemas.responses, DB dep from api.db.engine.
  3. Keep handlers thin: param parse → call repository (Phase 4) or, transitionally, the existing inline DB query → return Schema.model_validate(...).
  4. Register the router in main.py.
  5. Delete the original inline handler and any helpers used only by it.
  6. Run tests; ship; move to next router.

Acceptance. wc -l api/main.py drops below 600 by end of phase. Each router file is under ~150 lines.

3.3 Convert main.py into an app factory

Files to modify. api/main.py should end up as roughly:

python
def create_app() -> FastAPI:
    app = FastAPI(...)
    register_exception_handlers(app)
    configure_logging()           # Phase 6
    configure_middleware(app)     # CORS, slowapi, request-id (Phase 6)
    register_routers(app)
    app.openapi = custom_openapi(app)
    return app

app = create_app()

Acceptance. Module-level side effects limited to app = create_app(). Easy to instantiate a second app for testing.

Phase 4: Repository + Service Layers

Goal: handlers stop touching AsyncSession directly. Cross-cutting "DB-or-fallback" logic lives in one place.

4.1 Two DB dependencies

Problem. get_db() returns Optional[AsyncSession] and every handler null-checks.

ArjanCodes pattern. 2026/ports/api.py — explicit dependency providers; the handler declares what it needs and the DI layer enforces it.

Files to modify. api/db/engine.py — add two explicit dependencies:

  • async def get_db() -> AsyncIterator[AsyncSession] — raises DatabaseUnavailableError if engine not configured. Used by write endpoints (admin, contact-persistence) and reads with no static fallback.
  • async def get_optional_db() -> AsyncIterator[AsyncSession | None] — current behavior; used by read endpoints that have a static fallback.

Acceptance. Admin routes no longer need to null-check; they get a 503 automatically via the exception handler if the DB is down.

4.2 Repository functions per aggregate

ArjanCodes pattern. 2024/build_deploy_host_backend/skypulse/app/crud/cities.py — module of free async def functions that take db: AsyncSession and typed args, return ORM rows or raise domain errors. NOT 2024/repository/repository.py's ABC (see anti-pattern note above).

Files to create.

  • api/repositories/projects.py — list_cards(db), get_by_slug(db, slug), slugs_ordered(db), create(db, payload), update(db, slug, payload), delete(db, slug), reorder(db, ordered_slugs).
  • api/repositories/experience.py — list_entries(db), get_by_id(db, id), plus admin CRUD.
  • api/repositories/skills.py — list_categories(db), plus admin CRUD for skills and categories.
  • api/repositories/awards.py, api/repositories/certifications.py — list + admin CRUD.

Conventions.

  • Read functions return ORM models (Pydantic schemas validate at the router edge).
  • "Get one" functions raise NotFoundError(entity, key) instead of returning None — handler doesn't need to translate.
  • Write functions raise ConflictError on unique-constraint violations (catch IntegrityError, re-raise as domain error).
  • No transaction management inside repos; get_db() wraps each request in async with db.begin().

Files to modify. Each router file calls the repository instead of inlined select(...) statements. Handler shape becomes:

python
@router.get("/{slug}", response_model=ProjectDetail)
async def get_project(slug: str, db: AsyncSession = Depends(get_db)) -> Project:
    return await projects_repo.get_by_slug(db, slug)

Acceptance. grep -n "select(" api/routers/ returns nothing. All ORM queries live under api/repositories/.

4.3 Centralize the DB-or-fallback pattern

Problem. Read endpoints branch: if AsyncSessionLocal is None: return STATIC_LIST.

Decision. Push fallback into the repository, not into a service wrapper. Each read repo function accepts db: AsyncSession | None and falls back to seed.load_*() (Phase 5) when None. Single place per domain.

python
async def list_cards(db: AsyncSession | None) -> list[Project]:
    if db is None:
        return seed.load_projects()   # returns dicts; Pydantic handles both via from_attributes
    result = await db.execute(select(Project).order_by(...))
    return list(result.scalars())

Note. Pydantic schemas with from_attributes=True validate both ORM rows and plain dicts — no special case at the router edge.

Acceptance. No AsyncSessionLocal is None checks remain in router code.

4.4 Email service in api/services/email.py

Problem. Email send + Jinja2 rendering currently lives in the contact route handler.

ArjanCodes pattern. 2025/di/manual.py (DataPipeline composes injected pieces) and 2026/clean/report_after.py (free functions composing a pipeline).

Files to create. api/services/email.py:

  • class EmailService with constructor (settings: Settings, jinja_env: Environment) and async methods send_admin_notification(...), send_user_confirmation(...).
  • Module-level def get_email_service() -> EmailService for use as a FastAPI dependency.

Files to modify. api/routers/contact.py — handler injects EmailService via Depends(get_email_service) and calls it.

Acceptance. Contact handler under 30 lines.

4.5 CV generation service in api/services/cv.py — async subprocess

Problem. _run_cv_generation calls blocking subprocess.run(..., timeout=120) inside an async handler. Any concurrent CV request stalls the event loop.

Files to create. api/services/cv.py:

  • async def generate_cv(...) -> Path using asyncio.create_subprocess_exec(...) with await proc.communicate() and asyncio.wait_for(..., timeout=120).
  • Helpers for input validation and output path resolution.

Files to modify. api/routers/cv.py — handler awaits the service.

Acceptance. Two simultaneous POST /api/generate-cv requests don't block each other (verified with httpx.AsyncClient parallel calls in tests).

Phase 5: Single Source of Truth for Seed Content + Schema Cleanup

Goal: kill the dual-source drift between main.py static lists and the 1,103-line alembic seed migration. Replace the timeline_encoded array hack with a proper child table.

5.1 Single seed file

Files to create.

  • api/data/__init__.py
  • api/data/seed.json — canonical content for projects, work history, skills, awards, certifications. Structure mirrors the response schemas (snake_case keys to match ORM attributes; Pydantic handles aliasing).
  • api/data/seed.py — loader: def load_projects() -> list[dict], load_work_entries(), etc. Caches parsed JSON in module-level dicts after first read.

Files to modify.

  • Delete the hardcoded PROJECT_CARDS, PROJECT_DETAILS, WORK_HISTORY, SKILL_CATEGORIES, CERTIFICATIONS, AWARDS constants from main.py (largely gone already after Phase 4.3 if fallbacks point at seed.load_*).
  • Repository fallback paths import from api.data.seed.

Steps.

  1. Generate seed.json once: write a one-shot local script that dumps the existing Python lists from the old main.py to JSON. Commit only the JSON.
  2. Implement seed.py loader (lazy-cached).
  3. Wire fallback paths (Phase 4.3) to the loader.
  4. Verify: GET /api/projects with DB stopped returns the same payload as before.

5.2 Refactor alembic seed migration to read from seed.json

Files to modify. alembic/versions/0002_seed_initial_data.py — replace 1,103 lines of inline SQL with code that:

  1. Reads api/data/seed.json (via pathlib.Path(__file__).parents[2] / "data" / "seed.json").
  2. Issues bulk inserts via op.bulk_insert(...) per table.

Important — migration immutability policy. If the chosen policy is "migrations are immutable," do NOT modify 0002. Instead leave it as-is and create a new migration 0005_reseed_from_json.py that deletes and re-inserts from the JSON whenever content changes. Document the chosen policy in the ADR before proceeding.

Idempotency. Whichever approach is taken, use ON CONFLICT DO NOTHING on natural keys so rerunning against an already-seeded DB is a no-op.

Acceptance. Editing seed content means editing one JSON file; both DB seed and the no-DB fallback reflect the change immediately.

5.3 Replace timeline_encoded with a child table

Problem. Project phases are stored as ARRAY(Text) of "name|||description" strings.

Files to modify.

  • api/db/models.py — add class ProjectPhase with (id, project_id FK, ordinal, name, description) and relationship from Project.phases. Drop timeline_encoded from Project.
  • New migration alembic/versions/0005_project_phases_table.py — create table, copy data from timeline_encoded (parse on |||), drop column.
  • api/repositories/projects.py — load phases via relationship (selectinload(Project.phases)).
  • api/schemas/responses.py — ProjectDetail.phases: list[ProjectPhase] already in place from Phase 2; field stays the same shape.

Acceptance. Round-trip migration on a populated DB preserves phase order and content. timeline_encoded column gone.

Phase 6: Cross-Cutting Concerns — Logging, Request IDs, External Token Store

Goal: production-readiness fixes that don't fit a domain.

6.1 Structured logging + request ID middleware

ArjanCodes pattern. 2024/build_deploy_host_backend/skypulse/app/core/logging.py — loguru with InterceptHandler to capture stdlib + uvicorn logs into one structured stream.

Files to create.

  • api/core/logging.py — configure_logging() sets up loguru + InterceptHandler; JSON sink in production, pretty sink in dev (gated on settings.environment).
  • api/core/middleware.py — RequestIdMiddleware reads/generates X-Request-Id, binds it to a contextvars.ContextVar, attaches to response header. Loguru filter pulls it into every log line.

Files to modify. api/main.py create_app() calls configure_logging() and app.add_middleware(RequestIdMiddleware).

Add to requirements.txt. loguru.

Acceptance. Every log line includes request_id field. X-Request-Id round-trips on responses.

6.2 Move UIGen guest tokens to Redis (or DB-backed table)

Problem. _guest_tokens: dict is in-process — multi-replica Container Apps lose tokens across instances.

Decision. Choose the lighter option given existing infra:

  • Option A (preferred if Redis already provisioned): add redis.asyncio client; tokens get a TTL-bound key.
  • Option B (no new infra): add a uigen_tokens table (id, token_hash, issued_at, expires_at, revoked_at) and clean up via a periodic job or lazy expiry on read.

Files to create. api/services/uigen_tokens.py with issue(), validate(), revoke(). Backend (Redis vs. DB) is a single import boundary.

Files to modify. api/routers/uigen.py calls the service. The in-process dict is deleted.

Acceptance. Guest token issued by replica A validates against replica B.

6.3 Verify scale-safety of slowapi

slowapi's default in-memory storage has the same multi-replica issue. If multi-replica is in scope, point it at the Redis from 6.2 (or defer — slowapi is correctness-non-critical). Document the decision; do not silently leave it broken.

Phase Summary & Sequencing Rationale

Phase Risk Behavior change Unlocks
1. Foundation Very low None Removes circular import; gives every concept a home
2. Response schemas Low None (snapshot-tested) Kills hand-rolled converters
3. Routers + exceptions Low None Shrinks main.py; centralizes error handling
4. Repos + services Medium Equivalent Removes duplicated DB-or-fallback; async CV gen
5. Seed + schema Medium DB schema migration Eliminates dual-source drift; clean phases table
6. Logging + token store Low–Medium Token storage moves out-of-process Multi-replica safety

Each phase should be one PR (or a small handful), independently revertable. Do not start Phase N+1 until Phase N is merged and green in CI.

Critical Files for Implementation

File Role
api/main.py Shrinks from 2,404 lines to ~150-line app factory
api/core/config.py (new) Single Settings home; breaks circular import
api/core/exceptions.py (new) Domain exception hierarchy
api/schemas/responses.py (new) Replaces all _x_to_y hand-converters
api/repositories/projects.py (new) Template for all aggregate repos
api/data/seed.json (new) Single source of truth for static content
api/db/models.py Extended in Phase 5 with ProjectPhase child table
api/services/email.py (new) Extracted email + Jinja2 service
api/services/cv.py (new) Async subprocess CV generation

ArjanCodes Reference Map

This refactor ArjanCodes example
api/core/config.py 2024/build_deploy_host_backend/skypulse/app/core/config.py
Response schemas with from_attributes 2024/build_deploy_host_backend/skypulse/app/schemas/city.py
Per-domain routers 2024/build_deploy_host_backend/skypulse/app/api/routes/cities.py
Repository functions (function-style) 2024/build_deploy_host_backend/skypulse/app/crud/cities.py
Domain exceptions + HTTP translation 2026/ports/domain/errors.py + 2026/ports/api.py
Exception handler factory 2024/build_deploy_host_backend/skypulse/app/exceptions/exceptions.py
EmailService / CvService 2025/di/manual.py, 2026/clean/report_after.py
create_app() factory pattern 2024/build_deploy_host_backend/skypulse/app/main.py
Frozen dataclass config objects 2026/clean/report_after.py (TrainingConfig)
Two DB deps (required vs. optional) 2026/ports/api.py (explicit dependency declaration)
Structured logging + InterceptHandler 2024/build_deploy_host_backend/skypulse/app/core/logging.py
Anti-patterns to avoid 2025/anti/ (abstraction, static_methods, global_coupling, overengineering)