FastAPI – Introduction & Setup

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI()

class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: str = Field(..., pattern=r"^[\w\.-]+@[\w\.-]+\.\w+$")
    age: int = Field(..., ge=13, le=120)

@app.post("/users")
async def create_user(user: UserCreate):
    # If we reach this line, the data is guaranteed valid
    # FastAPI already validated username length, email format, and age range
    return {"message": f"User {user.username} created successfully"}

import httpx
from fastapi import FastAPI

app = FastAPI()

@app.get("/weather/{city}")
async def get_weather(city: str):
    # Non-blocking HTTP call - server handles other requests while waiting
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://api.weatherapi.com/v1/current.json?q={city}"
        )
    return response.json()

python3 --version
# Python 3.11.6 (or higher)

# Create project directory
mkdir fastapi-tutorial
cd fastapi-tutorial

# Create virtual environment
python3 -m venv venv

# Activate virtual environment
# On macOS/Linux:
source venv/bin/activate

# On Windows:
# venv\Scripts\activate

# Verify activation (should show venv path)
which python

# Install FastAPI with all optional dependencies
pip install "fastapi[standard]"

# Minimal installation
pip install fastapi uvicorn

# Add extras as needed
pip install python-multipart   # For form data
pip install python-jose[cryptography]  # For JWT tokens
pip install passlib[bcrypt]    # For password hashing
pip install sqlalchemy         # For database ORM
pip install alembic            # For database migrations

pip show fastapi

Name: fastapi
Version: 0.115.x
Summary: FastAPI framework, high performance, easy to learn, fast to code, ready for production
Home-page: https://github.com/fastapi/fastapi
Author-email: Sebastián Ramírez <tiangolo@gmail.com>
License: MIT
Requires: pydantic, starlette, typing-extensions

# Save current dependencies
pip freeze > requirements.txt

# Later, install from requirements file
pip install -r requirements.txt

from fastapi import FastAPI

# Create the FastAPI application instance
app = FastAPI()

# Define a route using a path operation decorator
@app.get("/")
async def root():
    """Root endpoint that returns a welcome message."""
    return {"message": "Hello, World!"}

uvicorn main:app --reload

INFO:     Will watch for changes in these directories: ['/path/to/fastapi-tutorial']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [28720] using StatReload
INFO:     Started server process [28722]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

{"message": "Hello, World!"}

curl http://127.0.0.1:8000
# {"message":"Hello, World!"}

from fastapi import FastAPI

app = FastAPI(
    title="My First FastAPI App",
    description="A tutorial application to learn FastAPI basics",
    version="1.0.0"
)

@app.get("/")
async def root():
    """Root endpoint - welcome message."""
    return {"message": "Hello, World!", "docs": "/docs"}

@app.get("/health")
async def health_check():
    """Health check endpoint for monitoring."""
    return {"status": "healthy"}

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    """Read a single item by its ID.

    The item_id parameter is automatically parsed as an integer.
    If a non-integer is provided, FastAPI returns a 422 error.
    """
    return {"item_id": item_id, "name": f"Item {item_id}"}

@app.get("/search")
async def search_items(q: str, limit: int = 10, offset: int = 0):
    """Search for items with query parameters.

    - q: required search query (string)
    - limit: optional, defaults to 10
    - offset: optional, defaults to 0
    """
    return {
        "query": q,
        "limit": limit,
        "offset": offset,
        "results": [f"Result for '{q}' #{i}" for i in range(offset, offset + limit)]
    }

curl http://127.0.0.1:8000/openapi.json | python3 -m json.tool

from fastapi import FastAPI

app = FastAPI(
    title="Tutorial API",
    description="""
## Tutorial API for learning FastAPI

This API demonstrates:
* **CRUD operations** on items
* **User management** with authentication
* **Search functionality** with pagination

### Authentication
Most endpoints require a valid JWT token in the Authorization header.
    """,
    version="2.0.0",
    terms_of_service="https://example.com/terms",
    contact={
        "name": "API Support",
        "url": "https://example.com/support",
        "email": "support@example.com",
    },
    license_info={
        "name": "MIT License",
        "url": "https://opensource.org/licenses/MIT",
    },
)

