KAHIBARO
Discord Login Register ☰
☰

32.4. Building the REST API

Table of Contents

Overview

In the final project you already planned the application, the architecture, and the database. In this chapter you will turn that plan into a working REST API.

You will not learn REST or FastAPI from scratch here. Instead, you will focus on how to design, organize, and implement a real, production‑grade API using the concepts you already know.

This chapter uses Python and FastAPI in examples, but the structure and ideas apply to any backend framework.

Key idea: In a production project, your REST API should be:

  • predictable,
  • consistent,
  • well structured,
  • easy to extend,
  • well validated,
  • and well documented.

Translating Requirements into Endpoints

You already defined what your application must do in the planning chapter. Now you map those features to REST resources and endpoints.

From features to resources

Start by listing the core entities in your domain. For example, in a generic production app you might have:

Each entity usually becomes one or more REST resources. A resource is typically represented by a URL path prefix, such as:

Create a table for your project that maps business features to resources:

FeatureResource(s)Notes
User registration/auth/registerReturns created user or minimal profile
Login and token issuance/auth/loginReturns access and refresh tokens
List current user tasks/tasksFilter by assignee_id = current user
Create new project/projectsOnly for authenticated users
Add task to a project/projects/{id}/tasksOr just /tasks with project_id field
Comment on a task/tasks/{id}/commentsNested resource

Write this mapping explicitly before coding. It prevents you from creating random paths that are difficult to maintain later.

RESTful URL Design for the Project

In this step you refine URLs, HTTP methods, and path parameters using REST principles you already learned in the REST and FastAPI chapters.

Basic URL patterns

Use the standard resource patterns:

Examples:

Rule: Use plural resource names for collections (for example /users, /orders),
and avoid verbs in paths (for example prefer /tasks over /createTask).

Example URL design for a task management style app

ResourceHTTP methodURLPurpose
TasksGET/tasksList tasks
TasksPOST/tasksCreate a task
TasksGET/tasks/{task_id}Get task detail
TasksPATCH/tasks/{task_id}Update part of a task
TasksPUT/tasks/{task_id}Replace a task
TasksDELETE/tasks/{task_id}Delete a task
TasksGET/projects/{project_id}/tasksList tasks in a project
CommentsPOST/tasks/{task_id}/commentsAdd comment to a task
CommentsGET/tasks/{task_id}/commentsList task comments

Nested URLs such as /tasks/{task_id}/comments are useful when the relationship is very tight and you always access a child through its parent.

Query parameters vs path parameters

Use path parameters for resource identity, and query parameters for filtering and pagination.

Examples:

Rule:

  • Use path parameters for unique resource identity.
  • Use query parameters for search, filters, sorting, and pagination.

Designing Request and Response Models

In a production API you almost never work directly with raw dictionaries. You define schemas or models for input and output.

In FastAPI you usually use Pydantic models for this.

Separate models for create, update, and response

You often need:

Example for a Task resource:

python 
from datetime import datetime
from typing import Optional, Literal
from pydantic import BaseModel, Field
TaskStatus = Literal["todo", "in_progress", "done"]
class TaskCreate(BaseModel):
    title: str = Field(..., max_length=200)
    description: Optional[str] = None
    project_id: int
    assignee_id: Optional[int] = None
    status: TaskStatus = "todo"
class TaskUpdate(BaseModel):
    title: Optional[str] = Field(None, max_length=200)
    description: Optional[str] = None
    assignee_id: Optional[int] = None
    status: Optional[TaskStatus] = None
class TaskBase(BaseModel):
    id: int
    title: str
    description: Optional[str]
    status: TaskStatus
    project_id: int
    assignee_id: Optional[int]
    created_at: datetime
    updated_at: datetime
    class Config:
        orm_mode = True
class TaskResponse(TaskBase):
    # you can extend this later, for example with nested user or project data
    pass

Rule:
Never expose internal fields that should stay secret, such as password hashes, internal IDs, or debug fields, in your response models.

Example: User models

A very common pattern:

python 
class UserCreate(BaseModel):
    email: str
    password: str
    full_name: Optional[str] = None
class UserResponse(BaseModel):
    id: int
    email: str
    full_name: Optional[str] = None
    is_active: bool
    class Config:
        orm_mode = True

Notice that UserResponse does not include password. You only use password in input models.

Implementing CRUD Endpoints

Now that you have URL patterns and models, you can implement CRUD logic.

In this project you should not put all logic directly in the FastAPI route functions. Use your architecture: service layer, repositories, and models. The route should be thin.

Folder and module structure for the API

