Database Migrations with Alembic

Install a specific version of Alembic:

pip install alembic==1.13.1

Initialize the Alembic project structure:

alembic init alembic

This creates an alembic.ini file. Update the sqlalchemy.url to match your database configuration:

[alembic]
script_location = alembic
sqlalchemy.url = mysql+pymysql://david:123456@localhost/summary

[loggers]
keys = root

[handlers]
keys = console

[formatters]
keys = generic

[logger_root]
level = WARN
handlers = console
qualname =

[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic

[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s

In alembic/env.py, import your SQLAlchemy base model and assign its metadtaa:

from models import Base
target_metadata = Base.metadata

Generate a migration script after defining a new model:

alembic revision --autogenerate -m "create_user_table"

Apply the migration:

alembic upgrade head

For future schema changes, repeat the autogenerate and upgrade steps:

alembic revision --autogenerate -m "add phone field to user"
alembic upgrade head

API Key Authentication

Use APIKeyHeader for simple token-based protection:

from fastapi import FastAPI, Security, HTTPException
from fastapi.security import APIKeyHeader

app = FastAPI()
api_key_header = APIKeyHeader(name="X-API-Key")

@app.get("/secure-endpoint")
async def secure_route(api_key: str = Security(api_key_header)):
    if api_key != "secret-token-123":
        raise HTTPException(status_code=401, detail="Unauthorized")
    return {"data": "protected content"}

Middleware Configuration

Add custom middleware to log request processing time and enable CORS:

import time
from fastapi import Request
from fastapi.middleware.cors import CORSMiddleware

@app.middleware("http")
async def add_timing_header(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start
    response.headers["X-Processing-Time"] = str(duration)
    return response

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Background Task Execution

Schedule lightweight background operations using BackgroundTasks:

from fastapi import APIRouter, BackgroundTasks

router = APIRouter()

def write_log(message: str):
    with open("activity.log", "a") as log:
        log.write(f"{message}\n")

@router.post("/trigger-task")
async def trigger_task(background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, "User initiated background job")
    return {"status": "accepted"}

Dependency injection can be used to conditionally schedule tasks:

from typing import Optional

def maybe_schedule_task(background_tasks: BackgroundTasks, flag: Optional[str] = None):
    if flag:
        background_tasks.add_task(write_log, f"Flag enabled: {flag}")
    return flag

@router.post("/conditional-task")
async def conditional_task(flag: str = Depends(maybe_schedule_task)):
    if flag:
        return {"result": "logged"}

Note: Avoid injecting dependencies like database sessions directly into background task functions. Pass only serializable data or manage connections inside the task.

Writing Unit Tests

Use TestClient from fastapi.testclient to simulate requests:

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_trigger_background_task():
    response = client.post("/trigger-task")
    assert response.status_code == 200
    assert response.json() == {"status": "accepted"}

def test_conditional_task_without_flag():
    response = client.post("/conditional-task")
    assert response.status_code == 200
    assert response.json() is None

def test_conditional_task_with_flag():
    response = client.post("/conditional-task?flag=test")
    assert response.status_code == 200
    assert response.json() == {"result": "logged"}