@app.get(
    "/items/{item_id}",
    summary="Get a single item",
    description="Retrieve an item by its unique integer ID. Returns 404 if not found.",
    response_description="The requested item",
    tags=["Items"],
    deprecated=False,
)
async def read_item(item_id: int):
    return {"item_id": item_id}

# Without type hints
def greet(name):
    return f"Hello, {name}"

# With type hints
def greet(name: str) -> str:
    return f"Hello, {name}"

# Primitive types
name: str = "FastAPI"
age: int = 30
price: float = 19.99
is_active: bool = True

# None type
result: None = None

# Function with type hints
def calculate_total(price: float, quantity: int, tax_rate: float = 0.08) -> float:
    """Calculate total price with tax."""
    subtotal = price * quantity
    return subtotal * (1 + tax_rate)

# Python 3.9+ syntax (recommended)
names: list[str] = ["Alice", "Bob", "Charlie"]
scores: dict[str, int] = {"Alice": 95, "Bob": 87}
unique_ids: set[int] = {1, 2, 3}
coordinates: tuple[float, float] = (40.7128, -74.0060)

# Python 3.7-3.8 syntax (use typing module)
from typing import List, Dict, Set, Tuple

names: List[str] = ["Alice", "Bob", "Charlie"]
scores: Dict[str, int] = {"Alice": 95, "Bob": 87}
unique_ids: Set[int] = {1, 2, 3}
coordinates: Tuple[float, float] = (40.7128, -74.0060)

# Python 3.10+ syntax
def find_user(user_id: int) -> dict | None:
    """Returns user dict or None if not found."""
    ...

def process_input(value: str | int) -> str:
    """Accepts either a string or integer."""
    return str(value)

# Python 3.7-3.9 syntax
from typing import Optional, Union

def find_user(user_id: int) -> Optional[dict]:
    """Returns user dict or None if not found."""
    ...

def process_input(value: Union[str, int]) -> str:
    """Accepts either a string or integer."""
    return str(value)

from fastapi import FastAPI

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(
    user_id: int,           # Path param: parsed as int, 422 if invalid
    include_posts: bool = False,  # Query param: parsed as bool, default False
    limit: int = 10,        # Query param: parsed as int, default 10
) -> dict:                  # Return type hint for documentation
    """
    FastAPI uses these type hints to:
    1. Extract user_id from the URL path and convert to int
    2. Extract include_posts and limit from query string
    3. Validate all types and return 422 on failure
    4. Document all parameters with their types in /docs
    """
    user = {"id": user_id, "name": f"User {user_id}"}
    if include_posts:
        user["posts"] = [{"title": f"Post {i}"} for i in range(limit)]
    return user

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

# In-memory "database" for demonstration
books_db: dict[int, dict] = {}
next_id: int = 1


class BookCreate(BaseModel):
    title: str
    author: str
    year: int
    isbn: str | None = None


class BookUpdate(BaseModel):
    title: str | None = None
    author: str | None = None
    year: int | None = None
    isbn: str | None = None


# CREATE - POST
@app.post("/books", status_code=201)
async def create_book(book: BookCreate):
    """Create a new book."""
    global next_id
    book_data = book.model_dump()
    book_data["id"] = next_id
    books_db[next_id] = book_data
    next_id += 1
    return book_data


# READ ALL - GET
@app.get("/books")
async def list_books(skip: int = 0, limit: int = 20):
    """List all books with pagination."""
    all_books = list(books_db.values())
    return all_books[skip : skip + limit]


# READ ONE - GET
@app.get("/books/{book_id}")
async def get_book(book_id: int):
    """Get a single book by ID."""
    if book_id not in books_db:
        raise HTTPException(status_code=404, detail="Book not found")
    return books_db[book_id]


