Backend Architecture¶

Overview¶

The backend is a FastAPI application providing email services and API endpoints.

Technology Stack¶

• Framework: FastAPI
• Language: Python 3.11
• ASGI Server: Uvicorn
• Validation: Pydantic v2 + pydantic-settings
• Email: SMTP (Gmail) via Jinja2-rendered HTML templates
• Rate Limiting: slowapi (in-memory, keyed by IP; single-replica only)
• Database ORM: SQLAlchemy 2 (async), asyncpg driver
• Migrations: Alembic
• Logging: loguru (structured JSON in production, pretty in dev)

Project Structure¶

The backend follows an ArjanCodes-inspired layered architecture (PRs #370, #375, #377, #379, #380, #381): core → schemas → repositories → services → routers. main.py is now a ~150-line create_app() factory that registers routers, middleware, and exception handlers. See Architecture Pattern below for the design rationale.

api/
├── main.py              # ~150-line create_app() factory
├── feature_flags.py     # Azure App Config feature flag cache (30 s TTL)
├── requirements.txt     # Python dependencies
├── __init__.py
├── core/
│   ├── config.py        # Settings (pydantic-settings); `settings` singleton imported everywhere
│   ├── exceptions.py    # PortfolioError, NotFoundError, ConflictError, ServiceUnavailableError
│   │                    # + register_exception_handlers()
│   ├── limiter.py       # Shared slowapi Limiter instance (in-memory, single-replica)
│   ├── logging.py       # configure_logging() — loguru + InterceptHandler for stdlib/uvicorn
│   ├── middleware.py    # RequestIdMiddleware — reads/generates X-Request-Id header
│   └── openapi.py       # custom_openapi() — x-uigen-id / x-uigen-view annotations
├── db/
│   ├── engine.py        # Async engine + get_db() + get_optional_db()
│   └── models.py        # Project, ProjectPhase, WorkEntry, SkillCategory, Skill,
│                        # Award, Certification, UigenToken
├── routers/
│   ├── admin.py         # Write endpoints (bearer-token protected), mounted at /api
│   ├── awards.py        # GET /api/awards
│   ├── certifications.py # GET /api/certifications
│   ├── contact.py       # POST /api/contact
│   ├── cv.py            # GET /api/cv/status, PATCH /api/cv/generate
│   ├── experience.py    # GET /api/experience[/{index}]
│   ├── feature_flags.py # GET + PATCH /api/feature-flags
│   ├── health.py        # GET /health (infra probe), GET /api/health
│   ├── projects.py      # GET /api/projects[/{slug}]
│   ├── skills.py        # GET /api/skills
│   └── uigen.py         # GET /api/uigen-token
├── repositories/
│   ├── awards.py        # list_awards(db) — DB or seed fallback
│   ├── certifications.py # list_certifications(db) — DB or seed fallback
│   ├── experience.py    # list_entries(db), get_by_id(db, id) — DB or seed fallback
│   ├── projects.py      # list_cards(db), get_by_slug(db, slug) — selectinload(Project.phases)
│   └── skills.py        # list_categories(db) — DB or seed fallback
├── schemas/
│   ├── experience.py    # Create/update request schemas for experience
│   ├── projects.py      # Create/update request schemas + ProjectPhaseCreate
│   ├── responses.py     # 11 Pydantic v2 response schemas (from_attributes=True):
│   │                    # ProjectCard, ProjectDetail, ProjectPhase, WorkEntry, Skill,
│   │                    # SkillCategory, Award, Certification, ContactResponse,
│   │                    # HealthResponse, FeatureFlagsResponse
│   └── skills.py        # Create/update request schemas for skills
├── services/
│   ├── cv.py            # generate_cv() via asyncio.create_subprocess_exec
│   ├── email.py         # EmailService (Jinja2 + SMTP)
│   └── uigen_tokens.py  # issue(), validate(), revoke() — SHA-256 hashed, DB-backed
├── templates/
│   ├── contact_admin.html  # Jinja2 template — notification email to portfolio owner
│   └── contact_user.html   # Jinja2 template — confirmation email to form submitter
└── data/
    ├── seed.json        # Canonical static fallback for all content (projects, experience,
    │                    # skills, awards, certifications)
    └── seed.py          # lru_cache loader: load_projects(), load_work_entries(), etc.
alembic/
└── versions/
    ├── 0001_initial_schema.py
    ├── 0002_seed_initial_data.py
    ├── 0003_add_iac_platform_project.py
    ├── 0004_add_awards_certifications.py
    ├── 0005_project_phases_table.py   # Replaces timeline_encoded ARRAY with child table
    └── 0006_uigen_tokens_table.py     # uigen_tokens table for DB-backed token store

API Endpoints¶

Health Check¶

GET /api/health     # Primary — accessible via nginx proxy, included in OpenAPI schema
GET /health         # Internal — for infra probes only, excluded from OpenAPI schema
Response: {"status": "healthy", "message": "Portfolio API is running"}

Contact Form¶

POST /api/contact
Request: {
  "name": "string",
  "email": "string",
  "message": "string",
  "company": "string" (optional)
}
Response: {
  "success": true,
  "message": "Thank you for your message!"
}

CV Download¶

GET /api/download-cv?lang=english|nederlands
Response: PDF file (application/pdf)
# Returns 404 if the PDF file is not present in the container

CV PDFs are copied into the container at build time (client/public/*.pdf → /app/public/). The lang query parameter selects English or Dutch. Defaults to english if omitted or unrecognised.

Projects¶

GET /api/projects                   # list of project cards (from database)
GET /api/projects/{project_id}      # detail by slug (e.g. "genai-framework")
# Returns 404 if the slug is not found

Experience¶

GET /api/experience                 # list of work history entries (from database)
GET /api/experience?top=N           # return only the N most recent entries
GET /api/experience/{index}         # single entry by 0-based index
# Returns 404 if index is out of range

Skills¶

GET /api/skills                     # list of skill categories with nested skills (from database)

Awards and Certifications¶

GET /api/awards                     # list of awards (from database)
GET /api/certifications             # list of certifications (from database)

Data for all groups is read from PostgreSQL via the SQLAlchemy async ORM (see Database section below). See API Endpoints for full request/response shapes.

Admin API¶

The admin write endpoints are protected by a bearer token enforced at the FastAPI router level via dependencies=[Depends(_require_admin)]. They share the same /api/ prefix as the read endpoints — there is no separate /api/admin/ prefix.

# Projects (admin)
POST   /api/projects                    # create a project record
PUT    /api/projects/{slug}             # replace a project (full update)
PATCH  /api/projects/{slug}             # partial update a project
DELETE /api/projects/{slug}             # delete a project
PATCH  /api/projects/reorder            # set sort order for all projects

# Experience (admin)
POST   /api/experience                  # create a work entry
PUT    /api/experience/{entry_id}       # replace a work entry (full update)
PATCH  /api/experience/{entry_id}       # partial update a work entry
DELETE /api/experience/{entry_id}       # delete a work entry
PATCH  /api/experience/reorder          # set sort order for all entries

# Skills (admin)
POST   /api/skills/categories                        # create a skill category
PUT    /api/skills/categories/{category_id}          # replace a skill category
PATCH  /api/skills/categories/{category_id}          # partial update a skill category
DELETE /api/skills/categories/{category_id}          # delete a skill category
POST   /api/skills/categories/{category_id}/skills   # add a skill to a category
PUT    /api/skills/{skill_id}                        # replace a skill
PATCH  /api/skills/{skill_id}                        # partial update a skill
DELETE /api/skills/{skill_id}                        # delete a skill

# All admin routes require: Authorization: Bearer <api-admin-token>

The admin token is stored in Azure Key Vault (api-admin-token secret) and injected as an environment variable via Bicep. These endpoints are used for seeding and updating database content without direct DB access.

The entry_id path parameter for experience is a UUID (matching the database primary key), not a 0-based index. The read endpoint GET /api/experience/{index} still uses a 0-based integer index.

UIGen Token¶

GET /api/uigen-token
# No Authorization header → issues a guest token (read-only + contact/CV)
# Valid Bearer <api-admin-token> → echoes back as admin token (full access)
# Invalid Bearer → 401
Response: {"token": "<token>", "role": "guest|admin"}

The AdminModal in the frontend calls this endpoint to get a scoped token, then opens https://admin.sven-relijveld.com/auth?token=<token>. The UIGen auth shim writes the token to sessionStorage["uigen_auth"] and redirects to UIGen root. Guest tokens expire after 1 hour.

Token storage is handled by api/services/uigen_tokens.py (issue(), validate(), revoke()). Tokens are stored as SHA-256 hashes in the uigen_tokens database table (migration 0006) when a DB is available, falling back to an in-process dict when the DB is unavailable (local dev / DB down). The DB-backed store survives container restarts and works correctly with multi-replica deployments; the in-process fallback is single-replica only.

Feature Flags¶

GET  /api/feature-flags
# Response: {"cvGeneration": true/false, "portfolioContentApi": true/false}

PATCH /api/feature-flags
# Request: {"cvGeneration": true/false, "portfolioContentApi": true/false}
# Requires: Authorization: Bearer <api-admin-token>
# Response: updated flag state

GET reads current flag values from Azure App Configuration (dna-portfolio-we-appcs). The backend uses managed identity (App Configuration Data Reader RBAC) with a 30-second TTL cache. Labels are environment-specific (dev/prod). Falls back to the FEATURE_CV_GENERATION env var if App Config is unavailable.

PATCH writes flag updates back to Azure App Configuration via ff.set_flag(). The admin bearer token is required. Both flags are optional in the request body — only supplied fields are updated.

CV Generation (feature-flagged)¶

GET   /api/cv/status                # returns {"enabled": true/false}
PATCH /api/cv/generate              # triggers Bun publishpipe build, returns PDF
# Request body: {"lang": "english" | "nederlands"}
# Only available when cv-generation feature flag is enabled in Azure App Config

The CV generation pipeline (cv/build.ts) uses Bun + pagedjs-cli to render a custom Nunjucks template into PDF. The PATCH method is used rather than GET because UIGen's view-hint classifier maps PATCH without a path parameter to an action button, which renders correctly as a generate button in the admin portal. Controlled by the cv-generation feature flag in Azure App Configuration (dev=on, prod=on as of v0.10.0).

Rate Limiting¶

All endpoints are rate-limited via slowapi (keyed by client IP). Limits: contact 5/min, CV download 10/min, read endpoints 60/min. Exceeding a limit returns 429 {"detail": "Rate limit exceeded"}.

Database¶

The backend uses Azure PostgreSQL Flexible Server (production) with SQLAlchemy 2 async and asyncpg as the driver. Alembic manages schema migrations.

Models¶

All tables use UUID primary keys generated by the database (gen_random_uuid()). Skill has a cascade-delete foreign key to SkillCategory. ProjectPhase has a cascade-delete foreign key to Project. Project.phases is loaded via selectinload to avoid N+1 queries on project list endpoints.

Connection¶

api/db/engine.py creates an async SQLAlchemy engine via _make_engine(database_url, use_managed_identity):

• USE_MANAGED_IDENTITY=true (Azure): uses ManagedIdentityCredential from azure-identity to fetch a short-lived AAD token and passes it as the asyncpg password callable. The MANAGED_IDENTITY_CLIENT_ID env var (set by Bicep) is passed to ManagedIdentityCredential(client_id=...) to target the specific user-assigned identity.
• USE_MANAGED_IDENTITY=false (local): uses a plain username/password from DATABASE_URL (docker-compose postgres service).

Important: The DATABASE_URL username must be the MI's principal name (e.g. dna-prd-portfolio-we-id-be), not its client ID UUID. PostgreSQL AAD authentication matches on the principal name. The value is written to Key Vault by the deploy workflow using the MI's principal name.

Two async generator dependencies are provided for route handlers:

• get_db() — raises ServiceUnavailableError (→ 503) if the engine is not configured. Used by write endpoints (admin) and any read endpoint with no static fallback.
• get_optional_db() — yields AsyncSession | None; returns None when the engine is not configured. Used by read endpoints that fall back to api/data/seed.json via api/data/seed.py.

Migrations¶

# Apply all pending migrations
cd api && alembic upgrade head

# Generate a new migration after model changes
cd api && alembic revision --autogenerate -m "description"

alembic/env.py uses _make_engine() (not a bare create_async_engine) so Alembic uses the same AAD token logic as the app runtime when USE_MANAGED_IDENTITY=true.

Architecture Pattern¶

The backend uses an ArjanCodes-inspired layered architecture. The dependency flow is strictly one-directional:

core (config, exceptions, logging, middleware)
    ↓
schemas (Pydantic v2 request + response models)
    ↓
repositories (DB queries; fall back to seed.json when DB unavailable)
    ↓
services (EmailService, CvService, UigenTokenService)
    ↓
routers (thin handlers: parse → call repo/service → return schema)

Key design decisions:

• No generic Repository[T] ABC — each aggregate has plain async def functions in its own module. Generic bases only pay off with multiple backend implementations; we have one (PostgreSQL via SQLAlchemy).
• No DI container — FastAPI's Depends() covers every injection need.
• Domain exceptions translate at the edge — repositories raise NotFoundError, ConflictError, etc.; register_exception_handlers() in api/core/exceptions.py maps them to HTTP status codes. Handlers focus on the happy path only.
• DB-or-fallback in repositories, not in handlers — each read repository function accepts db: AsyncSession | None; when None, it returns data from api/data/seed.json via seed.py. Pydantic v2 from_attributes=True schemas validate both ORM rows and plain dicts without a special case at the router edge.
• main.py is a thin factory — create_app() registers routers, middleware, and exception handlers; module-level side effects are limited to app = create_app().

See docs/planning/plan-api-arjan-refactor.md (archived) for the full rationale and ArjanCodes reference map.

Structured Logging and Request IDs¶

Logging (Phase 6, PR #381)¶

api/core/logging.py provides configure_logging(), called once in create_app():

• Installs an InterceptHandler on the root stdlib logger so that uvicorn, SQLAlchemy, and all third-party libraries funnel through loguru.
• In production (settings.environment != "development"): JSON sink to stdout — structured output suitable for Azure Monitor / Log Analytics.
• In development: pretty sink with color and human-readable timestamps.

Request ID Middleware (Phase 6, PR #381)¶

api/core/middleware.py provides RequestIdMiddleware:

• On each incoming request: reads X-Request-Id from the request headers; if absent, generates a new UUID v4.
• Binds the request ID to a contextvars.ContextVar so it is accessible anywhere in the call stack (repositories, services) without explicit threading.
• Echoes the ID back in the response X-Request-Id header.
• Loguru's context filter pulls the current request ID into every log line emitted during request handling, enabling full log correlation by request ID in Log Analytics.

Usage: clients that set X-Request-Id on outbound requests (e.g. the frontend, load balancers) will see their ID round-tripped in the response. Without a client-supplied ID the middleware generates one transparently.

Email Service¶

Features¶

• Dual Emails: Notification + Confirmation
• HTML Templates: Professional branded emails
• Error Handling: Graceful fallbacks
• Logging: Comprehensive error logging

Email Flow¶

SMTP Configuration¶

SMTP_SERVER = "smtp.gmail.com"
SMTP_PORT = 587
EMAIL_ADDRESS = os.getenv("EMAIL_ADDRESS")
EMAIL_PASSWORD = os.getenv("EMAIL_PASSWORD")  # App Password

Security¶

Secrets Management¶

• Azure Key Vault: Production secrets
• Environment Variables: Development
• No Hardcoding: All secrets externalized

CORS Configuration¶

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Configured per environment
    allow_credentials=False,
    allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
    allow_headers=["*"],
)

Docker Build¶

FROM python:3.11-slim

WORKDIR /app

# Install gcc for psycopg2
RUN apt-get update && apt-get install -y --no-install-recommends gcc && rm -rf /var/lib/apt/lists/*

COPY api/requirements.txt .
RUN pip install --no-cache-dir --upgrade pip && pip install --no-cache-dir -r requirements.txt

COPY api/ ./api/

# CV PDFs for /api/download-cv
COPY client/public/*.pdf ./public/

# Run as non-root user
RUN useradd -m -u 1001 appuser && chown -R appuser:appuser /app
USER appuser

EXPOSE 8000

HEALTHCHECK --interval=30s --timeout=3s --start-period=10s --retries=3 \
  CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')" || exit 1

# Run Alembic migrations, then start the server
CMD ["sh", "-c", "n=0; until alembic -c api/alembic.ini upgrade head; do n=$((n+1)); [ $n -ge 5 ] && echo 'Alembic failed after 5 attempts' && exit 1; echo \"Alembic attempt $n failed, retrying in 15s...\"; sleep 15; done && uvicorn api.main:app --host 0.0.0.0 --port 8000 --workers 1"]

Performance¶

• Async/Await: Non-blocking I/O throughout (FastAPI + SQLAlchemy async + asyncpg)
• Connection Pooling: SQLAlchemy async engine manages the DB connection pool; efficient SMTP connections for email
• Monitoring: Application Insights integration

Error Handling¶

SMTPAuthenticationError is caught and logged but does not fail the request — the API always returns success: true so the contact form shows a success message regardless of email delivery. This prevents credential misconfigurations from breaking the UX.

try:
    # Email sending logic
    return {"success": True}
except SMTPAuthenticationError:
    logger.error("Authentication failed")
    return {"success": True}  # always succeed from the user's perspective
except Exception as e:
    logger.error(f"Unexpected error: {e}")
    raise HTTPException(status_code=500)

Testing¶

Tests are split across two files and share a conftest.py that sets required environment variables before api.main is imported.

tests/conftest.py — sets EMAIL_ADDRESS and EMAIL_PASSWORD defaults so pydantic-settings can instantiate without real credentials.

tests/test_api_endpoints.py — uses FastAPI's TestClient with smtplib.SMTP mocked via unittest.mock.patch. A reset_rate_limiter autouse fixture clears slowapi state before each test. No running server is required.

tests/test_cv_download.py — tests the /api/download-cv endpoint by patching CV_FILES to use a temporary PDF on disk.

tests/test_cv_generation.py — tests the feature-flagged CV generation endpoints.

tests/test_cv_files_exist.py — validates that CV PDF files are present in the container's public/ directory.

Running locally:

pytest tests/ -v

CI: test.yml runs on PR (jobs: api, unit, typecheck, links); test-backend job in deploy-pr-preview.yml runs after each preview deploy.

See Also¶

• Frontend Architecture
• API Documentation
• Pipeline Overview