Building REST APIs with Python FastAPI — Complete Step-by-Step Guide
In this tutorial, you'll learn about Building REST APIs with Python FastAPI. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
FastAPI is a modern Python web framework for building REST APIs with automatic OpenAPI documentation, Pydantic validation, async support, and Dependency Injection that achieves Node.js-level performance.
What You'll Learn
You will build a complete REST API with Python FastAPI from scratch, including path operations, Pydantic models for validation, SQLAlchemy database integration, async endpoints, JWT authentication, and deployment.
Why FastAPI Matters
FastAPI is the fastest-growing Python web framework, used by Netflix, Microsoft, and Uber. It generates OpenAPI documentation automatically, validates request and response data with Pydantic, and supports asynchronous operations natively. Benchmarks show FastAPI handling up to 20,000 requests per second.
Real-World Use
DodaTech uses FastAPI for multiple services. DodaZIP update API is built with FastAPI for high-throughput update distribution, Durga Antivirus Pro uses FastAPI for its threat intelligence ingestion pipeline, and internal analytics services Process millions of events daily through FastAPI endpoints.
FastAPI Learning Path
flowchart LR
A[Python Basics] --> B[FastAPI Setup]
B --> C[Path Operations]
C --> D[Pydantic Models]
D --> E[Dependency Injection]
E --> F[SQLAlchemy & Async DB]
F --> G[Auth & Deployment]
B:::current
classDef current fill:#f90,color:#fff,stroke:#333,stroke-width:2px
Prerequisites
Understand RESTful Api Design Best Practices and intermediate Python Basics. Familiarity with HTTP Protocol Basics and JSON Data Format is required. Basic knowledge of SQL and databases helps.
Project Setup
Step 1: Create Project and Install Dependencies
mkdir dodatech-fastapi
cd dodatech-fastapi
python -m venv venv
source venv/bin/activate
Create requirements.txt:
fastapi==0.111.0
uvicorn[standard]==0.30.0
sqlalchemy==2.0.31
asyncpg==0.29.0
psycopg2-binary==2.9.9
alembic==1.13.1
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
python-multipart==0.0.9
pydantic==2.7.4
pydantic-settings==2.3.0
pip install -r requirements.txt
Step 2: Project Structure
dodatech-fastapi/
├── app/
│ ├── __init__.py
│ ├── main.py
│ ├── config.py
│ ├── database.py
│ ├── models/
│ │ ├── __init__.py
│ │ └── user.py
│ ├── schemas/
│ │ ├── __init__.py
│ │ └── user.py
│ ├── api/
│ │ ├── __init__.py
│ │ └── v1/
│ │ ├── __init__.py
│ │ ├── users.py
│ │ └── auth.py
│ ├── core/
│ │ ├── __init__.py
│ │ ├── security.py
│ │ └── dependencies.py
│ └── crud/
│ ├── __init__.py
│ └── user.py
├── alembic.ini
├── migrations/
└── requirements.txt
Building the Application
Step 1: Configuration
# app/config.py
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
app_name: str = "DodaTech API"
database_url: str = "postgresql+asyncpg://postgres:password@localhost:5432/dodatech"
secret_key: str = "your-secret-key-change-in-production"
algorithm: str = "HS256"
access_token_expire_minutes: int = 30
debug: bool = False
class Config:
env_file = ".env"
settings = Settings()
Step 2: Database Setup
# app/database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import DeclarativeBase, sessionmaker
from app.config import settings
engine = create_async_engine(settings.database_url, echo=settings.debug)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
class Base(DeclarativeBase):
pass
async def get_db():
async with async_session() as session:
try:
yield session
finally:
await session.close()
Step 3: SQLAlchemy Model
# app/models/user.py
from sqlalchemy import Column, String, Boolean, DateTime, Enum as SAEnum
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.sql import func
import uuid
import enum
from app.database import Base
class UserRole(str, enum.Enum):
ADMIN = "admin"
MEMBER = "member"
VIEWER = "viewer"
class User(Base):
__tablename__ = "users"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
name = Column(String(100), nullable=False)
email = Column(String(255), unique=True, nullable=False, index=True)
password_hash = Column(String(255), nullable=False)
role = Column(SAEnum(UserRole), default=UserRole.MEMBER, nullable=False)
is_active = Column(Boolean, default=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
Step 4: Pydantic Schemas
# app/schemas/user.py
from pydantic import BaseModel, EmailStr, UUID4
from datetime import datetime
from typing import Optional
from app.models.user import UserRole
class UserBase(BaseModel):
name: str
email: EmailStr
role: UserRole = UserRole.MEMBER
class UserCreate(UserBase):
password: str
class UserUpdate(BaseModel):
name: Optional[str] = None
email: Optional[EmailStr] = None
role: Optional[UserRole] = None
class UserResponse(UserBase):
id: UUID4
is_active: bool
created_at: datetime
updated_at: Optional[datetime] = None
model_config = {"from_attributes": True}
class UserListResponse(BaseModel):
success: bool
data: list[UserResponse]
total: int
page: int
limit: int
Step 5: CRUD Operations
# app/CRUD/user.py
from SQLAlchemy import select, func, delete
from SQLAlchemy.ext.asyncio import AsyncSession
from passlib.context import CryptContext
from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
async def get_users(db: AsyncSession, skip: int = 0, limit: int = 20):
query = select(User).offset(skip).limit(limit).order_by(User.created_at.desc())
result = await db.execute(query)
return result.scalars().all()
async def get_user_count(db: AsyncSession):
query = select(func.count(User.id))
result = await db.execute(query)
return result.scalar()
async def get_user_by_id(db: AsyncSession, user_id: UUID):
query = select(User).where(User.id == user_id)
result = await db.execute(query)
return result.scalar_one_or_none()
async def get_user_by_email(db: AsyncSession, email: str):
query = select(User).where(User.email == email)
result = await db.execute(query)
return result.scalar_one_or_none()
async def create_user(db: AsyncSession, user_in: UserCreate):
hashed_password = pwd_context.hash(user_in.password)
user = User(
name=user_in.name,
email=user_in.email,
password_hash=hashed_password,
role=user_in.role
)
db.add(user)
await db.commit()
await db.refresh(user)
return user
async def update_user(db: AsyncSession, user_id: UUID, user_in: UserUpdate):
user = await get_user_by_id(db, user_id)
if not user:
return None
update_data = user_in.model_dump(exclude_unset=True)
if "password" in update_data:
update_data["password_hash"] = pwd_context.hash(update_data.pop("password"))
for field, value in update_data.items():
setattr(user, field, value)
await db.commit()
await db.refresh(user)
return user
async def delete_user(db: AsyncSession, user_id: UUID):
user = await get_user_by_id(db, user_id)
if not user:
return False
await db.delete(user)
await db.commit()
return True
Step 6: API Routes
# app/API/v1/users.py
from FastAPI import APIRouter, Depends, HTTPException, Query
from SQLAlchemy.ext.asyncio import AsyncSession
from uuid import UUID
from app.database import get_db
from app.schemas.user import UserCreate, UserUpdate, UserResponse, UserListResponse
from app.CRUD import user as user_CRUD
router = APIRouter(prefix="/users", tags=["Users"])
@router.get("/", response_model=UserListResponse)
async def list_users(
page: int = Query(1, ge=1),
limit: int = Query(20, ge=1, le=100),
db: AsyncSession = Depends(get_db)
):
skip = (page - 1) * limit
users = await user_CRUD.get_users(db, skip=skip, limit=limit)
total = await user_CRUD.get_user_count(db)
return UserListResponse(
success=True,
data=users,
total=total,
page=page,
limit=limit
)
@router.get("/{user_id}", response_model=dict)
async def get_user(user_id: UUID, db: AsyncSession = Depends(get_db)):
user = await user_CRUD.get_user_by_id(db, user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return {"success": True, "data": UserResponse.model_validate(user)}
@router.post("/", response_model=dict, status_code=201)
async def create_user(user_in: UserCreate, db: AsyncSession = Depends(get_db)):
existing = await user_CRUD.get_user_by_email(db, user_in.email)
if existing:
raise HTTPException(status_code=409, detail="Email already registered")
user = await user_CRUD.create_user(db, user_in)
return {"success": True, "data": UserResponse.model_validate(user)}
@router.put("/{user_id}", response_model=dict)
async def update_user(
user_id: UUID,
user_in: UserUpdate,
db: AsyncSession = Depends(get_db)
):
user = await user_CRUD.update_user(db, user_id, user_in)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return {"success": True, "data": UserResponse.model_validate(user)}
@router.delete("/{user_id}", status_code=204)
async def delete_user(user_id: UUID, db: AsyncSession = Depends(get_db)):
deleted = await user_CRUD.delete_user(db, user_id)
if not deleted:
raise HTTPException(status_code=404, detail="User not found")
Step 7: Main Application
# app/main.py
from FastAPI import FastAPI
from FastAPI.middleware.CORS import CORSMiddleware
from app.API.v1 import users, auth
from app.config import settings
app = FastAPI(
title=settings.app_name,
description="DodaTech REST API built with FastAPI",
version="1.0.0",
docs_URL="/docs",
redoc_URL="/redoc"
)
# CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Include routers
app.include_router(users.router, prefix="/API/v1")
app.include_router(auth.router, prefix="/API/v1")
@app.get("/health")
async def health_check():
return {"status": "ok", "timestamp": "2026-06-23T10:00:00Z"}
Running the API
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
Expected output:
INFO: Uvicorn running on http://0.0.0.0:8000
INFO: (Press CTRL+C to quit)
Visit http://localhost:8000/docs to see the interactive OpenAPI documentation.
Test with curl:
curl -X POST "HTTP://localhost:8000/API/v1/users/" \
-H "Content-Type: application/JSON" \
-d '{"name":"Alice","email":"alice@example.com","password":"securepass123"}'
Expected response:
{
"success": true,
"data": {
"name": "Alice",
"email": "alice@example.com",
"role": "member",
"id": "550e8400-e29b-41d4-a716-446655440000",
"is_active": true,
"created_at": "2026-06-23T10:00:00Z",
"updated_at": null
}
}
Common Errors
Blocking the event loop — Using synchronous database drivers or libraries (psycopg2 instead of asyncpg) in async endpoints. Always use async database drivers with FastAPI.
Missing Pydantic validation — Accepting raw dictionaries without Pydantic model validation. Always define request and response schemas with Pydantic for automatic validation and documentation.
Circular imports — Importing modules that import each other. Use the
apppackage structure and import lazily within functions when needed.Not using Dependency Injection — Creating database sessions manually in each endpoint. Use Depends(get_db) for consistent session management and automatic cleanup.
Overlooking CORS configuration — Frontend applications cannot access the API without proper CORS headers. Configure CORS middleware with the correct allowed origins.
Exposing sensitive fields — Returning password hashes or internal IDs in API responses. Use Pydantic response models that exclude sensitive fields.
No database Migration strategy — Manually altering database tables. Use Alembic for schema migrations and version control your database changes.
Practice Questions
- How does FastAPI generate OpenAPI documentation automatically?
- What is the purpose of Pydantic models in FastAPI?
- How do you use Dependency Injection for database sessions?
- What is the difference between async and sync endpoints in FastAPI?
- How do you configure CORS in a FastAPI application?
Challenge
Build a complete REST API for a blog platform with FastAPI. Include: CRUD for posts with title, content, author, tags, and published status, user authentication with JWT tokens and refresh tokens, role-based access control (admin can edit any post, author can edit own posts), pagination, filtering by tags and author, and full-text search on post content. Write async endpoints with SQLAlchemy and PostgreSQL. Include Alembic migrations for the schema.
FAQ
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro