FastAPI for Modern APIs | learn.withkhent.com

Why FastAPI?

⚡

High Performance

FastAPI is one of the fastest Python frameworks available, on par with Node.js and Go.

📝

Automatic Docs

Generates interactive API documentation (Swagger UI and ReDoc) automatically.

✅

Type Checking

Based on Python type hints, providing automatic validation, serialization, and documentation.

FastAPI vs Other Frameworks

FastAPI is built on Starlette and Pydantic, offering significant advantages over other Python web frameworks:

Feature	FastAPI	Flask	Django
Performance	Very High (ASGI)	Medium (WSGI)	Medium (WSGI)
Async Support	Native	Limited	Limited
Type Hints	Core Feature	External (Pydantic/Marshmallow)	External (DRF Serializers)
Data Validation	Automatic	Manual	Forms/Serializers
API Documentation	Automatic (Swagger/ReDoc)	Manual (Extensions)	Manual (DRF)
Learning Curve	Easy	Easy	Steeper

Getting Started with FastAPI

Prerequisites

Before you start, make sure you have:

Python 3.7 or newer installed
Basic knowledge of Python programming
Understanding of RESTful API concepts
A text editor or IDE (VS Code, PyCharm, etc.)

Installation

The recommended way to install FastAPI is using pip within a virtual environment:

# Create a virtual environment
python -m venv venv

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

# Install FastAPI and Uvicorn (ASGI server)
pip install fastapi uvicorn[standard]
            

Tip: Uvicorn is an ASGI server required to run FastAPI applications. The [standard] extras include recommended optional dependencies.

Your First FastAPI Application

Let's create a simple "Hello, World!" API:

# main.py
from fastapi import FastAPI

# Create a FastAPI application instance
app = FastAPI()

# Define a route and handler function
@app.get("/")
async def root():
    return {"message": "Hello, World!"}
            

Save this code in a file named main.py and run it with Uvicorn:

uvicorn main:app --reload

You should now be able to visit http://127.0.0.1:8000/ in your browser and see a JSON response: {"message": "Hello, World!"}

Automatic Documentation

One of FastAPI's most powerful features is automatic interactive documentation. Visit:

FastAPI() creates a FastAPI application instance.
@app.get("/") is a decorator that tells FastAPI which URL path should trigger our function, and that it's a GET operation.
async def root() is an asynchronous function that will be called when a GET request is made to the root URL.
The function returns a dictionary that FastAPI automatically converts to JSON.

Tip: You can use async functions for endpoints that perform I/O operations (like database queries or HTTP requests) to improve performance, but it's not required - regular functions work fine too.

Path Parameters and Query Parameters

Path Parameters

Path parameters are parts of the URL path that are captured and passed to the function:

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}
            

In this example:

The {item_id} in the path is a path parameter.
The function parameter item_id with a type annotation int tells FastAPI to validate and convert the parameter to an integer.
If someone requests /items/42, FastAPI will pass item_id=42 to the function.
If someone requests /items/foo, FastAPI will return an error because "foo" can't be converted to an integer.

Query Parameters

Query parameters are the key-value pairs in the URL after the question mark (e.g., ?skip=0&limit=10):

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}
            

In this example:

The function parameters skip and limit are query parameters.
They have default values of 0 and 10 respectively.
If someone requests /items/?skip=20&limit=50, the function will receive skip=20 and limit=50.
If someone requests just /items/, the function will use the default values.

Optional Parameters

You can make parameters optional using Python's typing and default values:

from typing import Optional
from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int, q: Optional[str] = None):
    if q:
        return {"item_id": item_id, "q": q}
    return {"item_id": item_id}
            

In this example, q is an optional query parameter. It will be None if not provided in the URL.

Tip: Since Python 3.10, you can also use the | None syntax instead of Optional, like q: str | None = None.

Request Body

To receive data in the request body, you can declare a parameter with a Pydantic model type:

from fastapi import FastAPI
from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

app = FastAPI()

@app.post("/items/")
async def create_item(item: Item):
    return item
            

In this example:

The Item class is a Pydantic model that defines the structure of the request body.
FastAPI will automatically validate the incoming JSON against this model.
If the validation fails, it will return a detailed error message.
If the validation succeeds, the item parameter will be an instance of the Item class.

Data Validation with Pydantic Models

Pydantic Basics

Pydantic is a data validation library that uses Python type annotations. FastAPI uses Pydantic for:

