FastAPI Interview Pro — Production Setup Guide

A production-ready FastAPI project using UV, async SQLAlchemy 2.0, Alembic, Pydantic v2, middleware, and a layered architecture (API → Services → Repositories → Models).

---

0) Prerequisites

• Python 3.12+
• UV installed: https://docs.astral.sh/uv/
• SQLite (dev) or Postgres (prod)

# mac/linux
curl -LsSf https://astral.sh/uv/install.sh | sh
# windows (powershell)
iwr -useb https://astral.sh/uv/install.ps1 | iex

---

1) Project Setup with UV

mkdir interview-pro && cd interview-pro
uv init --package interview-pro

Install dependencies:

uv add alembic fastapi pydantic pydantic-settings sqlalchemy uvicorn \
       aiosqlite greenlet httpx pytest pytest-asyncio python-dotenv \
       python-jose starlette
uv sync

Project structure:

interview-pro/
├── src/interview_pro/
│   ├── api/v1/          # API routes & dependencies
│   ├── domain/
│   │   ├── config.py    # Settings (Pydantic v2)
│   │   ├── db.py        # Async SQLAlchemy engine & session
│   │   ├── logging.py   # Logging setup
│   │   ├── models/      # SQLAlchemy ORM models
│   │   ├── dtos/        # Pydantic schemas
│   │   ├── middleware/  # Request logging, timing
│   │   └── errors/      # Custom error handlers
│   ├── repositories/    # Data access layer
│   ├── services/        # Business logic layer
│   ├── tests/           # Pytest async tests
│   └── main.py          # FastAPI app entry
├── alembic/             # Database migrations
├── .env                 # Environment variables
├── Makefile             # Dev commands
└── pyproject.toml       # UV dependencies & pytest config

---

2) Environment & Makefile

.env (project root):

PYTHONPATH=src
APP_NAME=FastAPI Interview Pro
DEBUG=true
DATABASE_URL=sqlite+aiosqlite:///./app.db
SYNC_DATABASE_URL=sqlite:///./app.db
LOG_LEVEL=INFO

Makefile:

.PHONY: setup install dev run test lint

setup:
	uv sync

install:
	uv sync --mode install

dev:
	uv run uvicorn src.interview_pro.main:app --reload

run:
	uv run uvicorn src.interview_pro.main:app --host 0.0.0.0 --port 8000

test:
	uv run pytest src/interview_pro/tests
	
lint:
	uv run ruff src/interview_pro

---

3) Config & Logging (Pydantic v2)

src/interview_pro/domain/config.py — Key pattern: SettingsConfigDict

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    app_name: str = "FastAPI interview pro"
    database_url: str = "sqlite+aiosqlite:///./test.db"
    sync_database_url: str = "sqlite:///./app.db"  # For Alembic
    log_level: str = "INFO"

    model_config = SettingsConfigDict(
        case_sensitive=False,
        env_file=".env",
        extra="ignore",  # Ignore unknown env vars
    )

@lru_cache
def get_settings() -> Settings:
    return Settings()

src/interview_pro/domain/logging.py — Standard dictConfig setup

from logging.config import dictConfig

def setup_logging(level: str = "INFO") -> None:
    dictConfig({
        "version": 1,
        "formatters": {"default": {"format": "%(asctime)s %(levelname)s %(name)s %(message)s"}},
        "handlers": {"console": {"class": "logging.StreamHandler", "formatter": "default"}},
        "root": {"handlers": ["console"], "level": level},
    })

---

4) FastAPI App with Lifespan & Middleware

src/interview_pro/main.py — Core pattern: lifespan context manager

from contextlib import asynccontextmanager
from fastapi import FastAPI
from interview_pro.domain.db import get_engine, Base

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: create tables for dev/test
    engine = get_engine()
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    # Shutdown: cleanup
    await engine.dispose()

app = FastAPI(title="Interview Pro", lifespan=lifespan)

# Middleware & error handlers
app.add_middleware(RequestLoggerMiddleware)
app.exception_handler(ValueError)(value_error_handler)

# Routes
app.include_router(api_router, prefix="/api/v1")

Run: make dev → http://127.0.0.1:8000/docs

---

5) Database (Async SQLAlchemy)

src/interview_pro/domain/db.py — Lazy engine init (avoids import-time driver loading)

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.orm import declarative_base

_engine = None
Base = declarative_base()

def get_engine():
    global _engine
    if _engine is None:
        _engine = create_async_engine(get_settings().database_url)
    return _engine

async def get_async_session() -> AsyncSession:
    async with async_sessionmaker(get_engine(), expire_on_commit=False)() as session:
        yield session

src/interview_pro/domain/models/product.py — SQLAlchemy 2.0 typed mappings

from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy import String, Float, DateTime
from datetime import datetime

class Product(Base):
    __tablename__ = "products"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String, unique=True, index=True)
    price: Mapped[float] = mapped_column(Float, default=0.0)
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
    updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

---

6) Alembic (Database Migrations)

Setup:

uv run alembic init alembic
# Edit alembic.ini: set prepend_sys_path = .

alembic/env.py — Key changes:

from interview_pro.domain.config import get_settings
from interview_pro.domain.db import Base
from interview_pro.domain.models import product  # Import models for autogenerate

target_metadata = Base.metadata

def run_migrations_online() -> None:
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
        url=get_settings().sync_database_url  # Use sync URL
    )
    # ... rest of standard Alembic setup

Commands:

uv run alembic revision --autogenerate -m "create products"
uv run alembic upgrade head

---

7) Layered Architecture (DTOs → Services → Repositories)

src/interview_pro/domain/dtos/productDTO.py — Pydantic v2 DTOs

from pydantic import BaseModel, ConfigDict

class ProductCreate(BaseModel):
    name: str
    price: float

class ProductResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)  # ORM mode
    id: int
    name: str
    price: float

src/interview_pro/repositories/product_repo.py — Data access layer

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

class ProductRepository:
    def __init__(self, session: AsyncSession):
        self.session = session

    async def create(self, name: str, price: float) -> Product:
        p = Product(name=name, price=price)
        self.session.add(p)
        await self.session.commit()
        await self.session.refresh(p)
        return p

    async def list(self) -> Sequence[Product]:
        res = await self.session.execute(select(Product))
        return list(res.scalars().all())
    
    # ... get, update, delete methods

src/interview_pro/services/productService.py — Business logic layer

class ProductService:
    def __init__(self, repo: ProductRepository):
        self.repo = repo

    async def create(self, data: ProductCreate) -> ProductResponse:
        p = await self.repo.create(data.name, data.price)
        return ProductResponse.model_validate(p)

    async def list(self) -> Sequence[ProductResponse]:
        items = await self.repo.list()
        return [ProductResponse.model_validate(p) for p in items]
    
    # ... get, update, delete, search methods

src/interview_pro/api/v1/routes.py — API layer with dependency injection

from fastapi import APIRouter, Depends, HTTPException, status

router = APIRouter(prefix="/products", tags=["products"])

def get_service(session: AsyncSession = Depends(get_async_session)) -> ProductService:
    return ProductService(ProductRepository(session))

@router.post("", response_model=ProductResponse, status_code=201)
async def create_product(data: ProductCreate, svc: ProductService = Depends(get_service)):
    return await svc.create(data)

@router.get("", response_model=list[ProductResponse])
async def list_products(svc: ProductService = Depends(get_service)):
    return await svc.list()

@router.get("/{product_id}", response_model=ProductResponse)
async def get_product(product_id: int, svc: ProductService = Depends(get_service)):
    p = await svc.get(product_id)
    if not p:
        raise HTTPException(status_code=404, detail="Not found")
    return p

---

8) Middleware & Error Handlers

CORS middleware — Environment-aware configuration:

# src/interview_pro/domain/middleware/cors.py
def get_cors_config() -> dict:
    if settings.debug:
        return {"allow_origins": ["*"], "allow_credentials": True, "allow_methods": ["*"]}
    else:
        return {"allow_origins": ["https://yourdomain.com"], "allow_credentials": True}

# In main.py
from starlette.middleware.cors import CORSMiddleware
app.add_middleware(CORSMiddleware, **get_cors_config())

Request logging middleware:

class RequestLoggerMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        start = time()
        response = await call_next(request)
        logger.info("%s %s -> %s (%.2f ms)",
                    request.method, request.url.path, response.status_code, (time() - start) * 1000)
        return response

Custom error handler:

class DomainError(Exception):
    def __init__(self, message: str, code: int = 400):
        self.message = message
        self.code = code

async def domain_error_handler(request: Request, exc: DomainError):
    return JSONResponse(status_code=exc.code, content={"error": exc.message})

---

9) Testing (Async Pytest)

pyproject.toml — Enable async test support:

[tool.pytest.ini_options]
asyncio_mode = "auto"

src/interview_pro/tests/test_products_api.py — Use ASGITransport:

from httpx import AsyncClient, ASGITransport
from interview_pro.main import app

@pytest.mark.asyncio
async def test_crud_flow():
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
        # Create
        r = await ac.post("/api/v1/products", json={"name": "Widget", "price": 1999})
        assert r.status_code == 201
        
        # List
        r = await ac.get("/api/v1/products")
        assert r.status_code == 200

Run: make test

---

10) Common Commands

make setup               # Install dependencies (uv sync)
make dev                 # Run locally with reload
make run                 # Run production mode
make test                # Run pytest
make lint                # Run ruff linter

---

11) Key Features Implemented

✅ Pydantic v2 settings with SettingsConfigDict
✅ Async SQLAlchemy 2.0 with lazy engine initialization
✅ Alembic migrations with autogenerate
✅ Lifespan context manager (replaces deprecated on_event)
✅ Layered architecture: API → Services → Repositories → Models
✅ Middleware: CORS (environment-aware), request logging, timing headers
✅ Error handlers: Domain errors, ValueError mapping
✅ Request ID tracking via dependency injection
✅ Async pytest with httpx.ASGITransport
✅ Search endpoint with case-insensitive LIKE queries

---

12) Next Steps (Production Hardening)

• Auth: JWT tokens, refresh tokens, role-based access control
• Database: Migrate to Postgres with connection pooling
• Deployment: Dockerfile, docker-compose, Kubernetes manifests
• CI/CD: GitHub Actions, pre-commit hooks, automated tests
• Observability: Structured JSON logs, OpenTelemetry tracing, Prometheus metrics
• API Features: Pagination, rate limiting, idempotency keys, API versioning
• Security: HTTPS, secrets management (Vault, AWS Secrets Manager), rate limiting

---

13) Dependencies Reference

dependencies = [
    "alembic>=1.17.0",
    "fastapi>=0.119.0",
    "pydantic>=2.12.2",
    "pydantic-settings>=2.11.0",
    "pytest>=8.4.2",
    "pytest-asyncio>=0.23.0",
    "python-dotenv>=1.1.1",
    "python-jose>=3.5.0",
    "sqlalchemy>=2.0.44",
    "uvicorn>=0.37.0",
    "aiosqlite>=0.19",
    "httpx>=0.28.1",
    "greenlet>=3.0.0",
    "starlette>=0.48.0",
]

Key dependency notes:

• greenlet: Required for SQLAlchemy async-to-sync bridge
• aiosqlite: Async SQLite driver for development
• httpx: Async HTTP client for testing
• pytest-asyncio: Enables async test support

---

Happy building! 🚀