FastAPI with SQLAlchemy: Building Scalable APIs with a Database Backend

APIs are the backbone of modern software applications. The faster they are, the more responsive your app becomes. With Python’s ecosystem growing rapidly, there are two standout libraries for building APIs and managing databases: FastAPI and SQLAlchemy.

In our previous article “Mastering CRUD Operations with FastAPI: A Step-by-Step Guide” we explored basic CRUD Operations without using the Database connection, In this article, we’ll dive into how to leverage the speed of FastAPI and the robustness of SQLAlchemy to build scalable APIs with a solid database backend.

FastAPI-SQLAlchemy

Why FastAPI?

FastAPI has gained a lot of traction recently for building APIs in Python because of its key features:

• Speed: It claims to be one of the fastest Python frameworks.
• Asynchronous Support: With async/await, FastAPI can handle large numbers of concurrent requests.
• Ease of Use: FastAPI is easy to set up, providing automatic documentation with OpenAPI.
• Type Checking: FastAPI uses Python-type hints to ensure type safety in your APIs.

This makes FastAPI an ideal choice for building both simple and complex APIs.

Why SQLAlchemy?

On the database side, SQLAlchemy is one of the most mature and flexible ORMs (Object Relational Mappers) for Python. It allows you to interact with databases using Python objects rather than writing raw SQL.

• ORM Abstraction: Maps database tables to Python classes and records to Python objects.
• Flexibility: Works with many database backends (PostgreSQL, MySQL, SQLite, etc.).
• Mature Ecosystem: Well-documented and battle-tested.

With SQLAlchemy’s declarative system, you can seamlessly define database models, perform queries, and handle transactions.

Setting up FastAPI and SQLAlchemy with MySQL

Let’s jump into how you can combine FastAPI with SQLAlchemy using MySQL. We’ll walk through creating a simple API that manages items in an inventory, using a MySQL database as the backend.

Step 1: Installing the Dependencies

First, Install FastAPI, SQLAlchemy, uvicorn (a lightning-fast ASGI server to run FastAPI), and the MySQL database connector aiomysql for asynchronous database interactions.

pip install fastapi uvicorn sqlalchemy databases pymysql aiomysql

Here, pymysql is the standard MySQL driver, while aiomysql allows for asynchronous MySQL connections.

Step 2: Project Structure

Here’s how you might structure your project:

.
├──fastapi-mysql
│   ├── main.py
│   ├── database.py
│   ├── models.py
│   └── crud.py
└── requirements.txt

This structure helps separate concerns, making your application scalable and easier to maintain.

Step 3: Database Configuration

First, we need to define the MySQL database connection. Create a database.py file:

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

# Replace with your actual MySQL credentials
DATABASE_URL = "mysql+pymysql://username:password@localhost:3306/dbname"

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

• DATABASE_URL: Here, we’re using the MySQL dialect (mysql+pymysql) with the pymysql driver.
• SessionLocal: This is the SQLAlchemy session that manages database transactions.
• Base: All our models will inherit from this.

Step 4: Define the Models

In models.py, define the database tables using SQLAlchemy’s ORM. For example, let’s create a simple Item model:

from sqlalchemy import Column, Integer, String
from database import Base

class Item(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(255), index=True)
    description = Column(String(255), index=True)
    price = Column(Integer)

This defines an Items table with columns id, name, description, and price. The String(255) specifies that the maximum length for name and description is 255 characters. Below is the table structure forItems

CREATE TABLE `items` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `name` varchar(255) NOT NULL,
  `description` varchar(255) NOT NULL,
  `price` int(11) NOT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci

Step 5: Creating CRUD Operations

We’ll define the Create, Read, Update, and Delete functions. This goes into the crud.py file:

from sqlalchemy.orm import Session
from models import Item

def get_item(db: Session, item_id: int):
    return db.query(Item).filter(Item.id == item_id).first()

def create_item(db: Session, name: str, description: str, price: int):
    db_item = Item(name=name, description=description, price=price)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

These are basic CRUD operations you can later expand upon.

Step 6: Creating API Endpoints

Now that our database models and CRUD operations are ready, we can define our FastAPI routes in main.py:

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from database import SessionLocal, engine
from models import Base, Item
from crud import get_item, create_item

app = FastAPI()

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

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

@app.post("/items/")
async def create_item_endpoint(name: str, description: str, price: int, db: Session = Depends(get_db)):
    return create_item(db, name, description, price)

@app.get("/items/{item_id}")
async def get_item_endpoint(item_id: int, db: Session = Depends(get_db)):
    item = get_item(db, item_id)
    if item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@app.put("/items/{item_id}")
async def update_item_endpoint(item_id: int, name: str, description: str, price: int, db: Session = Depends(get_db)):
    item = db.query(Item).filter(Item.id == item_id).first()
    if item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    
    item.name = name
    item.description = description
    item.price = price
    db.commit()
    db.refresh(item)
    
    return item

@app.delete("/items/{item_id}")
async def delete_item_endpoint(item_id: int, db: Session = Depends(get_db)):
    item = db.query(Item).filter(Item.id == item_id).first()
    if item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    
    db.delete(item)
    db.commit()
    
    return {"detail": "Item deleted"}

Here’s what’s happening:

• Dependency Injection: Depends(get_db) injects a database session into your route.
• CRUD Operations: The create_item_endpoint and get_item_endpoint functions map to API routes for creating and retrieving items.

Step 7: Running the API

Finally, you can run the FastAPI application with Uvicorn:

uvicorn main:app --reload

Your API will be live at http://127.0.0.1:8000. FastAPI will also automatically generate interactive docs /docs using Swagger UI. check below the step-by-step screenshot for basic CRUD operation using MySQL Database.

FastAPI — MYSQL Database Connect API EndPoint

FastAPI — MYSQL Database Table Insertion

Conclusion

Combining FastAPI and SQLAlchemy gives you a powerful and efficient toolset for building APIs. FastAPI’s speed and async capabilities pair well with SQLAlchemy’s flexibility for handling databases, enabling you to build scalable, maintainable, and high-performance applications.

Switching to MySQL is easy with SQLAlchemy’s multi-dialect support. By modifying a few settings, you can integrate your preferred database backend seamlessly.

This setup is just the beginning. You can easily extend this with features like user authentication, testing, deployment strategies, and more to create robust, production-ready APIs.

Happy coding!