FastAPI from Scratch: Hands-On Tutorial:

What is FastAPI?

FastAPI is a modern, high-performance web framework for building APIs with Python. It was created by Sebastián Ramírez and first released in 2018. Built on top of Starlette for the web parts and Pydantic for the data parts, FastAPI leverages Python's type hints to provide automatic request validation, serialization, and interactive API documentation out of the box.

At its core, FastAPI is designed to be fast — both in terms of development speed and runtime performance. It rivals Node.js and Go in benchmarks thanks to its asynchronous capabilities built on top of ASGI (Asynchronous Server Gateway Interface) standards, using uvloop and httptools under the hood.

Key Features

Automatic interactive API docs — Swagger UI and ReDoc are generated automatically
Type-based validation and serialization — Uses Pydantic models to define request/response schemas
Async-first design — Built for asynchronous endpoints using async/await
Dependency injection system — Clean, reusable dependencies across routes
Automatic OpenAPI and JSON Schema generation
Great editor support — Autocompletion and type checking everywhere

Why FastAPI Matters

🚀 Deploy your AI agent in 10 minutes

Managed Hermes hosting. Zero DevOps. 100M tokens/mo included.

Try it free →

In the landscape of Python web frameworks, FastAPI fills a crucial gap. While Flask and Django are excellent choices for many projects, they were primarily designed before Python's async capabilities matured. FastAPI brings several compelling advantages that make it stand out:

Performance: FastAPI applications can handle thousands of requests per second. The async architecture means your server can handle many concurrent connections without blocking, making it ideal for I/O-bound workloads like database queries, external API calls, and file operations.
Developer productivity: The combination of type hints and Pydantic models eliminates boilerplate code. You define your data shapes once, and FastAPI handles validation, parsing, and documentation automatically.
API documentation: Interactive Swagger UI and ReDoc are generated without any additional configuration. This means your API consumers always have up-to-date, accurate documentation.
Robustness: Type-based validation catches errors at development time rather than at runtime. Editor autocompletion reduces typos and logic errors significantly.

FastAPI is particularly well-suited for microservices, data science APIs, IoT backends, and any scenario where you need a lightweight, fast, and well-documented API layer.

Setting Up Your Environment

Before we dive into code, let's set up a proper development environment. We'll create a virtual environment and install the necessary packages.

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

# Create a virtual environment
python -m venv venv

# Activate the virtual environment
# On macOS/Linux:
source venv/bin/activate
# On Windows:
venv\Scripts\activate

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

# This installs:
# - fastapi: the core framework
# - uvicorn: lightning-fast ASGI server
# - pydantic: data validation library
# - python-multipart: for form and file handling
# - jinja2: for template rendering (used by admin docs)
# - and other useful dependencies

The [all] extra installs everything you'll likely need. For production, you might want to install only what's necessary, but for learning, the full installation is perfect.

Your First FastAPI Application

Let's create the classic "Hello, World!" application and then build from there. Create a file named main.py:

from fastapi import FastAPI

# Create the FastAPI app instance
app = FastAPI(
    title="My First API",
    description="A hands-on FastAPI tutorial",
    version="1.0.0"
)

@app.get("/")
async def root():
    """Return a friendly greeting."""
    return {"message": "Hello, World!"}

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

Now run the application using uvicorn:

# Run the development server with hot reload
uvicorn main:app --reload --host 0.0.0.0 --port 8000

Let's break down what's happening:

main:app — Points to the app object in the main.py file
--reload — Enables hot reloading during development (auto-restarts on code changes)
--host 0.0.0.0 — Binds to all network interfaces
--port 8000 — Listens on port 8000

Visit http://localhost:8000 to see your API response. Even better, navigate to http://localhost:8000/docs to see the automatically generated Swagger UI documentation, or http://localhost:8000/redoc for ReDoc. Try executing the endpoints directly from the browser!

Path Parameters and Query Parameters

FastAPI makes working with URL parameters incredibly intuitive using Python's type hints.

Path Parameters

Path parameters are parts of the URL path that capture variable values. They're defined using curly braces in the route decorator and corresponding function parameters:

from fastapi import FastAPI
from enum import Enum