# FULL UPDATE - PUT
@app.put("/books/{book_id}")
async def replace_book(book_id: int, book: BookCreate):
    """Replace a book entirely."""
    if book_id not in books_db:
        raise HTTPException(status_code=404, detail="Book not found")
    book_data = book.model_dump()
    book_data["id"] = book_id
    books_db[book_id] = book_data
    return book_data


# PARTIAL UPDATE - PATCH
@app.patch("/books/{book_id}")
async def update_book(book_id: int, book: BookUpdate):
    """Partially update a book."""
    if book_id not in books_db:
        raise HTTPException(status_code=404, detail="Book not found")
    stored_book = books_db[book_id]
    update_data = book.model_dump(exclude_unset=True)
    stored_book.update(update_data)
    return stored_book


# DELETE - DELETE
@app.delete("/books/{book_id}", status_code=204)
async def delete_book(book_id: int):
    """Delete a book."""
    if book_id not in books_db:
        raise HTTPException(status_code=404, detail="Book not found")
    del books_db[book_id]
    return None

from fastapi import FastAPI
from enum import Enum

app = FastAPI()

# Basic path parameter with type validation
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return {"user_id": user_id}

# Multiple path parameters
@app.get("/organizations/{org_id}/teams/{team_id}")
async def get_team(org_id: int, team_id: int):
    return {"org_id": org_id, "team_id": team_id}

# Enum path parameter - restricts to predefined values
class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"

@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
    # Only accepts "alexnet", "resnet", or "lenet"
    # Returns 422 for any other value
    return {"model": model_name, "message": f"Selected {model_name.value}"}

# File path parameter (captures the rest of the path)
@app.get("/files/{file_path:path}")
async def read_file(file_path: str):
    # Matches /files/home/user/document.txt
    return {"file_path": file_path}

from fastapi import FastAPI, Query

app = FastAPI()

# Basic query parameters
@app.get("/items")
async def list_items(
    skip: int = 0,      # Optional with default
    limit: int = 10,    # Optional with default
    q: str | None = None  # Optional, can be None
):
    # URL: /items?skip=20&limit=5&q=search_term
    return {"skip": skip, "limit": limit, "q": q}

# Required query parameter (no default value)
@app.get("/search")
async def search(q: str):
    # URL: /search?q=python  (q is required)
    return {"query": q}

# Advanced query validation with Query()
@app.get("/products")
async def list_products(
    q: str | None = Query(
        default=None,
        min_length=3,
        max_length=50,
        pattern=r"^[a-zA-Z0-9\s]+$",
        title="Search query",
        description="Search products by name. Minimum 3 characters.",
        examples=["laptop", "wireless mouse"],
    ),
    page: int = Query(default=1, ge=1, description="Page number (starting from 1)"),
    size: int = Query(default=20, ge=1, le=100, description="Items per page (max 100)"),
    sort_by: str = Query(default="name", pattern=r"^(name|price|rating)$"),
):
    return {
        "query": q,
        "page": page,
        "size": size,
        "sort_by": sort_by,
    }

# Visualizing the request lifecycle

# Step 1: Client sends HTTP request
# GET /users/42?include_email=true HTTP/1.1
# Host: api.example.com
# Authorization: Bearer eyJhbGci...

# Step 2: Uvicorn (ASGI server) receives the raw TCP connection
#   - Parses HTTP protocol
#   - Creates ASGI scope dictionary
#   - Passes to the ASGI application (FastAPI/Starlette)

# Step 3: Starlette middleware stack processes the request
#   - CORS middleware
#   - GZip middleware
#   - Authentication middleware
#   - Custom middleware
#   (Each middleware can modify request/response or short-circuit)

# Step 4: Starlette router matches the URL path to a route
#   - Finds the handler for GET /users/{user_id}
#   - Extracts path parameters: user_id = "42"

# Step 5: FastAPI dependency injection
#   - Resolves dependencies declared with Depends()
#   - Runs authentication, database session creation, etc.

# Step 6: FastAPI request parsing and validation
#   - Converts path params: user_id = int("42") = 42
#   - Parses query params: include_email = bool("true") = True
#   - Validates request body against Pydantic model (if applicable)
#   - Returns 422 if validation fails

# Step 7: Your handler function executes
#   async def get_user(user_id: int, include_email: bool = False):
#       user = await db.get_user(user_id)
#       return user

# Step 8: FastAPI response processing
#   - Serializes return value to JSON
#   - Applies response_model filtering (if defined)
#   - Sets status code and headers
#   - Validates response against response_model (in debug mode)

# Step 9: Middleware stack processes the response (in reverse order)
#   - GZip compresses response body
#   - CORS adds Access-Control headers

# Step 10: Uvicorn sends HTTP response to client
#   HTTP/1.1 200 OK
#   Content-Type: application/json
#   {"id": 42, "name": "Alice", "email": "alice@example.com"}

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

app = FastAPI()

# Built-in CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "https://myapp.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Custom middleware: request timing
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    """Measure and log request processing time."""
    start_time = time.perf_counter()

    response = await call_next(request)

    process_time = time.perf_counter() - start_time
    response.headers["X-Process-Time"] = f"{process_time:.4f}"
    print(f"{request.method} {request.url.path} - {process_time:.4f}s")

    return response

# Custom middleware: request ID tracking
import uuid

@app.middleware("http")
async def add_request_id(request: Request, call_next):
    """Add a unique request ID to every request/response."""
    request_id = str(uuid.uuid4())
    request.state.request_id = request_id

    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id

    return response

@app.get("/")
async def root(request: Request):
    return {
        "message": "Hello",
        "request_id": request.state.request_id
    }

from fastapi import FastAPI, Request

app = FastAPI()

@app.get("/request-info")
async def request_info(request: Request):
    """Inspect the raw request object."""
    return {
        "method": request.method,
        "url": str(request.url),
        "base_url": str(request.base_url),
        "path": request.url.path,
        "query_params": dict(request.query_params),
        "path_params": request.path_params,
        "headers": dict(request.headers),
        "client_host": request.client.host if request.client else None,
        "client_port": request.client.port if request.client else None,
        "cookies": request.cookies,
    }

from fastapi import FastAPI
from fastapi.responses import (
    JSONResponse,
    HTMLResponse,
    PlainTextResponse,
    RedirectResponse,
    StreamingResponse,
    FileResponse,
)

app = FastAPI()

# Default: JSON response (automatic)
@app.get("/json")
async def json_response():
    return {"message": "This is JSON"}  # Automatically wrapped in JSONResponse

# Explicit JSON with custom status code and headers
@app.get("/custom-json")
async def custom_json():
    return JSONResponse(
        content={"message": "Custom response"},
        status_code=201,
        headers={"X-Custom-Header": "value"},
    )

# HTML response
@app.get("/html", response_class=HTMLResponse)
async def html_response():
    return "<h1>Hello from FastAPI</h1><p>This is HTML.</p>"

# Plain text response
@app.get("/text", response_class=PlainTextResponse)
async def text_response():
    return "Just plain text"

# Redirect response
@app.get("/old-page")
async def redirect():
    return RedirectResponse(url="/new-page", status_code=301)

# File download
@app.get("/download")
async def download_file():
    return FileResponse(
        path="report.pdf",
        filename="monthly-report.pdf",
        media_type="application/pdf",
    )

# Streaming response (for large data)
@app.get("/stream")
async def stream_data():
    async def generate():
        for i in range(100):
            yield f"data: chunk {i}\n\n"
    return StreamingResponse(generate(), media_type="text/event-stream")

fastapi-project/
├── main.py              # Application entry point, all routes
├── models.py            # Pydantic models (schemas)
├── database.py          # Database connection and session
├── requirements.txt     # Dependencies
├── .env                 # Environment variables (never commit)
├── .gitignore
└── tests/
    ├── __init__.py
    └── test_main.py

