Changeset Management API in FastAPI

Welcome back! In the previous lesson, you learned how to set up a FastAPI backend for a Code Review Assistant. You created your first endpoint and learned how to store code changes (called changesets) in a database. Now, we will build on that foundation by creating an API to manage these changesets.

A changeset is a group of code changes, usually submitted together for review. In a code review assistant, managing changesets means being able to list them, view their details, and perform actions like scanning or reviewing them. The Changeset Management API is the part of your backend that lets users and other systems interact with these changesets.

By the end of this lesson, you will know how to:

• List changesets with optional filters.
• Retrieve detailed information about a specific changeset.
• Add helpful endpoints for health checks and triggering scans.

Let’s get started!

Before we dive in, let’s quickly remind ourselves how FastAPI endpoints and models work. In the last lesson, you learned how to:

• Define a FastAPI app using FastAPI().
• Create models (like Changeset) to represent your data.
• Set up endpoints using decorators like @app.post() or @app.get().

For example, here’s a simple FastAPI endpoint that returns a list of items:

from fastapi import FastAPI

app = FastAPI()

@app.get("/items")
def list_items():
    return [{"id": 1, "name": "Item 1"}]

This is the basic pattern we will use for our changeset management endpoints. Now, let’s see how to list changesets.

The first step in managing changesets is being able to list them. Often, you want to filter the list (for example, by status) or limit how many results you get.

Let’s build this step by step.

We start by creating a GET endpoint at /api/changesets. This endpoint will return a list of changesets.

from fastapi import FastAPI, Depends
from typing import List, Optional

app = FastAPI()

@app.get("/api/changesets")
def list_changesets():
    # We'll fill this in next
    pass

We want to allow users to filter changesets by status (like "pending" or "reviewed") and limit the number of results. We do this by adding query parameters to our endpoint.

@app.get("/api/changesets")
def list_changesets(status: Optional[str] = None, limit: int = 10):
    # We'll filter and limit the results here
    pass

• status: Optional[str] = None means the user can provide a status to filter by, but it’s not required.
• limit: int = 10 sets a default limit of 10 results.

Let’s assume we have a database session and a Changeset model. We use these to get the data.

from sqlalchemy.orm import Session

def get_session():
    # Returns a database session (dummy for this example)
    pass

@app.get("/api/changesets")
def list_changesets(
    status: Optional[str] = None,
    limit: int = 10,
    db: Session = Depends(get_session)
):
    query = db.query(Changeset)
    if status:
        query = query.filter(Changeset.status == status)
    changesets = query.order_by(Changeset.created_at.desc()).limit(limit).all()
    return [
        {
            "id": cs.id,
            "title": cs.title,
            "description": cs.description,
            "author": cs.author,
            "status": cs.status,
            "created_at": cs.created_at.isoformat()
        }
        for cs in changesets
    ]

Explanation:

• We use Depends(get_session) to get a database session.
• We build a query for Changeset objects.
• If a status is provided, we filter by it.
• We order the results by creation date (newest first) and limit the number of results.
• We return a list of dictionaries with the changeset details.

Example Output:

[
  {
    "id": 2,
    "title": "Fix bug Y",
    "description": "Fixes bug Y in module Z",
    "author": "bob",
    "status": "pending",
    "created_at": "2024-06-02T15:30:00"
  },
  {
    "id": 1,
    "title": "Add feature X",
    "description": "Implements feature X",
    "author": "alice",
    "status": "reviewed",
    "created_at": "2024-06-01T12:00:00"
  }
]

Now you know how to list changesets with optional filtering and limiting.

Sometimes, you need more information about a specific changeset, such as the files it changed and any review comments. Let’s build an endpoint for that.

We want to get details for a specific changeset by its ID. We use a path parameter for this.

from fastapi import HTTPException

@app.get("/api/changesets/{changeset_id}")
def get_changeset(changeset_id: int, db: Session = Depends(get_session)):
    # We'll fill this in next
    pass

We query the database for the changeset with the given ID.

changeset = db.query(Changeset).filter_by(id=changeset_id).first()
if not changeset:
    raise HTTPException(status_code=404, detail="Changeset not found")

• If the changeset is not found, we return a 404 error.

Each changeset may have several files associated with it. We fetch these as well.

files = db.query(ChangesetFile).filter_by(changeset_id=changeset_id).all()

If the changeset has been reviewed, we can include review comments for each file. In this example, we generate dummy reviews.

reviews = {}
if changeset.status == 'reviewed':
    for file in files:
        review = review_engine.review_changeset_file(db, file)
        reviews[file.file_path] = review

Finally, we return all the information in a single response.

return {
    "id": changeset.id,
    "title": changeset.title,
    "description": changeset.description,
    "author": changeset.author,
    "status": changeset.status,
    "created_at": changeset.created_at.isoformat(),
    "files": [
        {"file_path": f.file_path, "diff_content": f.diff_content}
        for f in files
    ],
    "reviews": reviews
}

Example Output:

{
  "id": 1,
  "title": "Add feature X",
  "description": "Implements feature X",
  "author": "alice",
  "status": "reviewed",
  "created_at": "2024-06-01T12:00:00",
  "files": [
    {"file_path": "src/feature_x.py", "diff_content": "diff --git ..."},
    {"file_path": "src/utils.py", "diff_content": "diff --git ..."}
  ],
  "reviews": {
    "src/feature_x.py": "Review for src/feature_x.py: Looks good!",
    "src/utils.py": "Review for src/utils.py: Looks good!"
  }
}

This endpoint gives you all the details you need about a changeset, including its files and reviews.

Besides managing changesets, it’s useful to have endpoints for checking if your service is running and for triggering scans of your code repository.

A health check endpoint lets you confirm that your API is up and running.

@app.get("/api/health")
def health_check():
    return {"status": "healthy", "service": "code-review-assistant"}

Example Output:

{
  "status": "healthy",
  "service": "code-review-assistant"
}

You might want to trigger a scan of your repository to find new changesets.

from fastapi import HTTPException

@app.post("/api/scan")
def trigger_scan(repo_path: str, db: Session = Depends(get_session)):
    try:
        # Dummy scan_repository function
        def scan_repository(repo_path, db):
            # Pretend we scanned 3 files
            return 3
        count = scan_repository(repo_path, db)
        return {"message": f"Scanned {count} files"}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

Example Output:

{
  "message": "Scanned 3 files"
}

These endpoints help you monitor and interact with your code review assistant beyond just managing changesets.

In this lesson, you learned how to build a Changeset Management API using FastAPI. You saw how to:

• List changesets with optional filters and limits.
• Retrieve detailed information about a specific changeset, including its files and reviews.
• Add helpful endpoints for health checks and triggering repository scans.

These features are essential for any code review assistant, making it easy to manage and review code changes. Take a moment to review the code examples and explanations above. In the next section, you’ll get hands-on practice implementing and using these endpoints yourself. Good luck, and keep up the great work!