app = FastAPI()

# Simple path parameter
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    """Fetch an item by its ID."""
    return {"item_id": item_id, "type": type(item_id).__name__}

# Multiple path parameters
@app.get("/users/{user_id}/posts/{post_id}")
async def read_user_post(user_id: int, post_id: int):
    """Fetch a specific post by a specific user."""
    return {
        "user_id": user_id,
        "post_id": post_id,
        "message": f"Post {post_id} from user {user_id}"
    }

# Path parameter with validation constraints
@app.get("/products/{product_id}")
async def get_product(
    product_id: int,
):
    """Get a product with validated ID constraints."""
    return {"product_id": product_id}

# Enum-based path parameter
class Category(str, Enum):
    electronics = "electronics"
    clothing = "clothing"
    books = "books"

@app.get("/category/{category}")
async def get_category_items(category: Category):
    """Get items filtered by a predefined category."""
    return {"category": category.value}

Notice how item_id: int automatically converts the string from the URL to an integer and returns a validation error if it can't be converted. The enum example restricts the category to only valid options.

Query Parameters

Query parameters are key-value pairs after the ? in URLs. Any function parameter that isn't part of the path is automatically treated as a query parameter:

from fastapi import FastAPI, Query
from typing import Optional, List

app = FastAPI()

@app.get("/search")
async def search_items(
    q: str,
    page: int = 1,
    limit: int = 10,
    sort: Optional[str] = None
):
    """Search items with pagination and optional sorting."""
    results = {
        "query": q,
        "page": page,
        "limit": limit,
        "sort": sort,
        "offset": (page - 1) * limit
    }
    return results

# Query parameters with validation
@app.get("/items/")
async def list_items(
    skip: int = Query(0, ge=0, description="Number of items to skip"),
    limit: int = Query(10, ge=1, le=100, description="Max items to return"),
    min_price: Optional[float] = Query(None, ge=0.0),
    tags: List[str] = Query([], description="Filter by tags")
):
    """List items with advanced filtering and validation."""
    return {
        "skip": skip,
        "limit": limit,
        "min_price": min_price,
        "tags": tags
    }

Key points about query parameters:

Default values make parameters optional
Optional types allow None values
Query provides additional validation (ge=greater or equal, le=less or equal)
List types automatically parse repeated query parameters (like ?tags=a&tags=b)

Request Bodies and Pydantic Models

For POST, PUT, and PATCH requests, you typically send data in the request body. FastAPI uses Pydantic models to define, validate, and document request body schemas.

from fastapi import FastAPI, Path
from pydantic import BaseModel, Field, EmailStr, validator
from typing import Optional, List
from datetime import datetime
import uuid

app = FastAPI()

# Define Pydantic models for request/response schemas
class ItemBase(BaseModel):
    name: str = Field(..., min_length=1, max_length=100, description="Item name")
    description: Optional[str] = Field(None, max_length=500)
    price: float = Field(..., gt=0, description="Price must be greater than zero")
    tax: Optional[float] = Field(None, ge=0)
    tags: List[str] = Field(default_factory=list)
    
    @validator('name')
    def name_must_not_be_empty(cls, v):
        if not v.strip():
            raise ValueError('Name cannot be empty or whitespace only')
        return v.strip()
    
    class Config:
        schema_extra = {
            "example": {
                "name": "Widget Pro",
                "description": "A professional-grade widget",
                "price": 29.99,
                "tax": 2.99,
                "tags": ["widget", "pro"]
            }
        }

class ItemCreate(ItemBase):
    """Model for creating a new item."""
    pass

class ItemResponse(ItemBase):
    """Model for the response after creating/reading an item."""
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: Optional[datetime] = None
    
    class Config:
        orm_mode = True  # Enable reading data from ORM objects

# Simulated database
items_db = {}

@app.post("/items/", response_model=ItemResponse, status_code=201)
async def create_item(item: ItemCreate):
    """Create a new item in the database."""
    # Convert the input model to the response model
    item_dict = item.dict()
    item_dict['id'] = str(uuid.uuid4())
    item_dict['created_at'] = datetime.utcnow()
    items_db[item_dict['id']] = item_dict
    return item_dict

