Framework

FastAPI

Overview

FastAPI is a high-performance, modern web API framework that leverages Python 3.6+ type hints. It features automatic documentation generation and fast asynchronous processing.

Details

FastAPI was developed by Sebastián Ramírez in 2018 and is built on Pydantic and Starlette as its foundation. By maximizing the use of Python type hints, it achieves data validation, serialization, and automatic documentation generation. Based on ASGI (Asynchronous Server Gateway Interface) asynchronous architecture, it provides high performance comparable to Node.js and Go. It comes standard with features such as automatic generation of interactive API documentation compliant with OpenAPI (Swagger) specifications, automatic JSON schema creation, and request/response validation. It is adopted by large-scale companies such as Netflix, Microsoft, and Uber, and is optimized for the recent API-first approach trend. It is developed based on design principles of "Fast to code", "Few bugs", "Intuitive", "Easy", "Short", "Robust", and "Standards-based".

Merits and Demerits

Merits

• High performance: Speed comparable to Node.js and Go through ASGI asynchronous processing
• Type safety: Powerful type checking through Python type hints
• Automatic documentation: Automatic generation of interactive API documentation compliant with OpenAPI
• Development efficiency: High development productivity through editor support and auto-completion
• Validation: Automatic request/response validation through Pydantic
• Modern design: Compliance with latest web standards and best practices
• Rich ecosystem: Abundant extensions and community support

Demerits

• Relatively new: Less historical examples compared to Flask or Django
• Type hints required: Understanding of Python type system needed
• API-focused: Not suitable for traditional web application development
• Learning cost: Knowledge of asynchronous programming required
• Ecosystem: Some libraries have insufficient asynchronous support

Main Links

• FastAPI Official Site
• FastAPI Official Documentation
• FastAPI GitHub Repository
• FastAPI Tutorial
• FastAPI Community
• Awesome FastAPI

Code Examples

Hello World

# main.py
from fastapi import FastAPI

app = FastAPI(
    title="My API",
    description="This is a very fancy API",
    version="0.1.0",
)

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

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

# Startup commands
# pip install fastapi uvicorn
# fastapi dev main.py
# or
# uvicorn main:app --reload

Pydantic Models and Validation

from datetime import datetime
from typing import Optional, List
from pydantic import BaseModel, Field, EmailStr
from fastapi import FastAPI, HTTPException
from enum import Enum

app = FastAPI()

class ItemStatus(str, Enum):
    draft = "draft"
    published = "published"
    archived = "archived"

class UserBase(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr
    full_name: Optional[str] = None
    age: Optional[int] = Field(None, ge=0, le=150)

class UserCreate(UserBase):
    password: str = Field(..., min_length=8)

class UserResponse(UserBase):
    id: int
    created_at: datetime
    is_active: bool = True

    class Config:
        from_attributes = True

class ItemBase(BaseModel):
    title: str = Field(..., min_length=1, max_length=100)
    description: Optional[str] = Field(None, max_length=500)
    price: float = Field(..., gt=0)
    tax: Optional[float] = None
    status: ItemStatus = ItemStatus.draft

class ItemCreate(ItemBase):
    pass

class ItemResponse(ItemBase):
    id: int
    owner_id: int
    created_at: datetime
    updated_at: datetime

# API endpoints
@app.post("/users/", response_model=UserResponse)
async def create_user(user: UserCreate):
    # Password hashing and other processing
    user_dict = user.dict()
    user_dict["id"] = 1
    user_dict["created_at"] = datetime.now()
    return UserResponse(**user_dict)

@app.post("/items/", response_model=ItemResponse)
async def create_item(item: ItemCreate, user_id: int):
    item_dict = item.dict()
    item_dict.update({
        "id": 1,
        "owner_id": user_id,
        "created_at": datetime.now(),
        "updated_at": datetime.now()
    })
    return ItemResponse(**item_dict)

Path and Query Parameters

from fastapi import FastAPI, Query, Path, HTTPException
from typing import Optional, List
from enum import Enum

app = FastAPI()

class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"

# Path parameters
@app.get("/items/{item_id}")
async def read_item(
    item_id: int = Path(..., title="The ID of the item", ge=1),
    q: Optional[str] = Query(None, min_length=3, max_length=50)
):
    return {"item_id": item_id, "q": q}

# Path parameters with Enum
@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
    if model_name == ModelName.alexnet:
        return {"model_name": model_name, "message": "Deep Learning FTW!"}
    return {"model_name": model_name, "message": "Have some residuals"}

# Query parameters (multiple with validation)
@app.get("/items/")
async def read_items(
    q: Optional[str] = Query(
        None, 
        min_length=3, 
        max_length=50, 
        regex="^[a-zA-Z0-9 ]*$",
        title="Query string"
    ),
    skip: int = Query(0, ge=0),
    limit: int = Query(10, gt=0, le=100),
    tags: List[str] = Query([])
):
    results = {"skip": skip, "limit": limit}
    if q:
        results.update({"q": q})
    if tags:
        results.update({"tags": tags})
    return results

Authentication and Authorization

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

app = FastAPI()

# Security settings
SECRET_KEY = "your-secret-key-here"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

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

# Data models
class Token(BaseModel):
    access_token: str
    token_type: str

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

class UserInDB(User):
    hashed_password: str

# Fake user database
fake_users_db = {
    "testuser": {
        "username": "testuser",
        "full_name": "Test User",
        "email": "[email protected]",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
        "disabled": False,
    }
}

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

def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user or 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
    except JWTError:
        raise credentials_exception
    
    user = get_user(fake_users_db, username=username)
    if user is None:
        raise credentials_exception
    return user

# Authentication 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"}

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

Database Integration

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String, DateTime, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session, relationship
from sqlalchemy.sql import func
from pydantic import BaseModel
from datetime import datetime
from typing import List, Optional

# Database configuration
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# SQLAlchemy models
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    username = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    items = relationship("Item", back_populates="owner")

class Item(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    owner = relationship("User", back_populates="items")

Base.metadata.create_all(bind=engine)

app = FastAPI()

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

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

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

Testing

# test_main.py
import pytest
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_create_user():
    response = client.post(
        "/users/",
        json={"email": "[email protected]", "username": "testuser", "password": "testpass"},
    )
    assert response.status_code == 200
    data = response.json()
    assert data["email"] == "[email protected]"
    assert "id" in data

def test_read_item():
    response = client.get("/items/1?q=test")
    assert response.status_code == 200
    data = response.json()
    assert data["item_id"] == 1
    assert data["q"] == "test"

# Test execution commands
# pip install pytest httpx
# pytest