Request body validation
Response data validation
Automatic documentation
Serialization and deserialization

Creating Pydantic Models

A Pydantic model is a class that inherits from BaseModel:

from pydantic import BaseModel, Field, EmailStr
from typing import List, Optional
from datetime import datetime

class User(BaseModel):
    id: int
    username: str
    email: EmailStr
    full_name: Optional[str] = None
    disabled: bool = False
    created_at: datetime = Field(default_factory=datetime.now)
    
    class Config:
        schema_extra = {
            "example": {
                "id": 1,
                "username": "johndoe",
                "email": "john@example.com",
                "full_name": "John Doe",
                "disabled": False
            }
        }
            

In this example:

id, username, and email are required fields.
full_name is an optional string (can be None).
disabled has a default value of False.
created_at has a default value of the current datetime.
The Config class with schema_extra provides an example for the documentation.

Nested Models

Pydantic models can be nested to represent complex data structures:

from pydantic import BaseModel
from typing import List, Optional

class Image(BaseModel):
    url: str
    name: str

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: List[str] = []
    images: Optional[List[Image]] = None
            

In this example, images is a list of Image objects.

Field Validation

Pydantic provides several ways to add constraints to fields:

from pydantic import BaseModel, Field, EmailStr, constr

class User(BaseModel):
    username: constr(min_length=3, max_length=50, regex="^[a-zA-Z0-9_-]+$")
    email: EmailStr
    password: str = Field(..., min_length=8)
    age: int = Field(..., gt=0, lt=120)
            

In this example:

username must be 3-50 characters and match the regex pattern.
email must be a valid email address.
password must be at least 8 characters long.
age must be between 1 and 119.
The ... in Field means the field is required and has no default value.

Response Models

You can use Pydantic models to define the structure of responses:

from fastapi import FastAPI
from pydantic import BaseModel
from typing import List

class Item(BaseModel):
    name: str
    price: float
    
class ItemList(BaseModel):
    items: List[Item]
    total: int

app = FastAPI()

@app.get("/items/", response_model=ItemList)
async def read_items():
    items = [
        {"name": "Foo", "price": 42.0},
        {"name": "Bar", "price": 24.0}
    ]
    return {"items": items, "total": len(items)}
            

The response_model parameter:

Validates the response data against the model
Filters out extra data not in the model
Documents the response structure in the API docs

Tip: Using response_model is a good way to ensure your API doesn't accidentally leak sensitive data.

Database Integration

SQLAlchemy with FastAPI

FastAPI works well with SQLAlchemy for database operations. Here's a basic setup:

# database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
# For PostgreSQL:
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@localhost/dbname"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
# The check_same_thread argument is needed only for SQLite

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

Base = declarative_base()
            

Defining Models

Define SQLAlchemy models for your database tables:

# models.py
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String, Float
from sqlalchemy.orm import relationship

from .database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

    items = relationship("Item", back_populates="owner")


class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String)
    price = Column(Float)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")
            

Creating Schemas

Define Pydantic models for data validation and serialization:

# schemas.py
from pydantic import BaseModel
from typing import List, Optional

class ItemBase(BaseModel):
    name: str
    description: Optional[str] = None
    price: float

class ItemCreate(ItemBase):
    pass

class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True

class UserBase(BaseModel):
    email: str

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: int
    is_active: bool
    items: List[Item] = []

    class Config:
        orm_mode = True
            

Note the orm_mode = True setting, which allows Pydantic to work with SQLAlchemy models.

Dependency Injection for Database Sessions

FastAPI's dependency injection system is perfect for managing database sessions:

# dependencies.py
from fastapi import Depends
from sqlalchemy.orm import Session

from . import database

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

CRUD Operations

Define functions to perform database operations:

# crud.py
from sqlalchemy.orm import Session

from . import models, schemas

def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()

def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()

def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()

def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed_password = user.password + "notreallyhashed"
    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def get_items(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.Item).offset(skip).limit(limit).all()

def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
    db_item = models.Item(**item.dict(), owner_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item
            

API Endpoints

Use everything together in your API endpoints:

# main.py
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import engine, SessionLocal
from .dependencies import get_db

models.Base.metadata.create_all(bind=engine)

app = FastAPI()

@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)

@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users

@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return crud.create_user_item(db=db, item=item, user_id=user_id)

