Fastapi - asynchronous framework for API

What is FastAPI

FastAPI — a modern high‑performance web framework for building APIs in Python. Developed by Sebastián Ramírez, this framework is built on OpenAPI and JSON Schema standards, providing automatic documentation generation and data typing.

FastAPI leverages the latest Python features, including type hints and asynchronous programming, making it one of the fastest Python frameworks for API development. In terms of performance it rivals Node.js and Go, making it an excellent choice for high‑load applications.

Key Features of FastAPI

High Performance

FastAPI is built on Starlette for the web layer and Pydantic for data validation. This delivers performance comparable to the fastest Python frameworks. The framework fully supports ASGI (Asynchronous Server Gateway Interface), allowing it to handle thousands of concurrent connections.

Automatic Documentation Generation

One of the main features of FastAPI is automatic generation of interactive API documentation. The framework creates documentation in OpenAPI (Swagger) format and provides two interfaces:

• Swagger UI — available at /docs
• ReDoc — available at /redoc

Full Type Support

FastAPI uses standard Python type annotations for data validation, serialization, and automatic documentation generation. This ensures excellent IDE autocomplete support and reduces the number of bugs during development.

Built‑in Data Validation

Thanks to integration with Pydantic, FastAPI automatically validates incoming data, converts types, and returns clear error messages. This greatly simplifies development and improves application reliability.

Asynchronous Out of the Box

FastAPI fully supports asynchronous programming with async/await. This enables the creation of high‑throughput applications capable of handling many simultaneous requests.

Installation and Basic Setup

Installing FastAPI

To install FastAPI and all required dependencies, run the following command:

pip install fastapi[all]

This command installs FastAPI together with Uvicorn (ASGI server) and other useful dependencies.

For a minimal installation, use:

pip install fastapi
pip install uvicorn[standard]

Creating Your First Application

Create a file main.py with a basic app:

from fastapi import FastAPI

app = FastAPI(
    title="My FastAPI Project",
    description="This is an example FastAPI application",
    version="1.0.0"
)

@app.get("/")
async def root():
    return {"message": "Welcome to FastAPI!"}

@app.get("/health")
async def health_check():
    return {"status": "ok"}

Running the Application

Start the development server with Uvicorn:

uvicorn main:app --reload --host 0.0.0.0 --port 8000

Run options:

• --reload — automatic reload on code changes
• --host 0.0.0.0 — accept connections from any IP address
• --port 8000 — port for the server

Working with HTTP Methods

FastAPI supports all standard HTTP methods via corresponding decorators:

GET Requests

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

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

POST Requests

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    price: float
    is_offer: bool = False

@app.post("/items/")
async def create_item(item: Item):
    return {"item": item, "message": "Item created"}

PUT and PATCH Requests

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    return {"item_id": item_id, "item": item}

@app.patch("/items/{item_id}")
async def update_item_partial(item_id: int, item: dict):
    return {"item_id": item_id, "updated_fields": item}

DELETE Requests

@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    return {"message": f"Item {item_id} deleted"}

Working with Query Parameters

Path Parameters

from fastapi import Path

@app.get("/items/{item_id}")
async def get_item(
    item_id: int = Path(..., title="Item ID", ge=1)
):
    return {"item_id": item_id}

Query Parameters

from fastapi import Query

@app.get("/items/")
async def get_items(
    q: str = Query(None, min_length=3, max_length=50),
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100)
):
    return {"q": q, "skip": skip, "limit": limit}

Working with Headers

from fastapi import Header

@app.get("/items/")
async def get_items(
    user_agent: str = Header(None),
    x_token: str = Header(None, alias="X-Token")
):
    return {"User-Agent": user_agent, "X-Token": x_token}

Working with Cookies

from fastapi import Cookie

@app.get("/items/")
async def get_items(ads_id: str = Cookie(None)):
    return {"ads_id": ads_id}

Data Models with Pydantic

Creating Basic Models

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