@app.get("/items/{item_id}", response_model=ItemResponse)
async def get_item(item_id: str):
    """Get a single item by its ID."""
    if item_id not in items_db:
        # This will be caught by our error handlers
        raise HTTPException(status_code=404, detail="Item not found")
    return items_db[item_id]

@app.put("/items/{item_id}", response_model=ItemResponse)
async def update_item(item_id: str, item: ItemCreate):
    """Update an existing item."""
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    stored_item = items_db[item_id]
    update_data = item.dict(exclude_unset=True)  # Only update provided fields
    stored_item.update(update_data)
    stored_item['updated_at'] = datetime.utcnow()
    return stored_item

Pydantic models provide:

Data validation: Types, ranges, string lengths are enforced
Custom validators: The @validator decorator allows complex business rules
Documentation: Field descriptions and examples appear in the API docs
Serialization: response_model controls exactly what the client receives
ORM mode: Models can read data directly from ORM objects like SQLAlchemy results

Handling Form Data and File Uploads

Not all data comes as JSON. FastAPI handles form submissions and file uploads gracefully using python-multipart.

from fastapi import FastAPI, Form, File, UploadFile
from typing import Optional, List
import shutil
from pathlib import Path

app = FastAPI()

# Create uploads directory
upload_dir = Path("uploads")
upload_dir.mkdir(exist_ok=True)

@app.post("/login")
async def login(
    username: str = Form(...),
    password: str = Form(...),
    remember_me: bool = Form(False)
):
    """Process a form-based login."""
    # NEVER log passwords in production!
    return {
        "username": username,
        "remember_me": remember_me,
        "password_length": len(password)
    }

@app.post("/upload")
async def upload_single_file(
    file: UploadFile = File(...),
    description: Optional[str] = Form(None)
):
    """Upload a single file with optional description."""
    # Save the uploaded file
    file_path = upload_dir / file.filename
    with open(file_path, "wb") as buffer:
        shutil.copyfileobj(file.file, buffer)
    
    return {
        "filename": file.filename,
        "content_type": file.content_type,
        "size_bytes": file_path.stat().st_size,
        "description": description,
        "saved_to": str(file_path)
    }

@app.post("/upload-multiple")
async def upload_multiple_files(
    files: List[UploadFile] = File(...),
    tags: List[str] = Form([])
):
    """Upload multiple files at once."""
    saved_files = []
    for file in files:
        file_path = upload_dir / f"upload_{file.filename}"
        with open(file_path, "wb") as buffer:
            shutil.copyfileobj(file.file, buffer)
        saved_files.append({
            "filename": file.filename,
            "path": str(file_path),
            "size": file_path.stat().st_size
        })
    
    return {
        "files_uploaded": len(saved_files),
        "tags": tags,
        "files": saved_files
    }

Important notes about file uploads:

UploadFile uses streaming, so large files don't consume all memory
The file's metadata (filename, content_type) is automatically extracted
For multiple files, use List[UploadFile]
Always validate file types and sizes in production to prevent abuse

Dependency Injection

FastAPI's dependency injection system is one of its most powerful features. It allows you to define reusable logic that automatically runs before your endpoint handlers.

from fastapi import FastAPI, Depends, HTTPException, Header
from typing import Optional, Generator
import time

app = FastAPI()

# --- Simple dependencies ---

def get_current_timestamp() -> float:
    """Return the current Unix timestamp."""
    return time.time()

async def get_client_info(
    user_agent: Optional[str] = Header(None)
) -> dict:
    """Extract client information from headers."""
    return {"user_agent": user_agent}

@app.get("/timestamp")
async def timestamp_endpoint(
    ts: float = Depends(get_current_timestamp),
    client: dict = Depends(get_client_info)
):
    """Endpoint that uses multiple dependencies."""
    return {
        "timestamp": ts,
        "readable_time": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(ts)),
        "client": client
    }

# --- Dependencies with state (generators) ---

class DatabaseConnection:
    """Simulated database connection."""
    def __init__(self):
        self.connected = False
    
    def connect(self):
        self.connected = True
        print("DB connected")
    
    def disconnect(self):
        self.connected = False
        print("DB disconnected")
    
    def query(self, sql: str):
        return f"Executed: {sql}"

