FastAPI - Building Lightning-Fast APIs with Python

FastAPI is a modern, high-performance web framework for building APIs with Python. It's incredibly fast (comparable to NodeJS and Go), easy to learn, and comes with automatic API documentation. Let's build a complete API to see why developers love it!

🚀 Why FastAPI?

FastAPI has taken the Python world by storm, and for good reason:

• ⚡ Blazingly Fast: One of the fastest Python frameworks available
• 🎯 Type Safety: Built on Python type hints for automatic validation
• 📚 Auto Documentation: Interactive API docs (Swagger UI) generated automatically
• 🔒 Security: Built-in support for OAuth2, JWT tokens, and more
• 🎨 Modern Python: Uses async/await for concurrent operations
• ⚙️ Less Code: Reduce code duplication by up to 40%

Performance Comparison

FastAPI is on par with Node.js and Go:

• FastAPI: ~20,000-30,000 requests/second
• Flask: ~3,000-5,000 requests/second
• Django: ~2,000-4,000 requests/second

📦 Installation & Setup

Let's create a virtual environment and install FastAPI:

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

# Create virtual environment
python -m venv venv

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

# Install FastAPI and Uvicorn (ASGI server)
pip install fastapi uvicorn[standard]
pip install sqlalchemy pydantic-settings python-multipart

🏗️ Project Structure

Create a clean, organized structure:

fastapi-blog/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── models.py
│   ├── schemas.py
│   ├── database.py
│   └── routers/
│       ├── __init__.py
│       └── posts.py
├── venv/
└── requirements.txt

🎯 Building a Blog API

Let's build a complete blog API with CRUD operations.

Step 1: Database Configuration

Create app/database.py:

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# SQLite database
SQLALCHEMY_DATABASE_URL = "sqlite:///./blog.db"

# Create engine
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, 
    connect_args={"check_same_thread": False}
)

# Create session
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Base class for models
Base = declarative_base()

# Dependency to get database session
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Step 2: Database Models

Create app/models.py:

from sqlalchemy import Column, Integer, String, Text, DateTime
from sqlalchemy.sql import func
from app.database import Base

class Post(Base):
    __tablename__ = "posts"
    
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(200), nullable=False)
    content = Column(Text, nullable=False)
    author = Column(String(100), nullable=False)
    published = Column(Integer, default=1)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

Step 3: Pydantic Schemas

Create app/schemas.py for request/response validation:

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

# Base schema
class PostBase(BaseModel):
    title: str = Field(..., min_length=3, max_length=200)
    content: str = Field(..., min_length=10)
    author: str = Field(..., min_length=2, max_length=100)
    published: int = Field(default=1, ge=0, le=1)

# Schema for creating posts
class PostCreate(PostBase):
    pass

# Schema for updating posts
class PostUpdate(BaseModel):
    title: Optional[str] = Field(None, min_length=3, max_length=200)
    content: Optional[str] = Field(None, min_length=10)
    author: Optional[str] = None
    published: Optional[int] = Field(None, ge=0, le=1)

# Schema for responses
class PostResponse(PostBase):
    id: int
    created_at: datetime
    updated_at: Optional[datetime]
    
    class Config:
        from_attributes = True

💡 Key Concept: Pydantic models automatically validate data and convert types!

Step 4: API Routes

Create app/routers/posts.py:

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List
from app import models, schemas
from app.database import get_db

router = APIRouter(
    prefix="/api/posts",
    tags=["Posts"]
)

# Get all posts
@router.get("/", response_model=List[schemas.PostResponse])
async def get_posts(
    skip: int = 0,
    limit: int = 10,
    db: Session = Depends(get_db)
):
    """
    Retrieve all blog posts with pagination
    
    - **skip**: Number of posts to skip (default: 0)
    - **limit**: Maximum number of posts to return (default: 10)
    """
    posts = db.query(models.Post).offset(skip).limit(limit).all()
    return posts

# Get single post
@router.get("/{post_id}", response_model=schemas.PostResponse)
async def get_post(post_id: int, db: Session = Depends(get_db)):
    """
    Retrieve a specific blog post by ID
    """
    post = db.query(models.Post).filter(models.Post.id == post_id).first()
    
    if not post:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Post with ID {post_id} not found"
        )
    
    return post

# Create post
@router.post("/", response_model=schemas.PostResponse, status_code=status.HTTP_201_CREATED)
async def create_post(
    post: schemas.PostCreate,
    db: Session = Depends(get_db)
):
    """
    Create a new blog post
    """
    new_post = models.Post(**post.dict())
    db.add(new_post)
    db.commit()
    db.refresh(new_post)
    return new_post

# Update post
@router.put("/{post_id}", response_model=schemas.PostResponse)
async def update_post(
    post_id: int,
    post_update: schemas.PostUpdate,
    db: Session = Depends(get_db)
):
    """
    Update an existing blog post
    """
    post_query = db.query(models.Post).filter(models.Post.id == post_id)
    post = post_query.first()
    
    if not post:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Post with ID {post_id} not found"
        )
    
    # Update only provided fields
    update_data = post_update.dict(exclude_unset=True)
    post_query.update(update_data, synchronize_session=False)
    db.commit()
    db.refresh(post)
    
    return post

# Delete post
@router.delete("/{post_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_post(post_id: int, db: Session = Depends(get_db)):
    """
    Delete a blog post
    """
    post_query = db.query(models.Post).filter(models.Post.id == post_id)
    post = post_query.first()
    
    if not post:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Post with ID {post_id} not found"
        )
    
    post_query.delete(synchronize_session=False)
    db.commit()
    
    return None