A practical structure for a production FastAPI project:

text 
app/
  main.py
  api/
    v1/
      __init__.py
      router.py
      endpoints/
        __init__.py
        auth.py
        users.py
        projects.py
        tasks.py
  core/
    config.py
    security.py
  db/
    base.py
    session.py
  models/
    user.py
    project.py
    task.py
  schemas/
    user.py
    project.py
    task.py
  services/
    user_service.py
    project_service.py
    task_service.py

Example: router registration

python 
# app/api/v1/router.py
from fastapi import APIRouter
from .endpoints import auth, users, projects, tasks
api_router = APIRouter()
api_router.include_router(auth.router, prefix="/auth", tags=["auth"])
api_router.include_router(users.router, prefix="/users", tags=["users"])
api_router.include_router(projects.router, prefix="/projects", tags=["projects"])
api_router.include_router(tasks.router, prefix="/tasks", tags=["tasks"])

Then in main.py:

python 
from fastapi import FastAPI
from app.api.v1.router import api_router
app = FastAPI(title="Production Backend API")
app.include_router(api_router, prefix="/api/v1")

Example: create and read endpoints

Tasks endpoint implementation

python 
# app/api/v1/endpoints/tasks.py
from typing import List
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.db.session import get_db
from app.schemas.task import TaskCreate, TaskResponse, TaskUpdate
from app.services import task_service
from app.core.security import get_current_user
from app.models.user import User
router = APIRouter()
@router.post(
    "",
    response_model=TaskResponse,
    status_code=status.HTTP_201_CREATED,
)
def create_task(
    task_in: TaskCreate,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    # You can enforce ownership rules here if needed
    task = task_service.create_task(db=db, task_in=task_in, creator_id=current_user.id)
    return task
@router.get("", response_model=List[TaskResponse])
def list_tasks(
    project_id: int | None = None,
    status: str | None = None,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    tasks = task_service.list_tasks(
        db=db,
        current_user=current_user,
        project_id=project_id,
        status=status,
    )
    return tasks
@router.get("/{task_id}", response_model=TaskResponse)
def get_task(
    task_id: int,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    task = task_service.get_task_by_id_for_user(db, task_id=task_id, user=current_user)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return task
@router.patch("/{task_id}", response_model=TaskResponse)
def update_task(
    task_id: int,
    task_in: TaskUpdate,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    task = task_service.update_task_for_user(
        db=db,
        task_id=task_id,
        task_in=task_in,
        user=current_user,
    )
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return task
@router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_task(
    task_id: int,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    deleted = task_service.delete_task_for_user(
        db=db,
        task_id=task_id,
        user=current_user,
    )
    if not deleted:
        raise HTTPException(status_code=404, detail="Task not found")

The endpoints are very slim. Most logic is kept in task_service.

Example: service layer for tasks

python 
# app/services/task_service.py
from typing import List, Optional
from sqlalchemy.orm import Session
from app import models
from app.schemas.task import TaskCreate, TaskUpdate
from app.models.user import User
def create_task(db: Session, task_in: TaskCreate, creator_id: int) -> models.Task:
    task = models.Task(
        title=task_in.title,
        description=task_in.description,
        status=task_in.status,
        project_id=task_in.project_id,
        assignee_id=task_in.assignee_id,
        creator_id=creator_id,
    )
    db.add(task)
    db.commit()
    db.refresh(task)
    return task
def list_tasks(
    db: Session,
    current_user: User,
    project_id: Optional[int] = None,
    status: Optional[str] = None,
) -> List[models.Task]:
    query = db.query(models.Task)
    # simple example, you can add more complex permissions later
    query = query.filter(
        models.Task.assignee_id == current_user.id
        | (models.Task.creator_id == current_user.id)
    )
    if project_id is not None:
        query = query.filter(models.Task.project_id == project_id)
    if status is not None:
        query = query.filter(models.Task.status == status)
    return query.all()
def get_task_by_id_for_user(
    db: Session,
    task_id: int,
    user: User,
) -> Optional[models.Task]:
    return (
        db.query(models.Task)
        .filter(
            models.Task.id == task_id,
            (
                (models.Task.assignee_id == user.id)
                | (models.Task.creator_id == user.id)
            ),
        )
        .first()
    )
def update_task_for_user(
    db: Session,
    task_id: int,
    task_in: TaskUpdate,
    user: User,
) -> Optional[models.Task]:
    task = get_task_by_id_for_user(db, task_id=task_id, user=user)
    if not task:
        return None
    data = task_in.dict(exclude_unset=True)
    for field, value in data.items():
        setattr(task, field, value)
    db.add(task)
    db.commit()
    db.refresh(task)
    return task
def delete_task_for_user(
    db: Session,
    task_id: int,
    user: User,
) -> bool:
    task = get_task_by_id_for_user(db, task_id=task_id, user=user)
    if not task:
        return False
    db.delete(task)
    db.commit()
    return True

Validation and Error Handling in the API

Your REST API is the external contract of your backend. It must validate input strictly and return clear error messages.

Input validation with Pydantic

Use field types, constraints, and custom validators in your schemas.

Example:

python 
from pydantic import BaseModel, Field, EmailStr, validator
class RegisterUser(BaseModel):
    email: EmailStr
    password: str = Field(..., min_length=8, max_length=128)
    full_name: str = Field(..., min_length=2, max_length=100)
    @validator("password")
    def password_must_have_digit(cls, v: str) -> str:
        if not any(ch.isdigit() for ch in v):
            raise ValueError("Password must contain at least one digit")
        return v

If a client sends invalid data, FastAPI will automatically return a 422 Unprocessable Entity response that includes validation errors.

Rule:
Validate at the boundary of your system. Do not trust client input. Use schemas to filter and validate everything that comes into your API.

Standard error response structure

Define a consistent error shape. For example:

json 
{
  "detail": "Task not found",
  "code": "TASK_NOT_FOUND"
}

or with additional metadata:

json 
{
  "detail": "Task not found",
  "code": "TASK_NOT_FOUND",
  "status": 404
}

In FastAPI, you can raise HTTPException:

python 
from fastapi import HTTPException, status
raise HTTPException(
    status_code=status.HTTP_404_NOT_FOUND,
    detail="Task not found",
)

For more advanced handling, you can create custom exception classes and global exception handlers.

Example of a domain exception:

python 
class NotFoundError(Exception):
    def __init__(self, code: str, message: str):
        self.code = code
        self.message = message

And a handler:

python 
from fastapi import Request
from fastapi.responses import JSONResponse
@app.exception_handler(NotFoundError)
async def not_found_exception_handler(request: Request, exc: NotFoundError):
    return JSONResponse(
        status_code=404,
        content={"detail": exc.message, "code": exc.code},
    )

Then in your services:

python 
from app.core.exceptions import NotFoundError
def get_project_or_raise(db: Session, project_id: int) -> models.Project:
    project = db.query(models.Project).get(project_id)
    if not project:
        raise NotFoundError(code="PROJECT_NOT_FOUND", message="Project not found")
    return project

Pagination, Filtering, and Sorting

In a production API you rarely return all records. You implement pagination and basic filters to keep responses fast and predictable.

Pagination patterns

There are different strategies. The simplest is offset pagination using limit and offset.

text 
GET /tasks?limit=20&offset=40

This means "skip 40 items, then return the next 20."

python 
from typing import List
from fastapi import Query
@router.get("", response_model=List[TaskResponse])
def list_tasks(
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
    limit: int = Query(20, ge=1, le=100),
    offset: int = Query(0, ge=0),
):
    tasks = task_service.list_tasks_paginated(
        db=db,
        current_user=current_user,
        limit=limit,
        offset=offset,
    )
    return tasks

Service implementation:

python 
def list_tasks_paginated(
    db: Session,
    current_user: User,
    limit: int,
    offset: int,
) -> List[models.Task]:
    query = (
        db.query(models.Task)
        .filter(
            (models.Task.assignee_id == current_user.id)
            | (models.Task.creator_id == current_user.id)
        )
        .order_by(models.Task.created_at.desc())
    )
    return query.offset(offset).limit(limit).all()

You can also return metadata:

python 
class PaginatedTasks(BaseModel):
    total: int
    items: List[TaskResponse]
    limit: int
    offset: int

Then your endpoint can respond with pagination info.

Rule:
Always limit the maximum number of items returned by list endpoints. This protects your database and avoids huge responses.

Filtering

Add query parameters for fields that are commonly filtered.

Examples:

In FastAPI:

python 
@router.get("", response_model=List[TaskResponse])
def list_tasks(
    status: TaskStatus | None = None,
    project_id: int | None = None,
    assignee_id: int | None = None,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    tasks = task_service.list_tasks_filtered(
        db=db,
        current_user=current_user,
        status=status,
        project_id=project_id,
        assignee_id=assignee_id,
    )
    return tasks

In the service:

python 
from sqlalchemy import and_
def list_tasks_filtered(
    db: Session,
    current_user: User,
    status: Optional[str],
    project_id: Optional[int],
    assignee_id: Optional[int],
) -> List[models.Task]:
    filters = [
        (models.Task.assignee_id == current_user.id)
        | (models.Task.creator_id == current_user.id)
    ]
    if status is not None:
        filters.append(models.Task.status == status)
    if project_id is not None:
        filters.append(models.Task.project_id == project_id)
    if assignee_id is not None:
        filters.append(models.Task.assignee_id == assignee_id)
    return db.query(models.Task).filter(and_(*filters)).all()

Sorting

Use a sort query parameter. Common pattern:

Example parsing:

python 
from fastapi import Query
@router.get("", response_model=List[TaskResponse])
def list_tasks(
    sort: str = Query("-created_at"),
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    tasks = task_service.list_tasks_sorted(
        db=db,
        current_user=current_user,
        sort=sort,
    )
    return tasks

Service:

python 
from sqlalchemy import asc, desc
def list_tasks_sorted(
    db: Session,
    current_user: User,
    sort: str,
) -> List[models.Task]:
    field_name = sort.lstrip("-")
    direction = desc if sort.startswith("-") else asc
    field_map = {
        "created_at": models.Task.created_at,
        "updated_at": models.Task.updated_at,
        "title": models.Task.title,
        "status": models.Task.status,
    }
    order_field = field_map.get(field_name, models.Task.created_at)
    query = (
        db.query(models.Task)
        .filter(
            (models.Task.assignee_id == current_user.id)
            | (models.Task.creator_id == current_user.id)
        )
        .order_by(direction(order_field))
    )
    return query.all()

Versioning and URL Structure

You already saw the "/api/v1" prefix in earlier examples. This is a simple and popular way to version your API.

Rule:
Always version your public API from the beginning. Use a prefix like /api/v1 to avoid breaking existing clients in the future.

Common approaches:

ApproachExampleNotes
URL version/api/v1/tasksSimple and visible, often recommended
HeaderAccept: application/vnd.myapp.v1+jsonMore complex, usually for advanced APIs
Query param/tasks?version=1Not recommended for long term main version

In a typical FastAPI project you group routes by version folder, as seen earlier:

python 
# app/api/v1/router.py
# app/api/v2/router.py

When you need a new major version, you can:

Documentation and OpenAPI in the Final Project

FastAPI automatically generates an OpenAPI schema and interactive docs, but you still must design your endpoints and descriptions so that the documentation is useful.

Tags, summaries, and descriptions

Use tags and docstrings so that /docs and /redoc are readable.

python 
router = APIRouter(prefix="/tasks", tags=["tasks"])
@router.post(
    "",
    response_model=TaskResponse,
    status_code=status.HTTP_201_CREATED,
    summary="Create a new task",
    description="Create a new task in a project. The current user becomes the creator.",
)
def create_task(...):
    """
    Detailed description for API users.
    - **title**: Task title, max 200 characters.
    - **description**: Optional task details.
    - **status**: One of `todo`, `in_progress`, `done`.
    """
    ...

This appears in Swagger UI and helps other developers understand how to use your API.

Configuring metadata in the FastAPI app

In main.py:

python 
app = FastAPI(
    title="Production Backend API",
    description="Backend for the Final Project, providing tasks, projects, and authentication.",
    version="1.0.0",
    contact={
        "name": "Backend Team",
        "url": "https://example.com",
        "email": "backend@example.com",
    },
    license_info={
        "name": "MIT",
        "url": "https://opensource.org/licenses/MIT",
    },
)

FastAPI will expose:

You can then use this OpenAPI spec for:

Putting It All Together

In this chapter you focused on how to build the REST API layer of your production project without re‑explaining basic REST or FastAPI concepts.

Your main tasks for your own project are:

  1. List all features and map them to resources and endpoints.
  2. Design URLs using RESTful patterns, correct HTTP methods, and clear use of path and query parameters.
  3. Create Pydantic schemas for requests and responses, with strict validation.
  4. Implement CRUD endpoints with a thin routing layer and a clean service or repository layer underneath.
  5. Add pagination, filtering, and sorting to your list endpoints.
  6. Define consistent error handling and response shapes.
  7. Version your API using a clear prefix such as /api/v1.
  8. Use OpenAPI documentation features to make the API easy to understand and consume.

With these steps, you will have a well structured REST API that fits into the production architecture of the final project and is ready for authentication, PostgreSQL, Redis, background workers, and deployment in the following chapters.

Views: 7

Comments

Please login to add a comment.

Don't have an account? Register now!

© 2025 Kahibaro
Visitors: 2999 / 24h
Impressum / Imprint
Datenschutz / Data Policy