Lesson 40 of 40 Capstone Expert ⏱ 35 min

Capstone — Full Python Production App

Build a complete Python application — FastAPI backend, SQLAlchemy ORM, async tasks with Celery, Docker, CI/CD pipeline, and Azure deployment.

Part 1: Capstone Goal and Architecture

The final project combines the major production topics from the previous lessons into a small Task Processing API. The application accepts tasks through FastAPI, stores them with SQLAlchemy, can hand long-running work to a Celery worker, and is prepared for containerised deployment and CI.

API layer: FastAPI routes and Pydantic validation.
Persistence: SQLAlchemy ORM.
Background work: Celery worker for long-running jobs.
Packaging: Docker image.
Quality gate: automated linting and tests before deployment.

Part 2: FastAPI + SQLAlchemy Core

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel, ConfigDict
from sqlalchemy import create_engine, String
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    Session,
    mapped_column,
    sessionmaker,
)

DATABASE_URL = "sqlite:///./tasks.db"

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False},
)

SessionLocal = sessionmaker(
    bind=engine,
    autoflush=False,
    autocommit=False,
)

class Base(DeclarativeBase):
    pass

class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(120))
    status: Mapped[str] = mapped_column(
        String(30),
        default="pending",
    )

class TaskCreate(BaseModel):
    title: str

class TaskRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    title: str
    status: str

Base.metadata.create_all(engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

app = FastAPI(title="Task Processing API")

@app.post("/tasks", response_model=TaskRead)
def create_task(
    request: TaskCreate,
    db: Session = Depends(get_db),
) -> Task:
    task = Task(title=request.title)
    db.add(task)
    db.commit()
    db.refresh(task)
    return task

@app.get("/tasks/{task_id}", response_model=TaskRead)
def get_task(
    task_id: int,
    db: Session = Depends(get_db),
) -> Task:
    task = db.get(Task, task_id)

    if task is None:
        raise HTTPException(status_code=404, detail="Task not found")

    return task

pip install fastapi sqlalchemy "uvicorn[standard]"

Part 3: Background Worker with Celery

import os
from celery import Celery

broker_url = os.environ.get(
    "CELERY_BROKER_URL",
    "redis://localhost:6379/0",
)

celery_app = Celery(
    "task_worker",
    broker=broker_url,
    backend=broker_url,
)

@celery_app.task
def process_task(task_id: int) -> dict:
    # Replace this demonstration with real long-running work.
    return {
        "task_id": task_id,
        "status": "processed",
    }

pip install celery redis

The API can call process_task.delay(task.id) after saving a task. In a production design, the worker should open its own database session and update the task status after processing.

Part 4: Containerise the API

FROM python:3.13-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

fastapi
uvicorn[standard]
sqlalchemy
celery
redis

Part 5: CI and Deployment Checklist

Run ruff check . and pytest in GitHub Actions or Azure Pipelines.
Build the Docker image only after tests pass.
Inject database URLs, broker URLs, API keys, and cloud credentials through environment variables or a secret store.
For Azure deployment, push the image to a registry and deploy the same tested image to the selected Azure container service.
Use a production database instead of SQLite when multiple application instances must share durable state.
Add health checks, structured logging, metrics, backups, and migration tooling before treating the project as production-ready.

Part 6: Final Capstone Challenge

Extend the Task Processing API into a portfolio project. Add user authentication, PostgreSQL, Alembic migrations, a Redis-backed Celery worker, pytest API tests, Docker Compose for local development, and a CI pipeline. Then deploy the tested container image to your chosen cloud environment.

You have now reached Lesson 40. Return to Python Tutorial Home to revisit individual topics or use this capstone as the starting point for a larger Python application.

📘 Want the complete guide with projects? Get the book →