How to Build a REST API with FastAPI and Async Operations

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]

FastAPI → API framework
Uvicorn → ASGI web server for running FastAPI
SQLModel/Async SQLAlchemy → Async database layer

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:

/docs (Swagger UI)
/redoc

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

Async engine allows non-blocking database operations
AsyncSession is required for async CRUD

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?

Combines Pydantic + SQLAlchemy
Easy model creation
Async support

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

Async CRUD operations
Dependency injection using Depends()
Efficient async queries

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?

Database tables created on startup
User routes registered
Fully functional REST API

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

Clear separation of concerns
Easy to add new modules
Professional API structure

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

Hide internal DB fields
Prevent invalid input
Clear API docs in Swagger

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

Separation of routes and business logic
Easier to test and extend

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?

Lightweight
Stateless authentication
Works perfectly with FastAPI async routes

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

Faster responses
Less load on the database

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.