# Simulated connection pool
connection_pool = []

async def get_db() -> Generator[DatabaseConnection, None, None]:
    """Dependency that provides a database connection and cleans up after."""
    conn = DatabaseConnection()
    conn.connect()
    connection_pool.append(conn)
    try:
        yield conn  # This is where the dependency value is passed
    finally:
        conn.disconnect()
        connection_pool.remove(conn)

@app.get("/data")
async def read_data(
    db: DatabaseConnection = Depends(get_db)
):
    """Endpoint using a database connection dependency."""
    result = db.query("SELECT * FROM items LIMIT 10")
    return {"result": result, "active_connections": len(connection_pool)}

# --- Dependencies with parameters ---

def pagination_params(
    page: int = 1,
    page_size: int = 10
) -> dict:
    """Create standardized pagination parameters."""
    if page < 1:
        page = 1
    if page_size < 1:
        page_size = 10
    if page_size > 100:
        page_size = 100
    return {
        "page": page,
        "page_size": page_size,
        "offset": (page - 1) * page_size
    }

@app.get("/paginated-items")
async def get_paginated_items(
    pagination: dict = Depends(pagination_params)
):
    """Get items with pagination from dependency."""
    return {
        "pagination": pagination,
        "items": [f"Item {i}" for i in range(
            pagination['offset'],
            pagination['offset'] + pagination['page_size']
        )]
    }

The dependency injection system enables:

Code reuse: Share authentication, database connections, pagination logic across routes
Testing: Dependencies can be overridden for unit tests
Clean separation of concerns: Business logic stays separate from request handling
Automatic cleanup: Generator-based dependencies ensure resources are properly released

Middleware and CORS

Middleware runs before and after every request, allowing you to modify requests and responses globally.

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
import time
import logging

app = FastAPI()

# --- Custom middleware ---

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    """Add X-Process-Time header to every response."""
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(round(process_time, 4))
    response.headers["X-Response-Timestamp"] = str(int(time.time()))
    return response

@app.middleware("http")
async def log_requests(request: Request, call_next):
    """Log all incoming requests."""
    logging.info(f"Request: {request.method} {request.url.path}")
    response = await call_next(request)
    logging.info(f"Response status: {response.status_code}")
    return response

# --- CORS configuration ---

app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "https://your-frontend-domain.com",
        "http://localhost:3000",  # For local development
    ],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS", "PATCH"],
    allow_headers=[
        "Content-Type",
        "Authorization",
        "X-Custom-Header",
    ],
    expose_headers=[
        "X-Process-Time",
        "X-Response-Timestamp",
    ],
    max_age=600,  # Cache preflight requests for 10 minutes
)

# --- Trusted host middleware ---

app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=[
        "localhost",
        "127.0.0.1",
        "*.example.com",  # Wildcard subdomain matching
    ]
)

@app.get("/")
async def main():
    return {"message": "Middleware demo"}

Middleware execution order matters — they run in the order they're added. Custom @app.middleware("http") decorators run before added middleware. Always configure CORS carefully — being too permissive can expose your API to security risks.

Error Handling

FastAPI provides structured error handling through HTTPException and custom exception handlers.

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import Union

app = FastAPI()

# --- Custom exception classes ---

class ItemNotFoundException(Exception):
    """Raised when an item is not found."""
    def __init__(self, item_id: str):
        self.item_id = item_id

class InsufficientFundsException(Exception):
    """Raised when a user has insufficient funds."""
    def __init__(self, required: float, available: float):
        self.required = required
        self.available = available
        self.shortfall = required - available

# --- Custom exception handlers ---

@app.exception_handler(ItemNotFoundException)
async def item_not_found_handler(request: Request, exc: ItemNotFoundException):
    """Handle item not found exceptions."""
    return JSONResponse(
        status_code=404,
        content={
            "error": "Item not found",
            "item_id": exc.item_id,
            "message": f"The item with ID '{exc.item_id}' does not exist."
        }
    )

@app.exception_handler(InsufficientFundsException)
async def insufficient_funds_handler(request: Request, exc: InsufficientFundsException):
    """Handle insufficient funds exceptions."""
    return JSONResponse(
        status_code=402,  # Payment Required
        content={
            "error": "Insufficient funds",
            "required": exc.required,
            "available": exc.available,
            "shortfall": exc.shortfall,
            "message": f"You need ${exc.shortfall:.2f} more to complete this transaction."
        }
    )

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    """Customize the default HTTP exception response."""
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": exc.detail,
            "path": request.url.path,
            "method": request.method
        }
    )

# --- Routes that use exceptions ---

@app.get("/items/{item_id}")
async def get_item(item_id: str):
    """Get an item, raising a custom exception if not found."""
    if item_id not in ["a1", "b2", "c3"]:  # Simulated lookup
        raise ItemNotFoundException(item_id=item_id)
    return {"item_id": item_id, "name": f"Item {item_id}"}

@app.post("/purchase/{item_id}")
async def purchase_item(
    item_id: str,
    user_balance: float = 10.0,
    item_price: float = 25.0
):
    """Attempt to purchase an item, raising exception if funds insufficient."""
    if user_balance < item_price:
        raise InsufficientFundsException(
            required=item_price,
            available=user_balance
        )
    return {
        "item_id": item_id,
        "purchased": True,
        "new_balance": user_balance - item_price
    }

# --- Validation error handling ---

from fastapi.exceptions import RequestValidationError
from fastapi.encoders import jsonable_encoder

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    """Customize validation error responses."""
    return JSONResponse(
        status_code=422,
        content=jsonable_encoder({
            "error": "Validation failed",
            "details": exc.errors(),
            "body": exc.body
        })
    )

Best practices for error handling:

Create domain-specific exception classes for business logic errors
Register handlers for each exception type to maintain consistent response formats
Use HTTPException for standard HTTP error responses
Always provide helpful error messages that guide the client toward fixing the issue

Database Integration with SQLAlchemy

FastAPI works excellently with SQLAlchemy for database operations. Here's a complete async integration using SQLAlchemy 2.0 with async support:

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import (
    AsyncSession, create_async_engine, AsyncEngine,
    async_sessionmaker
)
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from sqlalchemy import String, Integer, Float, DateTime, ForeignKey, select, func
from sqlalchemy.ext.asyncio import AsyncAttrs
from typing import Optional, List, AsyncGenerator
from pydantic import BaseModel, Field, ConfigDict
from datetime import datetime
import uuid

# --- Database setup ---

DATABASE_URL = "sqlite+aiosqlite:///./tutorial.db"

engine: AsyncEngine = create_async_engine(
    DATABASE_URL,
    echo=False,  # Set to True for SQL logging
)

async_session_factory = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False
)

# --- SQLAlchemy models ---

class Base(AsyncAttrs, DeclarativeBase):
    pass

class UserModel(Base):
    __tablename__ = "users"
    
    id: Mapped[str] = mapped_column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
    email: Mapped[str] = mapped_column(String, unique=True, index=True)
    name: Mapped[str] = mapped_column(String(100))
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
    
    # Relationship
    posts: Mapped[List["PostModel"]] = relationship(
        "PostModel", back_populates="author", cascade="all, delete-orphan"
    )

class PostModel(Base):
    __tablename__ = "posts"
    
    id: Mapped[str] = mapped_column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
    title: Mapped[str] = mapped_column(String(200))
    content: Mapped[str] = mapped_column(String)
    user_id: Mapped[str] = mapped_column(String, ForeignKey("users.id"))
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
    
    author: Mapped["UserModel"] = relationship("UserModel", back_populates="posts")

# --- Pydantic schemas ---

class UserCreate(BaseModel):
    email: str
    name: str = Field(min_length=1, max_length=100)

class UserResponse(BaseModel):
    id: str
    email: str
    name: str
    created_at: datetime
    model_config = ConfigDict(from_attributes=True)

