Complete CRUD Operations in FastAPI: Update & Delete

How to Complete CRUD Operations in FastAPI: Update and Delete

In this tutorial, you will learn how to implement the remaining CRUD (Create, Read, Update, Delete) operations for your FastAPI application. We will cover how to update existing resources using both PUT (full replacement) and PATCH (partial update) methods, and how to delete resources. Additionally, we will configure cascade delete for related data and explore how these operations interact with your interactive API documentation.

Prerequisites

• A basic understanding of Python and FastAPI.
• Previous experience with setting up a database using SQLAlchemy in FastAPI (as covered in previous tutorials).
• Familiarity with Pydantic models for data validation.

Understanding PUT vs. PATCH for Updates

When updating resources in a REST API, you typically encounter two methods:

• PUT: This method is used for a full replacement of a resource. You must send all fields for the resource, effectively replacing the existing record with a new version.
• PATCH: This method is used for partial updates. You only send the fields that have changed. Any fields not included in the request will remain unchanged. PATCH is generally more practical for APIs as it doesn’t force clients to resend unchanged data.

Step 1: Implement Schema for Partial Updates (PATCH)

To support PATCH requests, we need a Pydantic schema where all fields are optional. This allows clients to send only the fields they intend to update.

In your schemas.py file, create a new schema, for example, PostUpdate, inheriting from PostBase. Make all fields optional and set their default values to None.

Example (PostUpdate schema):

from pydantic import BaseModel

class PostBase(BaseModel):
    title: str
    content: str
    user_id: int

class PostCreate(PostBase):
    pass

class PostUpdate(PostBase):
    title: str | None = None
    content: str | None = None
    user_id: int | None = None # Typically, you might not allow changing user_id in a partial update

Note: It’s generally not recommended to allow changing the user_id in a partial update. Consider a dedicated endpoint or a PUT request for such operations.

Step 2: Implement PUT Endpoint for Full Post Updates

In your main.py file, define a PUT endpoint for updating posts. This endpoint will use the PostCreate schema because it requires all fields.

1. Import the necessary schemas and models.
2. Define a new route function (e.g., update_post_full) using the PUT method and specify the post_id as a path parameter.
3. Use the PostCreate schema for the request body.
4. Retrieve the post by its ID. If the post doesn’t exist, return a 404 Not Found error.
5. (Optional but recommended) If the user_id is being updated, verify that the new user exists before proceeding.
6. Update all fields of the post object with the data from the request body.
7. Commit the changes to the database using db.commit().
8. Refresh the post object using db.refresh().
9. Return the updated post object.

Example Snippet (main.py):

from fastapi import FastAPI, HTTPException, status
from sqlalchemy.orm import Session
from . import schemas, models, crud # Assuming crud handles database operations

# ... other imports and app setup ...

@app.put("/posts/{post_id}", response_model=schemas.PostResponse)
def update_post_full(post_id: int, post_data: schemas.PostCreate, db: Session = Depends(get_db)):
    db_post = crud.get_post(db, post_id=post_id)
    if db_post is None:
        raise HTTPException(status_code=404, detail="Post not found")

    # Optional: Check if the new user_id exists if it's different
    if post_data.user_id is not None and db_post.user_id != post_data.user_id:
        db_user = crud.get_user(db, user_id=post_data.user_id)
        if db_user is None:
            raise HTTPException(status_code=404, detail="User not found")

    # Update all fields
    db_post.title = post_data.title
    db_post.content = post_data.content
    db_post.user_id = post_data.user_id

    db.commit()
    db.refresh(db_post)
    return db_post

Step 3: Implement PATCH Endpoint for Partial Post Updates

For partial updates, use the PostUpdate schema (with optional fields) and the PATCH HTTP method.

1. Define a new route function (e.g., update_post_partial) using the PATCH method and the post_id.
2. Use the PostUpdate schema for the request body.
3. Retrieve the post by its ID. Return 404 Not Found if it doesn’t exist.
4. Convert the Pydantic model to a dictionary, excluding unset fields. This is crucial for PATCH. Use post_data.model_dump(exclude_unset=True).
5. Iterate through the dictionary of updated fields.
6. Dynamically update the post object’s attributes using setattr() for each field that was provided in the request.
7. Commit changes, refresh the object, and return the updated post.

Example Snippet (main.py):

@app.patch("/posts/{post_id}", response_model=schemas.PostResponse)
def update_post_partial(post_id: int, post_data: schemas.PostUpdate, db: Session = Depends(get_db)):
    db_post = crud.get_post(db, post_id=post_id)
    if db_post is None:
        raise HTTPException(status_code=404, detail="Post not found")

    update_data = post_data.model_dump(exclude_unset=True)

    for field, value in update_data.items():
        # Basic check to prevent updating user_id if not intended, or add more robust checks
        if field == "user_id" and value is not None:
            db_user = crud.get_user(db, user_id=value)
            if db_user is None:
                raise HTTPException(status_code=404, detail=f"User with id {value} not found")
        setattr(db_post, field, value)

    db.commit()
    db.refresh(db_post)
    return db_post