@app.get("/items/", response_model=list[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items
            

Tip: The Depends function is a key part of FastAPI's dependency injection system. It allows you to declare dependencies that FastAPI will inject into your endpoint functions.

Authentication and Security

OAuth2 with Password Flow

FastAPI provides tools for implementing OAuth2 authentication:

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

# Setup
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

app = FastAPI()

# Models
class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    disabled: Optional[bool] = None

class UserInDB(User):
    hashed_password: str

class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    username: Optional[str] = None

# Fake database
fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "john@example.com",
        "hashed_password": pwd_context.hash("secret"),
        "disabled": False,
    }
}

# Helper functions
def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    return pwd_context.hash(password)

def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)

def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user

async def get_current_active_user(current_user: User = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

# Endpoints
@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/users/me/", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user
            

API Keys and Headers

For simpler authentication schemes, you can use API keys:

from fastapi import Depends, FastAPI, HTTPException, Security, status
from fastapi.security.api_key import APIKeyHeader, APIKey

API_KEY = "your-api-key"
API_KEY_NAME = "X-API-Key"

api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=False)

app = FastAPI()

async def get_api_key(api_key_header: str = Security(api_key_header)):
    if api_key_header == API_KEY:
        return api_key_header
    raise HTTPException(
        status_code=status.HTTP_403_FORBIDDEN, detail="Could not validate API key"
    )

@app.get("/secure-endpoint/")
async def secure_endpoint(api_key: APIKey = Depends(get_api_key)):
    return {"message": "You have accessed a secure endpoint"}
            

Security Note: In production, always use HTTPS and store sensitive values like API keys and passwords securely (e.g., using environment variables).

Advanced Features

Background Tasks

FastAPI allows you to run background tasks after returning a response:

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def write_notification(email: str, message=""):
    with open("log.txt", mode="a") as email_file:
        content = f"notification for {email}: {message}\n"
        email_file.write(content)

@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="Welcome!")
    return {"message": "Notification sent in the background"}
            

WebSockets

FastAPI supports WebSockets for real-time communication:

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active_connections: list[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def send_personal_message(self, message: str, websocket: WebSocket):
        await websocket.send_text(message)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.send_personal_message(f"You wrote: {data}", websocket)
            await manager.broadcast(f"Client #{client_id} says: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"Client #{client_id} left the chat")
            

CORS (Cross-Origin Resource Sharing)

Enable CORS to allow your API to be accessed from different domains:

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

origins = [
    "http://localhost:8000",
    "http://localhost:3000",  # React default port
    "https://example.com",
]

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

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

File Uploads

Handle file uploads with FastAPI:

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

app = FastAPI()

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    # Save the file
    path = Path(f"uploads/{file.filename}")
    with path.open("wb") as buffer:
        shutil.copyfileobj(file.file, buffer)
    
    return {"filename": file.filename, "content_type": file.content_type}

@app.post("/uploadfiles/")
async def create_upload_files(files: List[UploadFile] = File(...)):
    return {"filenames": [file.filename for file in files]}
            

Bigger Applications - Project Structure

For larger FastAPI applications, a modular structure is recommended:

my_app/
  |- main.py               # FastAPI application initialization
  |- dependencies.py       # Dependency injection functions
  |- routers/
      |- users.py          # User-related endpoints
      |- items.py          # Item-related endpoints
      |- auth.py           # Authentication endpoints
  |- models/
      |- user.py           # SQLAlchemy models
      |- item.py
  |- schemas/
      |- user.py           # Pydantic models
      |- item.py
  |- crud/
      |- user.py           # Database operations
      |- item.py
  |- config.py             # Configuration and settings
  |- database.py           # Database connection management
  |- tests/                # Unit and integration tests
            

In main.py, you can include routers from separate files:

from fastapi import FastAPI
from .routers import users, items, auth
from .config import settings

app = FastAPI(
    title=settings.PROJECT_NAME,
    description=settings.PROJECT_DESCRIPTION,
    version=settings.VERSION
)

app.include_router(users.router, prefix="/users", tags=["users"])
app.include_router(items.router, prefix="/items", tags=["items"])
app.include_router(auth.router, tags=["authentication"])
            

Additional Resources

Official FastAPI Documentation

Comprehensive and well-organized documentation for FastAPI.

Visit

FastAPI on GitHub

Source code and examples from the FastAPI repository.

Visit

Pydantic Documentation

Learn more about data validation with Pydantic.

Visit