Introduction
FastAPI is one of the fastest and most modern Python frameworks for building APIs. It is designed to be simple, fast, and developer-friendly. One of its biggest advantages is native support for async and await, which helps build highly scalable APIs that run efficiently under heavy load.
When combined with an async database library such as SQLModel, Tortoise ORM, or Async SQLAlchemy, you can build a truly high-performance backend.
This article will guide you step-by-step through building a REST API with FastAPI using async database operations, explained in simple and natural language.
1. Installing FastAPI and Uvicorn
Before building the API, install the required libraries.
pip install fastapi uvicorn
pip install sqlmodel[asyncio]
2. Creating the FastAPI Application
Create a file main.py.
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def home():
return {"message": "FastAPI is running!"}
Run the server
uvicorn main:app --reload
FastAPI automatically generates documentation at:
3. Setting Up the Async Database
We will use SQLModel for async database operations.
Create a database connection file: database.py
from sqlmodel import SQLModel
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import sessionmaker
DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
async def init_db():
async with engine.begin() as conn:
await conn.run_sync(SQLModel.metadata.create_all)
Explanation
4. Creating Database Models with SQLModel
Create models.py.
from sqlmodel import SQLModel, Field
from typing import Optional
class User(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str
email: str
Why SQLModel?
5. Creating CRUD Routes (Async Operations)
Create routes_user.py.
from fastapi import APIRouter, Depends
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.future import select
from models import User
from database import async_session
router = APIRouter()
async def get_session():
async with async_session() as session:
yield session
@router.post("/users")
async def create_user(user: User, session: AsyncSession = Depends(get_session)):
session.add(user)
await session.commit()
await session.refresh(user)
return user
@router.get("/users")
async def list_users(session: AsyncSession = Depends(get_session)):
result = await session.execute(select(User))
return result.all()
@router.get("/users/{user_id}")
async def get_user(user_id: int, session: AsyncSession = Depends(get_session)):
result = await session.execute(select(User).where(User.id == user_id))
return result.first()
@router.delete("/users/{user_id}")
async def delete_user(user_id: int, session: AsyncSession = Depends(get_session)):
result = await session.execute(select(User).where(User.id == user_id))
user = result.first()[0]
await session.delete(user)
await session.commit()
return {"message": "User deleted"}
Features
6. Include Routes in Main Application
Update main.py.
from fastapi import FastAPI
from routes_user import router as user_router
from database import init_db
app = FastAPI()
app.include_router(user_router)
@app.on_event("startup")
async def on_startup():
await init_db()
What Happens Now?
7. Testing Your API
Go to:
http://localhost:8000/docs
FastAPI automatically generates:
Try-out buttons
Input validation
JSON schema
8. Best Practices for FastAPI + Async DB
1. Use async everywhere
Avoid mixing sync and async operations.
2. Use connection pooling
Async engines reuse connections automatically.
3. Avoid blocking operations
Never use:
import time
time.sleep(5) # ❌ blocks async loop
Use:
await asyncio.sleep(5)
4. Use Pydantic models for request/response
Separate DB models and API schema.
5. Enable indexing in DB models
Faster queries.
6. Use background tasks for long-running work
FastAPI supports async background tasks.
9. Deployment Considerations
Use Production Server
gunicorn -k uvicorn.workers.UvicornWorker main:app
Use environment variables for database credentials
Never hardcode secrets.
Use connection pooling
For PostgreSQL or MySQL, set max connections.
10. Recommended Project Folder Structure
A clean folder structure makes your FastAPI project scalable and easier to maintain.
project/
│── main.py
│── database.py
│── models.py
│── schemas.py
│── routes/
│ └── users.py
│── services/
│ └── user_service.py
│── core/
│ ├── config.py
│ └── security.py
│── utils/
│ └── pagination.py
│── tests/
│ └── test_users.py
Why This Helps
11. Adding Pydantic Schemas (Best Practice)
Using separate request/response models improves API security and validation.
schemas.py
from pydantic import BaseModel
class UserCreate(BaseModel):
name: str
email: str
class UserRead(BaseModel):
id: int
name: str
email: str
class Config:
orm_mode = True
Why Schemas Matter
12. Async CRUD Using Service Layer (Cleaner Architecture)
Move CRUD logic to a service file.
services/user_service.py
from models import User
from sqlalchemy.future import select
from sqlmodel.ext.asyncio.session import AsyncSession
class UserService:
@staticmethod
async def create_user(data, session: AsyncSession):
user = User(**data.dict())
session.add(user)
await session.commit()
await session.refresh(user)
return user
@staticmethod
async def get_users(session: AsyncSession):
return (await session.execute(select(User))).all()
Benefit
13. Adding JWT Authentication (Optional Security Layer)
Secure APIs should include authentication.
Example Token Generation
from jose import jwt
from datetime import datetime, timedelta
SECRET = "MY_SECRET_KEY"
ALGO = "HS256"
def create_token(data: dict):
payload = data.copy()
payload["exp"] = datetime.utcnow() + timedelta(hours=1)
return jwt.encode(payload, SECRET, algorithm=ALGO)
Why JWT?
14. Adding Pagination for Large Data Responses
Avoid returning thousands of items at once.
utils/pagination.py
def paginate(query, page: int = 1, size: int = 10):
start = (page - 1) * size
end = start + size
return query[start:end]
Benefit
15. Async Optimizations for High Performance
Use connection pooling
Use prepared statements where possible
Avoid blocking code
Use background tasks for long jobs
Use caching (Redis) for repeated queries
Conclusion
Building a REST API with FastAPI and async database operations is straightforward yet incredibly powerful. With async/await, database operations become non-blocking, enabling your API to handle thousands of requests efficiently. By combining FastAPI, SQLModel (or async SQLAlchemy), and proper async patterns, you can build a clean, high-performance backend suitable for modern applications. This approach ensures scalability, speed, and excellent developer experience — perfect for both beginners and professional developers.