Operacje CRUD z SQLAlchemy

Welcome to Lesson 3! Today, we'll master CRUD operations using FastAPI and SQLAlchemy. CRUD stands for:

• Create: Adding new data.
• Read: Retrieving existing data.
• Update: Modifying existing data.
• Delete: Removing data.

These four operations are the foundation of almost any application that interacts with a database.

Before diving into CRUD, let's set up our SQLAlchemy model and Pydantic schemas. We'll use an in-memory SQLite database for our runnable examples.

First, our SQLAlchemy Todo model to represent a task:

from sqlalchemy import Column, Integer, String, Boolean
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Todo(Base):
    __tablename__ = "todos"
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    description = Column(String, default="")
    completed = Column(Boolean, default=False)

And Pydantic schemas for request/response:

from pydantic import BaseModel

class TodoCreate(BaseModel):
    title: str
    description: str = ""
    completed: bool = False

class TodoResponse(TodoCreate):
    id: int

    class Config:
        orm_mode = True

In FastAPI, we manage database sessions using dependencies. This ensures each request gets a fresh session and it's properly closed.

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)

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

This get_db function will be injected into our FastAPI endpoints.

To Create a new item, we'll use a POST request. The endpoint will receive data via a Pydantic model and use SQLAlchemy to add it to the database.

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

# ... (imports for Base, Todo, TodoCreate, TodoResponse, get_db, engine)

app = FastAPI()
Base.metadata.create_all(bind=engine) # Create tables

@app.post("/todos/", response_model=TodoResponse)
def create_todo(todo: TodoCreate, db: Session = Depends(get_db)):
    db_todo = Todo(title=todo.title, description=todo.description, completed=todo.completed)
    db.add(db_todo)
    db.commit()
    db.refresh(db_todo) # Refresh to get ID and updated fields
    return db_todo

The db.refresh() call updates our db_todo object with any database-generated values, like the id.

Let's see the SQLAlchemy 'Create' steps in action. This runnable script will add a new todo to our in-memory database.

To Read data, we use GET requests. We'll have two endpoints: one to get all todos, and another to get a single todo by its ID.

# ... (FastAPI app, imports, get_db, models)

@app.get("/todos/", response_model=list[TodoResponse])
def read_todos(db: Session = Depends(get_db)):
    todos = db.query(Todo).all()
    return todos

@app.get("/todos/{todo_id}", response_model=TodoResponse)
def read_todo(todo_id: int, db: Session = Depends(get_db)):
    todo = db.query(Todo).filter(Todo.id == todo_id).first()
    if todo is None:
        raise HTTPException(status_code=404, detail="Todo not found")
    return todo

We use .all() to get a list and .first() to get a single item. Remember to handle cases where an item isn't found!

Let's run a demo for the 'Read' operation. We'll first create a few todos, then fetch them all, and finally fetch a specific one by ID.

The Update operation (PUT or PATCH) allows us to modify an existing item. We'll typically find the item by ID, update its attributes, and commit the changes.

# ... (FastAPI app, imports, get_db, models)

class TodoUpdate(BaseModel):
    title: str | None = None
    description: str | None = None
    completed: bool | None = None

@app.put("/todos/{todo_id}", response_model=TodoResponse)
def update_todo(todo_id: int, todo_update: TodoUpdate, db: Session = Depends(get_db)):
    db_todo = db.query(Todo).filter(Todo.id == todo_id).first()
    if db_todo is None:
        raise HTTPException(status_code=404, detail="Todo not found")
    
    # Update fields only if provided
    for key, value in todo_update.dict(exclude_unset=True).items():
        setattr(db_todo, key, value)
    
    db.add(db_todo) # Re-add to session for update tracking
    db.commit()
    db.refresh(db_todo)
    return db_todo

Using exclude_unset=True in Pydantic ensures only provided fields are updated.

Here's a runnable example showing how to update a todo item's title and completion status using SQLAlchemy.

The Delete operation removes an item from the database. This is usually done with a DELETE request, targeting an item by its ID.

# ... (FastAPI app, imports, get_db, models)

@app.delete("/todos/{todo_id}", status_code=204) # 204 No Content for successful deletion
def delete_todo(todo_id: int, db: Session = Depends(get_db)):
    db_todo = db.query(Todo).filter(Todo.id == todo_id).first()
    if db_todo is None:
        raise HTTPException(status_code=404, detail="Todo not found")
    
    db.delete(db_todo)
    db.commit()
    return {"message": "Todo deleted successfully"} # FastAPI automatically handles 204

A successful deletion often returns a 204 No Content status code, meaning the request was fulfilled but there's no content to send back.

Let's run a script to demonstrate deleting a todo item. We'll create one, then delete it, and try to read it again to confirm its removal.

Great job! In this lesson, you've learned to implement the fundamental CRUD operations in FastAPI using SQLAlchemy:

• Create (POST): Using db.add(), db.commit(), and db.refresh().
• Read (GET): Using db.query().all() for lists and db.query().filter().first() for single items.
• Update (PUT): Fetching an item, modifying its attributes, then db.commit() and db.refresh().
• Delete (DELETE): Fetching an item, then db.delete() and db.commit().

These skills are crucial for building any data-driven API. In the next course, we'll dive into advanced topics like user authentication and authorization to secure your API endpoints!