Getting into building a CRUD (Create, Read, Update, Delete) application using FastAPI, SQLAlchemy, and Pydantic can feel like a daunting task at first, but it’s super empowering once you get the hang of it. This guide is designed to walk you through it casually and in easy-to-understand terms. By the end, you’ll have a nifty app to manage your data seamlessly.

Let’s kick things off by setting up your environment. Make sure you have Python installed on your machine. You’ll also need a few packages like FastAPI, SQLAlchemy, and Pydantic. Getting these is as simple as running a pip command.

pip install fastapi sqlalchemy pydantic uvicorn

With the necessary tools in hand, let’s get into creating your database models with SQLAlchemy. Think of these models as the blueprint of your database structure.

Here’s a basic setup:

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship

# Create a database engine
engine = create_engine('sqlite:///example.db')
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    items = relationship("Item", back_populates="owner")

class Item(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))
    owner = relationship("User", back_populates="items")

Base.metadata.create_all(bind=engine)

Essentially, this code sets up a simple SQLite database with two tables: users and items. The User class represents users in the app, and each user can own multiple items, represented by the Item class.

Next, we need to define Pydantic schemas. These schemas serve as the intermediary between your database models and the API, ensuring that the data conforms to certain rules.

Here’s what that looks like:

from pydantic import BaseModel

class UserBase(BaseModel):
    email: str

class UserCreate(UserBase):
    pass

class User(UserBase):
    id: int
    items: list = []

    class Config:
        orm_mode = True

class ItemBase(BaseModel):
    title: str
    description: str

class ItemCreate(ItemBase):
    pass

class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True

In this setup, UserCreate and ItemCreate schemas are used when creating new users or items. The User and Item schemas include additional fields needed when reading data from the database.

Setting up the FastAPI framework is the next step. It’s the backbone of your app, managing all the API requests.

Here’s the initial setup:

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session

app = FastAPI()

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

With this setup, we define the main app instance using FastAPI and a dependency to get a database session. This session is used to interact with the database in each request.

Now, let’s dive into the fun part: creating the CRUD operations via FastAPI path operations.

Create

Creating new users or items involves a POST request. Here’s the code for creating a user or an item:

@app.post("/users/", response_model=User)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = User(email=user.email)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

@app.post("/items/", response_model=Item)
def create_item_for_user(item: ItemCreate, db: Session = Depends(get_db)):
    db_item = Item(title=item.title, description=item.description, owner_id=1)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

Here, users and items are added to the database when we receive a POST request at /users/ or /items/. The data is committed to the database, and the newly created record is returned.

Read

Reading data uses GET requests to fetch user or item details:

@app.get("/users/", response_model=list[User])
def read_users(db: Session = Depends(get_db)):
    return db.query(User).all()

@app.get("/items/", response_model=list[Item])
def read_items(db: Session = Depends(get_db)):
    return db.query(Item).all()

@app.get("/users/{user_id}", response_model=User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

@app.get("/items/{item_id}", response_model=Item)
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

These endpoints allow you to fetch all users or items or a specific user or item by ID from the database.

Update

Updating existing records uses PATCH requests:

@app.patch("/users/{user_id}", response_model=User)
def update_user(user_id: int, user: UserCreate, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    update_data = user.dict(exclude_unset=True)
    for key, value in update_data.items():
        setattr(db_user, key, value)
    db.commit()
    db.refresh(db_user)
    return db_user

@app.patch("/items/{item_id}", response_model=Item)
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")
    update_data = item.dict(exclude_unset=True)
    for key, value in update_data.items():
        setattr(db_item, key, value)
    db.commit()
    db.refresh(db_item)
    return db_item

These endpoints allow updating user or item details. The provided data only updates the fields included in the request, leaving the rest unchanged.

Delete

Deleting records uses a DELETE request:

@app.delete("/users/{user_id}")
def delete_user(user_id: int, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    db.delete(db_user)
    db.commit()
    return {"detail": "User deleted"}

@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 {"detail": "Item deleted"}

These endpoints enable deleting users or items from the database. If the user or item does not exist, a 404 error is raised.

Handling relationships between models ensures that the related data is correctly loaded. For example, when fetching a user, you might also want to fetch their related items. Here’s how you can handle relationships:

from sqlalchemy.orm import selectinload

@app.get("/users/{user_id}", response_model=User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = db.query(User).options(selectinload(User.items)).filter(User.id == user_id).first()
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

Using selectinload, you can load related items in the same database query, avoiding multiple queries and improving performance.

If you’re keen on boosting performance with asynchronous operations, async SQLAlchemy is your friend. Here’s a brief peek into using it:

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

async_engine = create_async_engine('sqlite+aiosqlite:///example.db')

async def get_db():
    async with AsyncSession(async_engine) as session:
        yield session

@app.get("/users/")
async def read_users(db: AsyncSession = Depends(get_db)):
    return await db.execute(select(User))

This setup uses create_async_engine for async operations, making your app faster and more efficient.

In conclusion, building a CRUD application with FastAPI, SQLAlchemy, and Pydantic is quite straightforward. You’ll define your database models with SQLAlchemy, create Pydantic schemas for validation, and whip up FastAPI path operations to manage the data. Keep an eye out for relationships and opt for async operations when needed. This ensures your app runs smooth and efficient.

Now you’re all set to create your own CRUD application. Happy coding!