class User(BaseModel):
    id: Optional[int] = None
    name: str = Field(..., min_length=1, max_length=100)
    email: str = Field(..., regex=r'^[\w\.-]+@[\w\.-]+\.\w+$')
    age: int = Field(..., ge=0, le=150)
    created_at: datetime = Field(default_factory=datetime.now)
    
    class Config:
        schema_extra = {
            "example": {
                "name": "Ivan Ivanov",
                "email": "ivan@example.com",
                "age": 30
            }
        }

Nested Models

from typing import List

class Address(BaseModel):
    street: str
    city: str
    country: str

class UserWithAddress(BaseModel):
    name: str
    email: str
    addresses: List[Address]

Response Models

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    created_at: datetime

@app.post("/users/", response_model=UserResponse)
async def create_user(user: User):
    # User creation logic
    return UserResponse(
        id=1,
        name=user.name,
        email=user.email,
        created_at=datetime.now()
    )

Asynchronous Programming

Async Handlers

import asyncio
import aiohttp

@app.get("/async-data")
async def get_async_data():
    async with aiohttp.ClientSession() as session:
        async with session.get('https://api.example.com/data') as response:
            data = await response.json()
            return data

@app.get("/delayed")
async def delayed_response():
    await asyncio.sleep(2)
    return {"message": "Response after 2 seconds"}

Working with Async Databases

import asyncpg
from databases import Database

DATABASE_URL = "postgresql://user:password@localhost/database"
database = Database(DATABASE_URL)

@app.on_event("startup")
async def startup():
    await database.connect()

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    query = "SELECT * FROM users WHERE id = :user_id"
    user = await database.fetch_one(query, {"user_id": user_id})
    return user

Error Handling

Standard HTTP Exceptions

from fastapi import HTTPException, status

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    if item_id == 0:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Item not found"
        )
    return {"item_id": item_id}

Custom Exception Handlers

from fastapi import Request
from fastapi.responses import JSONResponse

class CustomException(Exception):
    def __init__(self, name: str):
        self.name = name

@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
    return JSONResponse(
        status_code=418,
        content={"message": f"An error occurred: {exc.name}"}
    )

Dependencies and Dependency Injection

Simple Dependencies

from fastapi import Depends

async def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
async def get_items(commons: dict = Depends(common_parameters)):
    return commons

Stateful Dependencies

class DatabaseManager:
    def __init__(self):
        self.connection = None
    
    async def connect(self):
        # Connect to the database
        pass
    
    async def disconnect(self):
        # Disconnect from the database
        pass

db_manager = DatabaseManager()

async def get_db():
    return db_manager

@app.get("/items/")
async def get_items(db: DatabaseManager = Depends(get_db)):
    return {"database": "connected"}

Authentication and Authorization

OAuth2 with JWT

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from passlib.context import CryptContext
from jose import JWTError, jwt
from datetime import datetime, timedelta
from typing import Optional

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

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

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 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
    return username

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # User verification
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": form_data.username}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/protected")
async def protected_route(current_user: str = Depends(get_current_user)):
    return {"message": f"Hello {current_user}"}

Middleware and CORS

Adding CORS

from fastapi.middleware.cors import CORSMiddleware

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

Custom Middleware

from fastapi import Request
import time

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

Working with Databases

SQLAlchemy with FastAPI

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    email = Column(String, unique=True, index=True)

Base.metadata.create_all(bind=engine)

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

@app.post("/users/")
async def create_user(user: User, db: Session = Depends(get_db)):
    db_user = User(name=user.name, email=user.email)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

Project Structure

Recommended Structure for Large Projects

project/
├── main.py                 # Entry point
├── core/
│   ├── __init__.py
│   ├── config.py          # Configuration
│   ├── security.py        # Security
│   └── database.py        # DB connection
├── models/
│   ├── __init__.py
│   ├── user.py           # User models
│   └── item.py           # Item models
├── schemas/
│   ├── __init__.py
│   ├── user.py           # Pydantic schemas
│   └── item.py
├── api/
│   ├── __init__.py
│   ├── deps.py           # Dependencies
│   └── v1/
│       ├── __init__.py
│       ├── endpoints/
│       │   ├── __init__.py
│       │   ├── users.py
│       │   └── items.py
│       └── api.py
├── tests/
│   ├── __init__.py
│   └── test_main.py
└── requirements.txt

Using APIRouter