fastapi-project/
├── app/
│   ├── __init__.py
│   ├── main.py           # Application factory, startup/shutdown events
│   ├── config.py          # Settings and configuration
│   ├── dependencies.py    # Shared dependencies (auth, db session)
│   ├── database.py        # Database engine and session setup
│   ├── models/            # SQLAlchemy ORM models
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   ├── schemas/           # Pydantic request/response models
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   ├── routers/           # API route handlers (grouped by domain)
│   │   ├── __init__.py
│   │   ├── users.py
│   │   └── items.py
│   ├── services/          # Business logic layer
│   │   ├── __init__.py
│   │   ├── user_service.py
│   │   └── item_service.py
│   └── middleware/        # Custom middleware
│       ├── __init__.py
│       └── logging.py
├── tests/
│   ├── __init__.py
│   ├── conftest.py        # Shared fixtures
│   ├── test_users.py
│   └── test_items.py
├── alembic/               # Database migrations
│   ├── env.py
│   └── versions/
├── alembic.ini
├── requirements.txt
├── .env
└── .gitignore

fastapi-project/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── core/              # Core application components
│   │   ├── __init__.py
│   │   ├── config.py      # Pydantic Settings for configuration
│   │   ├── security.py    # JWT, password hashing, OAuth2
│   │   ├── database.py    # Database engine, session, base model
│   │   ├── dependencies.py # Shared dependencies
│   │   └── exceptions.py  # Custom exception handlers
│   ├── features/          # Feature modules (domain-driven)
│   │   ├── auth/
│   │   │   ├── __init__.py
│   │   │   ├── router.py
│   │   │   ├── schemas.py
│   │   │   ├── models.py
│   │   │   ├── service.py
│   │   │   └── dependencies.py
│   │   ├── users/
│   │   │   ├── __init__.py
│   │   │   ├── router.py
│   │   │   ├── schemas.py
│   │   │   ├── models.py
│   │   │   ├── service.py
│   │   │   └── repository.py
│   │   └── products/
│   │       ├── __init__.py
│   │       ├── router.py
│   │       ├── schemas.py
│   │       ├── models.py
│   │       ├── service.py
│   │       └── repository.py
│   └── common/            # Shared utilities
│       ├── __init__.py
│       ├── pagination.py
│       ├── responses.py
│       └── validators.py
├── tests/
├── alembic/
├── docker/
│   ├── Dockerfile
│   └── docker-compose.yml
├── scripts/
│   ├── seed_db.py
│   └── create_admin.py
├── pyproject.toml         # Modern Python project config
├── requirements/
│   ├── base.txt
│   ├── dev.txt
│   └── prod.txt
├── .env.example
└── .gitignore

# app/routers/users.py
from fastapi import APIRouter, HTTPException, Depends
from app.schemas.user import UserCreate, UserResponse, UserUpdate
from app.services.user_service import UserService
from app.dependencies import get_db, get_current_user

router = APIRouter(
    prefix="/users",
    tags=["Users"],
    responses={404: {"description": "User not found"}},
)

@router.get("/", response_model=list[UserResponse])
async def list_users(
    skip: int = 0,
    limit: int = 20,
    db=Depends(get_db),
):
    """List all users with pagination."""
    service = UserService(db)
    return await service.get_users(skip=skip, limit=limit)

