python

What’s the Secret to Building a Slick CRUD App with FastAPI, SQLAlchemy, and Pydantic?

Mastering the Art of CRUD with FastAPI, SQLAlchemy, and Pydantic

What’s the Secret to Building a Slick CRUD App with FastAPI, SQLAlchemy, and Pydantic?

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!

Keywords: FastAPI, SQLAlchemy, Pydantic, CRUD application, Python, FastAPI tutorial, SQLAlchemy tutorial, Pydantic schemas, API development, Python web framework



Similar Posts
Blog Image
Is Your FastAPI Application Ready for a Global Makeover?

Deploying FastAPI Globally: Crafting A High-Performance, Resilient API Network

Blog Image
Harnessing Python's Metaprogramming to Write Self-Modifying Code

Python metaprogramming enables code modification at runtime. It treats code as manipulable data, allowing dynamic changes to classes, functions, and even code itself. Decorators, exec(), eval(), and metaclasses are key features for flexible and adaptive programming.

Blog Image
Supercharge Your Python: Mastering Structural Pattern Matching for Cleaner Code

Python's structural pattern matching, introduced in version 3.10, revolutionizes control flow. It allows for sophisticated analysis of complex data structures, surpassing simple switch statements. This feature shines when handling nested structures, sequences, mappings, and custom classes. It simplifies tasks that previously required convoluted if-else chains, making code cleaner and more readable. While powerful, it should be used judiciously to maintain clarity.

Blog Image
Unleashing Python’s Hidden Power: Advanced Generator Patterns You Never Knew About

Python generators offer lazy evaluation, memory efficiency, and versatility. They enable coroutines, infinite sequences, data pipelines, file processing, and asynchronous programming. Generators simplify complex tasks and improve code performance.

Blog Image
Can FastAPI Make Long-Running Tasks a Breeze?

Harnessing FastAPI’s Magical Background Tasks to Improve API Performance

Blog Image
How Fun and Easy Is It to Build a URL Shortener with Flask?

Turning Long URLs into Bite-Sized Links with Flask Magic