from fastapi import APIRouter

router = APIRouter(
    prefix="/api/v1",
    tags=["items"],
    responses={404: {"description": "Not found"}}
)

@router.get("/items/")
async def get_items():
    return [{"item_id": 1}]

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

# In main.py
app.include_router(router)

Testing FastAPI Applications

Basic Testing

from fastapi.testclient import TestClient
import pytest

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Welcome to FastAPI!"}

def test_create_item():
    response = client.post(
        "/items/",
        json={"name": "Test Item", "price": 10.5, "is_offer": True}
    )
    assert response.status_code == 200
    assert response.json()["item"]["name"] == "Test Item"

@pytest.mark.asyncio
async def test_async_endpoint():
    response = client.get("/async-data")
    assert response.status_code == 200

Testing with a Database

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(SQLALCHEMY_DATABASE_URL)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def override_get_db():
    try:
        db = TestingSessionLocal()
        yield db
    finally:
        db.close()

app.dependency_overrides[get_db] = override_get_db

@pytest.fixture
def client():
    with TestClient(app) as c:
        yield c

Deploying FastAPI Applications

Using Docker

FROM python:3.9

WORKDIR /app

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

COPY . .

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

Production with Gunicorn

pip install gunicorn
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

Nginx Configuration

server {
    listen 80;
    server_name yourdomain.com;
    
    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
    }
}

WebSocket Support

from fastapi import WebSocket

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message: {data}")

Background Tasks

from fastapi import BackgroundTasks

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

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

Complete FastAPI Methods and Functions Reference

Practical Usage Examples

E‑Commerce API

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel, Field
from typing import List, Optional
from datetime import datetime

app = FastAPI(title="E‑Commerce API")

class Product(BaseModel):
    id: Optional[int] = None
    name: str
    price: float
    description: str
    category_id: int

class Category(BaseModel):
    id: Optional[int] = None
    name: str

@app.get("/products/", response_model=List[Product])
async def get_products(category_id: Optional[int] = None, skip: int = 0, limit: int = 100):
    # Logic to retrieve products
    return []

@app.post("/products/", response_model=Product)
async def create_product(product: Product):
    # Logic to create a product
    return product

@app.get("/categories/", response_model=List[Category])
async def get_categories():
    # Logic to retrieve categories
    return []

Blog API

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

class Post(BaseModel):
    id: Optional[int] = None
    title: str
    content: str
    author_id: int
    created_at: datetime = Field(default_factory=datetime.now)
    published: bool = False

class Comment(BaseModel):
    id: Optional[int] = None
    post_id: int
    author_id: int
    content: str
    created_at: datetime = Field(default_factory=datetime.now)

@app.get("/posts/", response_model=List[Post])
async def get_posts(published: bool = True, skip: int = 0, limit: int = 10):
    return []

@app.post("/posts/", response_model=Post)
async def create_post(post: Post):
    return post

@app.get("/posts/{post_id}/comments/", response_model=List[Comment])
async def get_post_comments(post_id: int):
    return []

Frequently Asked Questions

What are the main differences between FastAPI and Flask?

FastAPI provides automatic data validation, documentation generation, built‑in type support, and asynchronous capabilities. Flask is more minimalistic and requires additional libraries for these features.

Does FastAPI support GraphQL?

Yes, FastAPI can be integrated with GraphQL using libraries such as Strawberry or Graphene.

Can FastAPI be used with Django ORM?

It is possible, but not recommended. Prefer SQLAlchemy, Tortoise ORM, or other async‑compatible ORMs.

How to secure a FastAPI application?

Use HTTPS, proper authentication (JWT, OAuth2), input validation, CORS settings, and keep dependencies up to date.

Does FastAPI include database migration tools?

FastAPI does not include migration tools. Use Alembic for SQLAlchemy or Aerich for Tortoise ORM.

Can FastAPI be used to build full‑stack web applications?

FastAPI is optimized for API development. For full‑stack web apps, frameworks like Django or Flask with templating are more suitable.

FastAPI — a powerful and modern framework that is perfect for building high‑performance APIs. Its ease of use, automatic documentation, and excellent performance make it an ideal choice for projects of any scale.

$