@router.get("/{user_id}", response_model=UserResponse)
async def get_user(user_id: int, db=Depends(get_db)):
    """Get a single user by ID."""
    service = UserService(db)
    user = await service.get_user(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@router.post("/", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate, db=Depends(get_db)):
    """Create a new user."""
    service = UserService(db)
    return await service.create_user(user)

@router.put("/{user_id}", response_model=UserResponse)
async def update_user(
    user_id: int,
    user: UserUpdate,
    db=Depends(get_db),
    current_user=Depends(get_current_user),
):
    """Update a user. Requires authentication."""
    service = UserService(db)
    return await service.update_user(user_id, user)

@router.delete("/{user_id}", status_code=204)
async def delete_user(
    user_id: int,
    db=Depends(get_db),
    current_user=Depends(get_current_user),
):
    """Delete a user. Requires authentication."""
    service = UserService(db)
    await service.delete_user(user_id)

# app/routers/items.py
from fastapi import APIRouter, Depends
from app.schemas.item import ItemCreate, ItemResponse
from app.dependencies import get_db

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

@router.get("/", response_model=list[ItemResponse])
async def list_items(db=Depends(get_db)):
    """List all items."""
    ...

@router.post("/", response_model=ItemResponse, status_code=201)
async def create_item(item: ItemCreate, db=Depends(get_db)):
    """Create a new item."""
    ...

# app/main.py
from fastapi import FastAPI
from app.routers import users, items

app = FastAPI(
    title="My Application API",
    version="1.0.0",
)

# Include routers
app.include_router(users.router)
app.include_router(items.router)

# You can also add a prefix when including
# app.include_router(users.router, prefix="/api/v1")

@app.get("/")
async def root():
    return {"message": "Welcome to the API", "docs": "/docs"}

# app/core/config.py
from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    """Application settings loaded from environment variables."""

    # Application
    app_name: str = "FastAPI Tutorial"
    debug: bool = False
    api_version: str = "v1"

    # Server
    host: str = "0.0.0.0"
    port: int = 8000
    workers: int = 4

    # Database
    database_url: str = "sqlite:///./app.db"

    # Security
    secret_key: str = "change-me-in-production"
    access_token_expire_minutes: int = 30
    algorithm: str = "HS256"

    # CORS
    allowed_origins: list[str] = ["http://localhost:3000"]

    model_config = {
        "env_file": ".env",
        "env_file_encoding": "utf-8",
        "case_sensitive": False,
    }

@lru_cache
def get_settings() -> Settings:
    """Cached settings instance. Call this instead of Settings() directly."""
    return Settings()

# .env file
APP_NAME=My Production API
DEBUG=false
DATABASE_URL=postgresql://user:pass@localhost:5432/mydb
SECRET_KEY=your-super-secret-key-here
ALLOWED_ORIGINS=["https://myapp.com","https://admin.myapp.com"]

# Using settings in your application
from fastapi import FastAPI, Depends
from app.core.config import Settings, get_settings

app = FastAPI()

@app.get("/info")
async def info(settings: Settings = Depends(get_settings)):
    return {
        "app_name": settings.app_name,
        "debug": settings.debug,
        "api_version": settings.api_version,
    }

# app/dependencies.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
from app.database import SessionLocal

# Database session dependency
def get_db():
    """Yield a database session, auto-close when done."""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Authentication dependency
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db),
):
    """Validate JWT token and return current user."""
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = decode_token(token)
        user_id = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except Exception:
        raise credentials_exception

    user = db.query(User).filter(User.id == user_id).first()
    if user is None:
        raise credentials_exception
    return user

# Role-based access dependency
def require_role(required_role: str):
    """Factory that creates a dependency requiring a specific role."""
    async def role_checker(current_user=Depends(get_current_user)):
        if current_user.role != required_role:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Role '{required_role}' required",
            )
        return current_user
    return role_checker

# Usage in routes:
# @router.delete("/users/{id}", dependencies=[Depends(require_role("admin"))])
# async def delete_user(id: int): ...

from contextlib import asynccontextmanager
from fastapi import FastAPI
import httpx