# Search posts
@router.get("/search/", response_model=List[schemas.PostResponse])
async def search_posts(
    q: str,
    db: Session = Depends(get_db)
):
    """
    Search blog posts by title or content
    """
    posts = db.query(models.Post).filter(
        (models.Post.title.contains(q)) | 
        (models.Post.content.contains(q))
    ).all()
    
    return posts

Step 5: Main Application

Create app/main.py:

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.database import engine
from app import models
from app.routers import posts

# Create database tables
models.Base.metadata.create_all(bind=engine)

# Initialize FastAPI
app = FastAPI(
    title="Blog API",
    description="A modern blog API built with FastAPI",
    version="1.0.0",
    docs_url="/docs",  # Swagger UI
    redoc_url="/redoc"  # ReDoc
)

# Configure CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # In production, specify actual origins
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Include routers
app.include_router(posts.router)

# Root endpoint
@app.get("/")
async def root():
    return {
        "message": "Welcome to Blog API",
        "docs": "/docs",
        "version": "1.0.0"
    }

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

🚀 Running the Application

Start your FastAPI server:

uvicorn app.main:app --reload

Your API is now running at:

• API: http://localhost:8000
• Interactive Docs (Swagger): http://localhost:8000/docs
• Alternative Docs (ReDoc): http://localhost:8000/redoc

🧪 Testing the API

Using curl

Create a post:

curl -X POST "http://localhost:8000/api/posts/" \
  -H "Content-Type: application/json" \
  -d '{
    "title": "Getting Started with FastAPI",
    "content": "FastAPI is an amazing framework for building APIs quickly!",
    "author": "Shutiye Dev",
    "published": 1
  }'

Get all posts:

curl "http://localhost:8000/api/posts/"

Get a specific post:

curl "http://localhost:8000/api/posts/1"

Update a post:

curl -X PUT "http://localhost:8000/api/posts/1" \
  -H "Content-Type: application/json" \
  -d '{
    "title": "FastAPI Advanced Guide",
    "content": "Updated content about FastAPI features"
  }'

Search posts:

curl "http://localhost:8000/api/posts/search/?q=FastAPI"

Using Python Requests

import requests

# Create a post
response = requests.post(
    "http://localhost:8000/api/posts/",
    json={
        "title": "Python FastAPI Tutorial",
        "content": "Learn FastAPI step by step",
        "author": "Dev Team"
    }
)
print(response.json())

# Get all posts
response = requests.get("http://localhost:8000/api/posts/")
print(response.json())

✨ Advanced Features

1. Async Database Operations

For better performance, use async database operations:

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

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

engine = create_async_engine(DATABASE_URL)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_async_db():
    async with AsyncSessionLocal() as session:
        yield session

2. Authentication with JWT

from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@router.get("/protected")
async def protected_route(token: str = Depends(oauth2_scheme)):
    # Verify token
    return {"message": "This is protected!"}

3. File Upload

from fastapi import File, UploadFile

@router.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    contents = await file.read()
    # Process file
    return {"filename": file.filename, "size": len(contents)}

4. Background Tasks

from fastapi import BackgroundTasks

def send_email(email: str, message: str):
    # Send email logic
    print(f"Sending email to {email}: {message}")

@router.post("/send-notification")
async def send_notification(
    email: str,
    background_tasks: BackgroundTasks
):
    background_tasks.add_task(send_email, email, "Post created!")
    return {"message": "Notification will be sent"}

🎯 Best Practices

1. Use Dependency Injection

from fastapi import Depends

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

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

2. Implement Proper Error Handling

from fastapi import HTTPException

@router.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id not in items:
        raise HTTPException(
            status_code=404,
            detail="Item not found",
            headers={"X-Error": "Custom header"}
        )
    return items[item_id]

3. Use Environment Variables

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str
    secret_key: str
    
    class Config:
        env_file = ".env"

settings = Settings()

4. Add Response Models

@router.get("/users/", response_model=List[UserResponse])
async def get_users():
    return users  # Automatically validates and serializes

📊 Performance Tips

1. Use async/await for I/O operations
2. Implement caching with Redis
3. Add database connection pooling
4. Use pagination for large datasets
5. Optimize database queries with proper indexing
6. Compress responses with gzip

🚀 Deployment

Using Docker

Create Dockerfile:

FROM python:3.11-slim

WORKDIR /app

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

COPY ./app ./app

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

Build and run:

docker build -t fastapi-blog .
docker run -p 8000:8000 fastapi-blog

Using Uvicorn in Production

uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4

🎓 Key Takeaways

• FastAPI is fast: Performance comparable to Node.js and Go
• Type hints: Automatic validation and documentation
• Modern Python: Async/await support out of the box
• Great DX: Amazing developer experience with auto-completion
• Production-ready: Built-in features for security and performance

📚 What's Next?

• Add authentication with OAuth2 and JWT
• Implement WebSocket for real-time features
• Add GraphQL support
• Integrate with Celery for background tasks
• Set up monitoring with Prometheus
• Deploy to AWS, Azure, or Google Cloud

FastAPI is perfect for building modern, high-performance APIs. Its simplicity and speed make it an excellent choice for both beginners and experienced developers!

---

Want to learn more about FastAPI? Check out the official documentation or explore our Python tutorials.