FastAPI SQLAlchemy CRUD Guide

Building APIs is a core skill for developers.

FastAPI makes it fast and simple. It is a modern Python web framework.

It offers high performance and easy coding. You can create powerful APIs quickly.

This guide teaches you to build a CRUD API. We will use SQLAlchemy for the database.

CRUD stands for Create, Read, Update, Delete. These are basic data operations.

We will build a simple item management API. Follow each step carefully.

Project Setup and Installation

First, ensure you have Python installed. Use Python 3.7 or higher.

Create a new directory for your project. Then, set up a virtual environment.

This keeps your project dependencies isolated. Activate the virtual environment.

Now, install the required packages. You need FastAPI, SQLAlchemy, and a database driver.

We will use SQLite for simplicity. Install Uvicorn as the ASGI server.

If you face issues, see our guide on How to Install FastAPI in Python.

Run this command in your terminal.

pip install fastapi sqlalchemy uvicorn

Creating the Database Model

Models define your data structure. They map Python classes to database tables.

We will use SQLAlchemy's ORM. ORM means Object-Relational Mapper.

Create a file named models.py. Import necessary modules first.

Define a base class using declarative_base. Then, create your Item model.

Each model attribute is a table column. Define id, name, description, and price.

# models.py
from sqlalchemy import Column, Integer, String, Float
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Item(Base):
    __tablename__ = "items"

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

Setting Up the Database Session

You need a session to talk to the database. Create a database.py file.

Define the database URL. For SQLite, it's a local file.

Create an engine instance. The engine manages database connections.

Then, create a SessionLocal class. Each request will get its own session.

Finally, create all tables with Base.metadata.create_all.

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

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

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Import Base from models and create tables
from models import Base
Base.metadata.create_all(bind=engine)

Defining Pydantic Schemas

Schemas validate and serialize data. FastAPI uses Pydantic for this.

Create a schemas.py file. Define schemas for your data.

We need an ItemCreate schema for input. Also, an Item schema for output.

Pydantic ensures data types are correct. It also handles conversion.

# schemas.py
from pydantic import BaseModel

class ItemBase(BaseModel):
    name: str
    description: str | None = None
    price: float

class ItemCreate(ItemBase):
    pass

class Item(ItemBase):
    id: int

    class Config:
        orm_mode = True

Building the CRUD Operations

Now, create the main application file. Name it main.py.

Import FastAPI and your modules. Initialize the FastAPI app.

Create a dependency to get the database session. Use Depends from FastAPI.

Then, define your API endpoints. Each corresponds to a CRUD operation.

We will create four routes. They are POST, GET, PUT, and DELETE.

If you are new, check our FastAPI Tutorial: Build Your First Python REST API.

1. Create an Item (POST)

The POST endpoint creates a new item. It accepts data in the request body.

Use the ItemCreate schema for validation. Add the item to the database.

Then commit the session. Return the created item.

# main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List

from models import Item
from schemas import ItemCreate, Item as ItemSchema
from database import SessionLocal, engine

app = FastAPI()

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

@app.post("/items/", response_model=ItemSchema)
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
    db_item = Item(**item.dict())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

2. Read Items (GET)

We need two GET endpoints. One gets all items. The other gets a single item by ID.

Use db.query to fetch data. Handle missing items with a 404 error.

@app.get("/items/", response_model=List[ItemSchema])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = db.query(Item).offset(skip).limit(limit).all()
    return items

@app.get("/items/{item_id}", response_model=ItemSchema)
def read_item(item_id: int, db: Session = Depends(get_db)):
    db_item = db.query(Item).filter(Item.id == item_id).first()
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item

3. Update an Item (PUT)

The PUT endpoint updates an existing item. It needs the item ID and new data.

First, fetch the item from the database. If not found, raise an error.

Then, update its attributes. Commit the changes to the database.

@app.put("/items/{item_id}", response_model=ItemSchema)
def update_item(item_id: int, item: ItemCreate, db: Session = Depends(get_db)):
    db_item = db.query(Item).filter(Item.id == item_id).first()
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    for key, value in item.dict().items():
        setattr(db_item, key, value)
    db.commit()
    db.refresh(db_item)
    return db_item

4. Delete an Item (DELETE)

The DELETE endpoint removes an item. It also uses the item ID.

Find the item in the database. If it exists, delete it.

Commit the session to save the change.

@app.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()
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    db.delete(db_item)
    db.commit()
    return {"message": "Item deleted successfully"}

Running and Testing the API

Your API is now ready. Start the server using Uvicorn.

Open your terminal in the project folder. Run the following command.

uvicorn main:app --reload

You should see output indicating the server is running.

Open your browser. Go to http://127.0.0.1:8000/docs.

This is the automatic interactive API documentation. It is provided by FastAPI.

You can test all endpoints here. Try creating a new item first.

Use the POST endpoint. Provide JSON data in the request body.

Then, use the GET endpoints to retrieve your data.

Test update and delete as well. The docs make testing very easy.

If you get an import error, see Fix Python ImportError: No Module Named FastAPI.

Conclusion

You have built a complete CRUD API. It uses FastAPI and SQLAlchemy.

This combination is powerful and efficient. FastAPI handles requests fast.

SQLAlchemy manages database interactions smoothly. You learned to create models.

You set up database sessions and Pydantic schemas. You implemented all CRUD operations.

The API is ready for use. You can extend it with more features.

Add user authentication or more complex queries. The foundation is solid.

Remember to manage your dependencies. If needed, learn How to Uninstall FastAPI in Python.

Keep practicing to master these tools. Happy coding!