# Shared resources
http_client: httpx.AsyncClient | None = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Manage application lifecycle - startup and shutdown."""
    # --- STARTUP ---
    global http_client
    print("Starting up... initializing resources")

    # Create a shared HTTP client with connection pooling
    http_client = httpx.AsyncClient(
        base_url="https://api.example.com",
        timeout=30.0,
    )

    # Initialize database connection pool
    # await database.connect()

    # Load ML models, warm caches, etc.
    print("Startup complete")

    yield  # Application runs here

    # --- SHUTDOWN ---
    print("Shutting down... cleaning up resources")
    await http_client.aclose()
    # await database.disconnect()
    print("Shutdown complete")

app = FastAPI(lifespan=lifespan)

@app.get("/external-data")
async def get_external_data():
    """Use the shared HTTP client."""
    response = await http_client.get("/data")
    return response.json()

from fastapi import FastAPI, HTTPException, status

app = FastAPI()

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    if item_id < 0:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Item ID must be positive",
        )

    item = fake_db.get(item_id)
    if not item:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Item with ID {item_id} not found",
            headers={"X-Error": "item-not-found"},
        )

    return item

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()

# Custom exception class
class AppException(Exception):
    def __init__(self, status_code: int, detail: str, error_code: str):
        self.status_code = status_code
        self.detail = detail
        self.error_code = error_code

# Handler for custom exceptions
@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": {
                "code": exc.error_code,
                "message": exc.detail,
                "path": str(request.url),
            }
        },
    )

# Override default validation error handler for consistent format
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={
            "error": {
                "code": "VALIDATION_ERROR",
                "message": "Request validation failed",
                "details": exc.errors(),
            }
        },
    )

# Override default HTTP exception handler
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": {
                "code": f"HTTP_{exc.status_code}",
                "message": exc.detail,
            }
        },
    )

# Usage
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = await find_user(user_id)
    if not user:
        raise AppException(
            status_code=404,
            detail=f"User {user_id} not found",
            error_code="USER_NOT_FOUND",
        )
    return user

# Install Gunicorn (Linux/macOS only)
pip install gunicorn

# Run with Gunicorn + Uvicorn workers
gunicorn app.main:app \
    --workers 4 \
    --worker-class uvicorn.workers.UvicornWorker \
    --bind 0.0.0.0:8000 \
    --access-logfile - \
    --error-logfile -

# Dockerfile for FastAPI
# app/Dockerfile

FROM python:3.11-slim

WORKDIR /app

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

# Copy application code
COPY . .

# Run with Uvicorn (for single-container deployments)
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

# Build and run the Docker container
docker build -t fastapi-app .
docker run -p 8000:8000 fastapi-app

# Project structure
task-manager/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── config.py
│   ├── models.py
│   └── routers/
│       ├── __init__.py
│       └── tasks.py
├── requirements.txt
└── .env

# app/config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    app_name: str = "Task Manager API"
    version: str = "1.0.0"
    debug: bool = True

    model_config = {"env_file": ".env"}

settings = Settings()

# app/models.py
from pydantic import BaseModel, Field
from datetime import datetime
from enum import Enum

class Priority(str, Enum):
    low = "low"
    medium = "medium"
    high = "high"
    critical = "critical"

class TaskStatus(str, Enum):
    todo = "todo"
    in_progress = "in_progress"
    done = "done"

class TaskCreate(BaseModel):
    """Schema for creating a new task."""
    title: str = Field(..., min_length=1, max_length=200, examples=["Write unit tests"])
    description: str | None = Field(None, max_length=2000)
    priority: Priority = Field(default=Priority.medium)
    assignee: str | None = None

class TaskUpdate(BaseModel):
    """Schema for updating an existing task (all fields optional)."""
    title: str | None = Field(None, min_length=1, max_length=200)
    description: str | None = Field(None, max_length=2000)
    priority: Priority | None = None
    status: TaskStatus | None = None
    assignee: str | None = None

class TaskResponse(BaseModel):
    """Schema for task responses."""
    id: int
    title: str
    description: str | None
    priority: Priority
    status: TaskStatus
    assignee: str | None
    created_at: str
    updated_at: str

# app/routers/tasks.py
from fastapi import APIRouter, HTTPException, Query
from app.models import TaskCreate, TaskUpdate, TaskResponse, TaskStatus, Priority
from datetime import datetime

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

# In-memory storage (use a real database in production)
tasks_db: dict[int, dict] = {}
next_id: int = 1


@router.get("/", response_model=list[TaskResponse])
async def list_tasks(
    status: TaskStatus | None = Query(None, description="Filter by status"),
    priority: Priority | None = Query(None, description="Filter by priority"),
    assignee: str | None = Query(None, description="Filter by assignee"),
    skip: int = Query(0, ge=0),
    limit: int = Query(20, ge=1, le=100),
):
    """List all tasks with optional filters and pagination."""
    results = list(tasks_db.values())

    if status:
        results = [t for t in results if t["status"] == status]
    if priority:
        results = [t for t in results if t["priority"] == priority]
    if assignee:
        results = [t for t in results if t["assignee"] == assignee]

    return results[skip : skip + limit]


@router.get("/{task_id}", response_model=TaskResponse)
async def get_task(task_id: int):
    """Get a single task by ID."""
    if task_id not in tasks_db:
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
    return tasks_db[task_id]


@router.post("/", response_model=TaskResponse, status_code=201)
async def create_task(task: TaskCreate):
    """Create a new task."""
    global next_id
    now = datetime.utcnow().isoformat()

    task_data = {
        "id": next_id,
        **task.model_dump(),
        "status": TaskStatus.todo,
        "created_at": now,
        "updated_at": now,
    }
    tasks_db[next_id] = task_data
    next_id += 1
    return task_data


@router.patch("/{task_id}", response_model=TaskResponse)
async def update_task(task_id: int, task: TaskUpdate):
    """Partially update a task."""
    if task_id not in tasks_db:
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")

    stored = tasks_db[task_id]
    update_data = task.model_dump(exclude_unset=True)
    stored.update(update_data)
    stored["updated_at"] = datetime.utcnow().isoformat()
    return stored


@router.delete("/{task_id}", status_code=204)
async def delete_task(task_id: int):
    """Delete a task."""
    if task_id not in tasks_db:
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
    del tasks_db[task_id]


@router.get("/stats/summary")
async def task_stats():
    """Get task statistics."""
    tasks = list(tasks_db.values())
    return {
        "total": len(tasks),
        "by_status": {
            s.value: len([t for t in tasks if t["status"] == s])
            for s in TaskStatus
        },
        "by_priority": {
            p.value: len([t for t in tasks if t["priority"] == p])
            for p in Priority
        },
    }

# app/main.py
import time
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from app.config import settings
from app.routers import tasks

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Application lifecycle manager."""
    print(f"Starting {settings.app_name} v{settings.version}")
    yield
    print("Shutting down...")

app = FastAPI(
    title=settings.app_name,
    version=settings.version,
    description="A task management API built with FastAPI",
    lifespan=lifespan,
)

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

# Request timing middleware
@app.middleware("http")
async def add_process_time(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    duration = time.perf_counter() - start
    response.headers["X-Process-Time"] = f"{duration:.4f}"
    return response

# Include routers
app.include_router(tasks.router)

@app.get("/", tags=["Health"])
async def root():
    """API root - health check and documentation links."""
    return {
        "app": settings.app_name,
        "version": settings.version,
        "status": "healthy",
        "docs": "/docs",
        "redoc": "/redoc",
    }

@app.get("/health", tags=["Health"])
async def health():
    """Detailed health check."""
    return {"status": "healthy", "version": settings.version}

# requirements.txt
fastapi[standard]>=0.115.0
pydantic-settings>=2.0.0
uvicorn[standard]>=0.30.0

# From the project root
uvicorn app.main:app --reload

# Test the API
curl -X POST http://127.0.0.1:8000/tasks \
  -H "Content-Type: application/json" \
  -d '{"title": "Learn FastAPI", "priority": "high"}'

curl http://127.0.0.1:8000/tasks
curl http://127.0.0.1:8000/tasks?status=todo&priority=high
curl http://127.0.0.1:8000/tasks/stats/summary
curl http://127.0.0.1:8000/docs