FastAPI has quickly become one of the most popular Python frameworks for building APIs. It's fast, easy to use, and provides automatic interactive documentation. In this tutorial, we'll build a complete REST API from scratch.
Why FastAPI?
FastAPI offers several advantages:
- Performance: One of the fastest Python frameworks available
- Type Hints: Full support for Python type hints
- Auto Documentation: Swagger UI and ReDoc out of the box
- Validation: Automatic request/response validation with Pydantic
- Async Support: Native async/await support
Setting Up Your Project
Create a new project directory and virtual environment:
mkdir fastapi-tutorial
cd fastapi-tutorial
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
Install FastAPI and dependencies:
pip install fastapi uvicorn[standard] sqlalchemy pydantic
Your First API
Create main.py:
from fastapi import FastAPI
app = FastAPI(
title="My API",
description="A sample API built with FastAPI",
version="1.0.0"
)
@app.get("/")
async def root():
return {"message": "Hello, World!"}
@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
return {"item_id": item_id, "query": q}
Run the server:
uvicorn main:app --reload
Visit http://localhost:8000/docs to see the automatic Swagger documentation!
Request Validation with Pydantic
Pydantic models provide automatic validation:
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime
class ItemBase(BaseModel):
name: str = Field(..., min_length=1, max_length=100)
description: Optional[str] = None
price: float = Field(..., gt=0)
is_available: bool = True
class ItemCreate(ItemBase):
pass
class Item(ItemBase):
id: int
created_at: datetime
class Config:
from_attributes = True
Use models in your endpoints:
from fastapi import FastAPI, HTTPException
app = FastAPI()
# In-memory storage for demo
items_db = {}
item_counter = 0
@app.post("/items/", response_model=Item)
async def create_item(item: ItemCreate):
global item_counter
item_counter += 1
db_item = Item(
id=item_counter,
created_at=datetime.now(),
**item.model_dump()
)
items_db[item_counter] = db_item
return db_item
@app.get("/items/{item_id}", response_model=Item)
async def get_item(item_id: int):
if item_id not in items_db:
raise HTTPException(status_code=404, detail="Item not found")
return items_db[item_id]
Path and Query Parameters
FastAPI makes parameter handling intuitive:
from typing import List, Optional
from enum import Enum
class SortOrder(str, Enum):
asc = "asc"
desc = "desc"
@app.get("/items/")
async def list_items(
skip: int = 0,
limit: int = 10,
search: Optional[str] = None,
sort: SortOrder = SortOrder.desc
):
results = list(items_db.values())
if search:
results = [i for i in results if search.lower() in i.name.lower()]
if sort == SortOrder.asc:
results.sort(key=lambda x: x.created_at)
else:
results.sort(key=lambda x: x.created_at, reverse=True)
return results[skip:skip + limit]
Database Integration with SQLAlchemy
Create database.py:
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = "sqlite:///./app.db"
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
# Dependency
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
Create models.py:
from sqlalchemy import Column, Integer, String, Float, Boolean, DateTime
from sqlalchemy.sql import func
from database import Base
class ItemModel(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(100), nullable=False)
description = Column(String(500))
price = Column(Float, nullable=False)
is_available = Column(Boolean, default=True)
created_at = Column(DateTime, server_default=func.now())
Update your endpoints to use the database:
from fastapi import Depends
from sqlalchemy.orm import Session
from database import get_db
import models
@app.post("/items/", response_model=Item)
async def create_item(item: ItemCreate, db: Session = Depends(get_db)):
db_item = models.ItemModel(**item.model_dump())
db.add(db_item)
db.commit()
db.refresh(db_item)
return db_item
@app.get("/items/", response_model=List[Item])
async def list_items(
skip: int = 0,
limit: int = 10,
db: Session = Depends(get_db)
):
return db.query(models.ItemModel).offset(skip).limit(limit).all()
Authentication with JWT
Install additional dependencies:
pip install python-jose[cryptography] passlib[bcrypt]
Create auth.py:
from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
SECRET_KEY = "your-secret-key-here" # Use env variable in production!
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
return pwd_context.hash(password)
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
to_encode = data.copy()
expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
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
# Fetch user from database here
return {"username": username}
Error Handling
Create custom exception handlers:
from fastapi import Request
from fastapi.responses import JSONResponse
class ItemNotFoundException(Exception):
def __init__(self, item_id: int):
self.item_id = item_id
@app.exception_handler(ItemNotFoundException)
async def item_not_found_handler(request: Request, exc: ItemNotFoundException):
return JSONResponse(
status_code=404,
content={
"error": "Item not found",
"item_id": exc.item_id,
"message": f"Item with id {exc.item_id} does not exist"
}
)
Testing Your API
Create test_main.py:
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
def test_read_root():
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Hello, World!"}
def test_create_item():
response = client.post(
"/items/",
json={"name": "Test Item", "price": 9.99}
)
assert response.status_code == 200
assert response.json()["name"] == "Test Item"
def test_read_item_not_found():
response = client.get("/items/999")
assert response.status_code == 404
Run tests:
pip install pytest
pytest
Project Structure
For larger projects, organize your code:
fastapi-tutorial/
├── app/
│ ├── __init__.py
│ ├── main.py
│ ├── config.py
│ ├── database.py
│ ├── models/
│ │ ├── __init__.py
│ │ └── item.py
│ ├── schemas/
│ │ ├── __init__.py
│ │ └── item.py
│ ├── routers/
│ │ ├── __init__.py
│ │ └── items.py
│ └── services/
│ ├── __init__.py
│ └── item_service.py
├── tests/
├── requirements.txt
└── README.md
Conclusion
FastAPI makes building robust APIs straightforward with its automatic validation, documentation, and async support. The combination of Python type hints and Pydantic models ensures your API is both well-documented and type-safe.
Explore more tutorials or check out the official FastAPI documentation to continue learning.