KAHIBARO
Discord Login Register ☰
☰

8.12 Building a Complete REST API

Table of Contents

Planning the Complete REST API

Before writing any code, always start by deciding what your API will do. For this chapter, we will build a complete REST API for a simple “Tasks” application, similar to a mini version of a task management tool.

We will:

Think of this chapter as a complete walkthrough from “empty folder” to “working REST API” using FastAPI.

Our example domain will have:

We will focus on API design and FastAPI usage, not database specifics or authentication details, since those belong to other chapters.

Goal: By the end of this chapter you should be able to create a non‑trivial FastAPI application with:

  • Clear project structure
  • CRUD endpoints for a resource
  • Pydantic models for requests and responses
  • Validation and error handling
  • Pagination and simple filtering

Project Structure

You can start simple, but even for a small app, a bit of structure helps a lot.

A reasonable layout for this example:

text 
task_api/
    app/
        __init__.py
        main.py
        api/
            __init__.py
            v1/
                __init__.py
                tasks.py
        models/
            __init__.py
            task.py
        schemas/
            __init__.py
            task.py
        core/
            __init__.py
            config.py
        services/
            __init__.py
            tasks.py
    requirements.txt

Explanation:

Starting with structure that separates “API”, “schemas”, “models”, and “services” makes it easier to grow the project without turning main.py into a giant file.

Setting Up the FastAPI Application

Create app/main.py:

python 
from fastapi import FastAPI
from app.api.v1.tasks import router as tasks_router
app = FastAPI(
    title="Task Management API",
    version="1.0.0",
)
app.include_router(tasks_router, prefix="/api/v1", tags=["tasks"])

Explanation:

To run the app:

bash 
uvicorn app.main:app --reload

Visit:

Defining Task Schemas with Pydantic

The schemas define how data is sent to and from the API. They are not database models, but shapes of data you accept or return.

Create app/schemas/task.py:

python 
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, Field
class TaskBase(BaseModel):
    title: str = Field(..., min_length=1, max_length=100)
    description: Optional[str] = Field(None, max_length=1000)
    completed: bool = False
class TaskCreate(TaskBase):
    # Additional fields for creation can be added here
    pass
class TaskUpdate(BaseModel):
    title: Optional[str] = Field(None, min_length=1, max_length=100)
    description: Optional[str] = Field(None, max_length=1000)
    completed: Optional[bool] = None
class TaskInDBBase(TaskBase):
    id: int
    owner_id: int
    created_at: datetime
    updated_at: datetime
    class Config:
        from_attributes = True  # for ORM objects later
class Task(TaskInDBBase):
    """Public response model for a single task."""
    pass
class TaskList(BaseModel):
    """Response model for a list of tasks with pagination."""
    total: int
    page: int
    size: int
    items: list[Task]

Common patterns:

Simple In‑Memory Data Model

In a real application you would use an ORM and a real database. For this chapter we will simulate data storage with an in‑memory store but still keep a “model” layer.

Create app/models/task.py:

python 
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime
@dataclass
class TaskModel:
    id: int
    title: str
    description: str | None
    completed: bool
    owner_id: int
    created_at: datetime = field(default_factory=datetime.utcnow)
    updated_at: datetime = field(default_factory=datetime.utcnow)

This TaskModel is a simple dataclass to store attribute values. We will keep data in a Python dictionary.

We also need some storage and ID generation. For a small example, we can keep this in the service layer.

Business Logic in a Service Layer

We want to keep HTTP details (FastAPI, Request, Response) separate from business logic. That way we can change the API without breaking the core rules, or test logic without HTTP.

Create app/services/tasks.py:

python 
from collections.abc import Sequence
from datetime import datetime
from typing import Optional
from app.models.task import TaskModel
from app.schemas.task import TaskCreate, TaskUpdate
# In-memory "database"
_TASKS: dict[int, TaskModel] = {}
_NEXT_ID: int = 1