Step 4: Implement DELETE Endpoint for Posts

To delete a post, you’ll use the DELETE HTTP method. Typically, a successful deletion returns a 204 No Content status code.

1. Define a route function (e.g., delete_post) using the DELETE method and the post_id.
2. Retrieve the post by its ID. Return 404 Not Found if it doesn’t exist.
3. Use db.delete(db_post) to mark the post for deletion.
4. Commit the changes using db.commit().
5. Do not return a response body; instead, set the status code to 204 No Content.

Example Snippet (main.py):

from fastapi import FastAPI, HTTPException, status, Response
from sqlalchemy.orm import Session
from . import schemas, models, crud

# ... other imports and app setup ...

@app.delete("/posts/{post_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_post(post_id: int, db: Session = Depends(get_db)):
    db_post = crud.get_post(db, post_id=post_id)
    if db_post is None:
        raise HTTPException(status_code=404, detail="Post not found")
    db.delete(db_post)
    db.commit()
    return Response(status_code=status.HTTP_204_NO_CONTENT)

Step 5: Implement User Updates (PATCH)

Similar to posts, you can implement update functionality for users. For simplicity and common practice, we’ll focus on the PATCH method for partial updates.

1. In schemas.py, create a UserUpdate schema similar to PostUpdate, making fields like username, email, and image_file optional and defaulting to None.
2. In main.py, define a PATCH endpoint (e.g., update_user) for a specific user_id.
3. Use the UserUpdate schema for the request body.
4. Retrieve the user by ID; return 404 Not Found if absent.
5. Before updating, check if the provided username or email already exists for another user. If so, return a 400 Bad Request error.
6. Dynamically update the user object’s attributes based on the provided fields in the request body, similar to the partial post update.
7. Commit, refresh, and return the updated user.

Example Snippet (main.py):

@app.patch("/users/{user_id}", response_model=schemas.UserResponse)
def update_user(user_id: int, user_data: schemas.UserUpdate, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")

    update_data = user_data.model_dump(exclude_unset=True)

    # Check for username/email conflicts if they are being updated
    if "username" in update_data and update_data["username"] != db_user.username:
        existing_user = crud.get_user_by_username(db, username=update_data["username"])
        if existing_user:
            raise HTTPException(status_code=400, detail="Username already registered")

    if "email" in update_data and update_data["email"] != db_user.email:
        existing_user = crud.get_user_by_email(db, email=update_data["email"])
        if existing_user:
            raise HTTPException(status_code=400, detail="Email already registered")

    for field, value in update_data.items():
        setattr(db_user, field, value)

    db.commit()
    db.refresh(db_user)
    return db_user

Step 6: Configure Cascade Delete for Users and Posts

To ensure that deleting a user also deletes all their associated posts, you need to configure this behavior in your SQLAlchemy models.

1. Open your models.py file.
2. Locate the User model and its relationship to Post.
3. Modify the relationship definition to include cascade="all, delete-orphan".

Example Snippet (models.py):

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

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    email = Column(String, unique=True, index=True)
    image_file = Column(String, nullable=True)

    # Configure cascade delete for posts
    posts = relationship("Post", back_populates="author", cascade="all, delete-orphan")

class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    content = Column(String)
    user_id = Column(Integer, ForeignKey("users.id"))

    author = relationship("User", back_populates="posts")

Warning: Cascade deletion is a powerful feature. Ensure you have proper user confirmation and warnings in place before deleting resources that have associated data.

Step 7: Implement User Deletion (DELETE)

Implement the DELETE endpoint for users. Thanks to the cascade delete configuration, deleting a user will automatically remove their posts.

1. Define a DELETE route function (e.g., delete_user) for a specific user_id.
2. Retrieve the user by ID; return 404 Not Found if absent.
3. Use db.delete(db_user) to mark the user for deletion.
4. Commit the changes using db.commit().
5. Return a 204 No Content status code.

Example Snippet (main.py):

@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    db.delete(db_user)
    db.commit()
    return Response(status_code=status.HTTP_204_NO_CONTENT)

Step 8: Updating User Profile Pictures

While this tutorial doesn’t cover file uploads directly, we can demonstrate updating the image_file field for a user.

1. Ensure you have an image file (e.g., corey.png) in your media directory (e.g., media/profile_pics/). In a real application, this step would involve file upload handling.
2. Use the PATCH /users/{user_id} endpoint.
3. In the request body, provide the image_file field with the name of the image file you placed in the media directory.
4. The image_path property defined in your UserModel should automatically construct the correct URL to the image.

By following these steps, you have successfully implemented full CRUD operations for both your users and posts, including robust update and delete functionalities with cascade deletion for related data.

Source: Python FastAPI Tutorial (Part 6): Completing CRUD – Update and Delete (PUT, PATCH, DELETE) (YouTube)