Database Integration in FastAPI (SQLAlchemy CRUD)

 

Introduction

So far, we’ve built APIs without storing data.But real-world applications need a database to:

• Store user data.
• Manage products.
• Persist application state.

In this blog, we’ll connect FastAPI with a database using SQLAlchemy and perform CRUD operations.

What is SQLAlchemy?

SQLAlchemy is a powerful Python ORM (Object Relational Mapper) that allows you to:

• Interact with databases using Python code.
• Avoid writing raw SQL queries.
• Work with models instead of tables.

Step 1: Install Dependencies

pip install sqlalchemy

(For SQLite, no extra driver needed)

Project Structure Update

app/
├── main.py
├── database/
│   ├── connection.py
├── models/
│   ├── item.py
├── schemas/
│   ├── item.py
├── routes/
│   ├── item.py

Step 2: Database Connection

app/database/connection.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})

SessionLocal = sessionmaker(bind=engine)

Base = declarative_base()

Step 3: Create Model

app/models/item.py

from sqlalchemy import Column, Integer, String
from app.database.connection import Base

class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String)

Step 4: Create Schema (Pydantic)

app/schemas/item.py

from pydantic import BaseModel

class ItemCreate(BaseModel):
    name: str
    description: str

class ItemResponse(BaseModel):
    id: int
    name: str
    description: str

    class Config:
        orm_mode = True

Step 5: Create Database Tables

Add this in main.py:

from app.database.connection import engine, Base

Base.metadata.create_all(bind=engine)

Step 6: Dependency for DB Session

from app.database.connection import SessionLocal
from fastapi import Depends

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Step 7: CRUD Operations
app/routes/item.py
    1.Create Item

from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from app.models.item import Item
from app.schemas.item import ItemCreate
from app.database.connection import SessionLocal
router = APIRouter()
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
@router.post("/items/")
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
    db_item = Item(name=item.name, descript
Read Items
@router.get("/items/")
def get_items(db: Session = Depends(get_db)):
    return db.query(Item).all()
ion=item.description)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item
Update Item
@router.put("/items/{item_id}")
def update_item(item_id: int, item: ItemCreate, db: Session = Depends(get_db)):
    db_item = db.query(Item).filter(Item.id == item_id).first()
    db_item.name = item.name
    db_item.description = item.description
    db.commit()
    return db_item

Delete Item
@router.delete("/items/{item_id}")
def delete_item(item_id: int, db: Session = Depends(get_db)):
    db_item = db.query(Item).filter(Item.id == item_id).first()
    db.delete(db_item)
    db.commit()
    return {"message": "Item deleted"}
Step 8: Include Router
main.py
from app.routes import item

app.include_router(item.router)
Common Mistakes
1. Forgetting commit()
2. Not using Depends(get_db)
3. Not enabling orm_mode
4. Not closing DB session
Best Practices
1. Use separate files for models, schemas, routes
2. Use dependency injection for DB
3. Always validate input with schemas
4. Handle errors (we’ll improve later)
Key Takeaways
SQLAlchemy connects FastAPI to databases
Models = DB tables
Schemas = validation layer
CRUD operations are easy with ORM
Dependency injection manages DB sessions