We will add functions to perform CRUD:

python 
def _get_next_id() -> int:
    global _NEXT_ID
    next_id = _NEXT_ID
    _NEXT_ID += 1
    return next_id

Create a task:

python 
def create_task(data: TaskCreate, owner_id: int) -> TaskModel:
    task_id = _get_next_id()
    now = datetime.utcnow()
    task = TaskModel(
        id=task_id,
        title=data.title,
        description=data.description,
        completed=data.completed,
        owner_id=owner_id,
        created_at=now,
        updated_at=now,
    )
    _TASKS[task_id] = task
    return task

Get a single task:

python 
def get_task(task_id: int) -> Optional[TaskModel]:
    return _TASKS.get(task_id)

List tasks with simple pagination and filtering:

python 
def list_tasks(
    owner_id: Optional[int] = None,
    completed: Optional[bool] = None,
    page: int = 1,
    size: int = 10,
) -> tuple[int, list[TaskModel]]:
    """
    Returns (total_count, items_for_this_page).
    """
    tasks: Sequence[TaskModel] = list(_TASKS.values())
    if owner_id is not None:
        tasks = [t for t in tasks if t.owner_id == owner_id]
    if completed is not None:
        tasks = [t for t in tasks if t.completed == completed]
    # sort by created_at for consistency
    tasks = sorted(tasks, key=lambda t: t.created_at)
    total = len(tasks)
    # simple pagination
    start = (page - 1) * size
    end = start + size
    page_items = list(tasks)[start:end]
    return total, page_items

Update a task (partial update):

python 
def update_task(task: TaskModel, data: TaskUpdate) -> TaskModel:
    updated = False
    if data.title is not None:
        task.title = data.title
        updated = True
    if data.description is not None:
        task.description = data.description
        updated = True
    if data.completed is not None:
        task.completed = data.completed
        updated = True
    if updated:
        task.updated_at = datetime.utcnow()
    return task

Delete a task:

python 
def delete_task(task_id: int) -> None:
    _TASKS.pop(task_id, None)

Note:

Implementing the Tasks API Router

Now we connect our business logic to HTTP endpoints using FastAPI.

Create app/api/v1/tasks.py:

python 
from typing import Optional
from fastapi import APIRouter, HTTPException, Query, status
from app.schemas.task import Task, TaskCreate, TaskList, TaskUpdate
from app.services import tasks as task_service
router = APIRouter()

Creating Tasks (POST /tasks)

python 
@router.post(
    "/tasks",
    response_model=Task,
    status_code=status.HTTP_201_CREATED,
)
def create_task(task_in: TaskCreate) -> Task:
    # In a real app, owner_id would come from the authenticated user
    fake_owner_id = 1
    task_model = task_service.create_task(task_in, owner_id=fake_owner_id)
    return task_model

Listing Tasks with Pagination and Filtering (GET /tasks)

We want:

python 
@router.get(
    "/tasks",
    response_model=TaskList,
)
def list_tasks(
    page: int = Query(1, ge=1, description="Page number, starting from 1"),
    size: int = Query(10, ge=1, le=100, description="Page size"),
    completed: Optional[bool] = Query(
        None,
        description="Filter by completion status",
    ),
    owner_id: Optional[int] = Query(
        None,
        ge=1,
        description="Filter by owner id (for demo only)",
    ),
) -> TaskList:
    total, items = task_service.list_tasks(
        owner_id=owner_id,
        completed=completed,
        page=page,
        size=size,
    )
    return TaskList(
        total=total,
        page=page,
        size=size,
        items=items,
    )

Examples:

Getting a Single Task (GET /tasks/{task_id})

python 
@router.get(
    "/tasks/{task_id}",
    response_model=Task,
)
def get_task(task_id: int) -> Task:
    task = task_service.get_task(task_id)
    if task is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task not found",
        )
    return task

Updating a Task (PATCH /tasks/{task_id})

We use PATCH for partial updates. All fields are optional in TaskUpdate.

python 
@router.patch(
    "/tasks/{task_id}",
    response_model=Task,
)
def update_task(task_id: int, task_in: TaskUpdate) -> Task:
    task = task_service.get_task(task_id)
    if task is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task not found",
        )
    updated_task = task_service.update_task(task, task_in)
    return updated_task

You could also implement PUT for full replacement, but PATCH for partial fields is often more convenient for clients.

Deleting a Task (DELETE /tasks/{task_id})

python 
@router.delete(
    "/tasks/{task_id}",
    status_code=status.HTTP_204_NO_CONTENT,
)
def delete_task(task_id: int) -> None:
    task = task_service.get_task(task_id)
    if task is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task not found",
        )
    task_service.delete_task(task_id)
    # 204 means "No Content", so we return nothing

Summary of main endpoints:

HTTP MethodPathDescriptionRequest BodyResponse
POST/api/v1/tasksCreate a taskTaskCreate201 Created, body Task
GET/api/v1/tasksList tasks (paginated)None200 OK, body TaskList
GET/api/v1/tasks/{id}Get a single taskNone200 OK, body Task or 404
PATCH/api/v1/tasks/{id}Update a task (partial)TaskUpdate200 OK, body Task or 404
DELETE/api/v1/tasks/{id}Delete a taskNone204 No Content or 404

Validation and Error Handling in the API

Validation happens at multiple levels:

  1. Pydantic models
    For example, TaskCreate.title has min_length=1, max_length=100.
    If a client sends an empty title, FastAPI returns HTTP 422 automatically.
  2. Query parameter constraints
    We used Query(1, ge=1) for page and Query(10, ge=1, le=100) for size.
    If a client sends page=0, FastAPI will return a 422 validation error.
  3. Business rules
    You can add further checks in the service layer or API layer. For example, you may restrict that title cannot be “test” in production.
  4. Error responses using HTTPException
    We used HTTPException for 404 Not Found when a task is missing.

Rule: Always validate input at the boundary of your system, and always return clear HTTP status codes:

  • 201 for resource creation
  • 200 for successful retrieval or update
  • 204 for successful deletion without content
  • 400 or 422 for invalid input
  • 404 when a resource does not exist

Example of adding a simple business rule:

python 
from fastapi import HTTPException
@router.post("/tasks", response_model=Task, status_code=status.HTTP_201_CREATED)
def create_task(task_in: TaskCreate) -> Task:
    if task_in.title.lower() == "forbidden":
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="This title is not allowed",
        )
    fake_owner_id = 1
    task_model = task_service.create_task(task_in, owner_id=fake_owner_id)
    return task_model

Adding Simple Dependency Injection

FastAPI has a powerful dependency system. Even with an in‑memory store, we can get a taste of how it works.

Imagine we want a common function that fetches a task or raises 404. Instead of repeating this logic in every endpoint, we can make a dependency.

In app/api/v1/tasks.py:

python 
from fastapi import Depends
from app.models.task import TaskModel

Define the dependency:

python 
def get_task_or_404(task_id: int) -> TaskModel:
    task = task_service.get_task(task_id)
    if task is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task not found",
        )
    return task

Use it in endpoints:

python 
@router.get("/tasks/{task_id}", response_model=Task)
def get_task(task: TaskModel = Depends(get_task_or_404)) -> Task:
    return task
@router.patch("/tasks/{task_id}", response_model=Task)
def update_task(
    task_in: TaskUpdate,
    task: TaskModel = Depends(get_task_or_404),
) -> Task:
    updated_task = task_service.update_task(task, task_in)
    return updated_task
@router.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_task(task: TaskModel = Depends(get_task_or_404)) -> None:
    task_service.delete_task(task.id)

Benefits:

Basic Filtering, Sorting, and Pagination

Our list_tasks endpoint already has:

This is a common pattern for REST APIs.

The pagination formula is:

$$
\text{start} = (\text{page} - 1) \times \text{size}
$$

$$
\text{end} = \text{start} + \text{size}
$$

We then slice the list:

python 
page_items = tasks[start:end]

Rule: For any non‑trivial list endpoint, provide:

  • Pagination parameters, usually page and size
  • A total field in the response, to let clients know how many items exist
  • Stable sorting, typically by created_at or id

You can extend filters easily:

python 
def list_tasks(
    owner_id: Optional[int] = None,
    completed: Optional[bool] = None,
    search: Optional[str] = None,
    page: int = 1,
    size: int = 10,
) -> tuple[int, list[TaskModel]]:
    tasks: Sequence[TaskModel] = list(_TASKS.values())
    if owner_id is not None:
        tasks = [t for t in tasks if t.owner_id == owner_id]
    if completed is not None:
        tasks = [t for t in tasks if t.completed == completed]
    if search:
        lowered = search.lower()
        tasks = [
            t for t in tasks
            if lowered in t.title.lower()
            or (t.description and lowered in t.description.lower())
        ]
    tasks = sorted(tasks, key=lambda t: t.created_at)
    total = len(tasks)
    start = (page - 1) * size
    end = start + size
    page_items = list(tasks)[start:end]
    return total, page_items

And in the router:

python 
@router.get("/tasks", response_model=TaskList)
def list_tasks(
    page: int = Query(1, ge=1),
    size: int = Query(10, ge=1, le=100),
    completed: Optional[bool] = Query(None),
    owner_id: Optional[int] = Query(None, ge=1),
    search: Optional[str] = Query(None, min_length=1),
) -> TaskList:
    total, items = task_service.list_tasks(
        owner_id=owner_id,
        completed=completed,
        search=search,
        page=page,
        size=size,
    )
    return TaskList(total=total, page=page, size=size, items=items)

Testing the API Manually

With the API running via Uvicorn, you can test each endpoint.

  1. Create a task
    • Method: POST
    • URL: http://127.0.0.1:8000/api/v1/tasks
    • Body:
json 
     {
       "title": "Learn FastAPI",
       "description": "Build a complete REST API",
       "completed": false
     }
  1. List tasks
    • Method: GET
    • URL: http://127.0.0.1:8000/api/v1/tasks?page=1&size=5
    • Expected: 200 OK and a TaskList object:
json 
     {
       "total": 1,
       "page": 1,
       "size": 5,
       "items": [
         {
           "id": 1,
           "title": "Learn FastAPI",
           "description": "Build a complete REST API",
           "completed": false,
           "owner_id": 1,
           "created_at": "2024-01-01T12:00:00",
           "updated_at": "2024-01-01T12:00:00"
         }
       ]
     }
  1. Get a single task
    • Method: GET
    • URL: http://127.0.0.1:8000/api/v1/tasks/1
  2. Update a task
    • Method: PATCH
    • URL: http://127.0.0.1:8000/api/v1/tasks/1
    • Body:
json 
     {
       "completed": true
     }
  1. Delete a task
    • Method: DELETE
    • URL: http://127.0.0.1:8000/api/v1/tasks/1

Evolving This API Further

The example so far is fully functional, but uses an in‑memory store and a fake owner_id. In a real project, you would extend it by:

The important part is the pattern:

  1. Schemas for input and output
  2. Models (ORM or otherwise) for persistent data
  3. Services that contain the actual logic
  4. Routers that translate HTTP requests to service calls and back
  5. Main app that ties everything together and exposes a consistent REST API

Key idea: A “complete REST API” in FastAPI is not just a collection of routes. It is:

  • Clear resource definitions and URLs
  • Thoughtful request and response models
  • Proper HTTP methods and status codes
  • Validation and error handling
  • A structure that can grow as your application grows

With this structure and these patterns, you can confidently build more complex REST APIs on top of FastAPI.

Views: 7

Comments

Please login to add a comment.

Don't have an account? Register now!

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