class PostCreate(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    content: str

class PostResponse(BaseModel):
    id: str
    title: str
    content: str
    user_id: str
    created_at: datetime
    model_config = ConfigDict(from_attributes=True)

# --- FastAPI app ---

app = FastAPI(title="SQLAlchemy Async Tutorial")

# --- Database dependency ---

async def get_session() -> AsyncGenerator[AsyncSession, None]:
    """Provide an async database session."""
    async with async_session_factory() as session:
        try:
            yield session
        finally:
            await session.close()

# --- Startup/shutdown events ---

@app.on_event("startup")
async def startup():
    """Create tables on application startup."""
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

@app.on_event("shutdown")
async def shutdown():
    """Dispose of the engine on shutdown."""
    await engine.dispose()

# --- CRUD endpoints ---

@app.post("/users/", response_model=UserResponse, status_code=201)
async def create_user(
    user: UserCreate,
    session: AsyncSession = Depends(get_session)
):
    """Create a new user."""
    # Check for duplicate email
    result = await session.execute(
        select(UserModel).where(UserModel.email == user.email)
    )
    if result.scalar_one_or_none():
        raise HTTPException(status_code=409, detail="Email already exists")
    
    db_user = UserModel(**user.model_dump())
    session.add(db_user)
    await session.commit()
    await session.refresh(db_user)
    return db_user

@app.get("/users/", response_model=List[UserResponse])
async def list_users(
    skip: int = 0,
    limit: int = 100,
    session: AsyncSession = Depends(get_session)
):
    """List all users with pagination."""
    result = await session.execute(
        select(UserModel).offset(skip).limit(limit)
    )
    return result.scalars().all()

@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: str,
    session: AsyncSession = Depends(get_session)
):
    """Get a single user by ID."""
    result = await session.execute(
        select(UserModel).where(UserModel.id == user_id)
    )
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.delete("/users/{user_id}", status_code=204)
async def delete_user(
    user_id: str,
    session: AsyncSession = Depends(get_session)
):
    """Delete a user and all their posts (cascade)."""
    result = await session.execute(
        select(UserModel).where(UserModel.id == user_id)
    )
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    await session.delete(user)
    await session.commit()
    return None

@app.post("/users/{user_id}/posts/", response_model=PostResponse, status_code=201)
async def create_post(
    user_id: str,
    post: PostCreate,
    session: AsyncSession = Depends(get_session)
):
    """Create a post for a specific user."""
    # Verify user exists
    result = await session.execute(
        select(UserModel).where(UserModel.id == user_id)
    )
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    
    db_post = PostModel(**post.model_dump(), user_id=user_id)
    session.add(db_post)
    await session.commit()
    await session.refresh(db_post)
    return db_post

Key patterns in this integration:

Async sessions: The get_session dependency provides a fresh session for each request
Model separation: SQLAlchemy models (database) are separate from Pydantic models (API)
from_attributes=True: Allows Pydantic to read from ORM objects directly
Lifecycle events: Tables are created on startup, connections cleaned up on shutdown
Proper error handling: 404 for missing resources, 409 for conflicts

Authentication and Security

FastAPI provides built-in security utilities for implementing authentication. Here's a complete JWT-based authentication system:

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import (
    OAuth2PasswordBearer, OAuth2PasswordRequestForm,
    HTTPBearer, HTTPAuthorizationCredentials
)
from passlib.context import CryptContext
from jose import JWTError, jwt
from pydantic import BaseModel
from typing import Optional
from datetime import datetime, timedelta

app = FastAPI()

# --- Configuration ---

SECRET_KEY = "your-secret-key-change-in-production-use-env-vars"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# --- Password hashing ---

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain_password: str, hashed_password: str) -> bool:
    """Verify a plain password against a hash."""
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    """Hash a password."""
    return pwd_context.hash(password)

# --- Simulated user database ---

class UserInDB(BaseModel):
    username: str
    email: str
    full_name: str
    hashed_password: str
    disabled: bool = False
    role: str = "user"

fake_users_db = {
    "alice": {
        "username": "alice",
        "email": "alice@example.com",
        "full_name": "Alice Johnson",
        "hashed_password": get_password_hash("secret123"),
        "disabled": False,
        "role": "admin"
    },
    "bob": {
        "username": "bob",
        "email": "bob@example.com",
        "full_name": "Bob Smith",
        "hashed_password": get_password_hash("password456"),
        "disabled": False,
        "role": "