Build a Production-Ready CRUD API with FastAPI (Validation, DB, Pagination, and Error Handling)

Pictalingo

·

Build a Production-Ready CRUD API with FastAPI (Validation, DB, Pagination, and Error Handling)

FastAPI is a modern Python framework for building APIs quickly, with automatic request validation, interactive docs, and great performance. In this hands-on guide, you’ll build a small “Tasks” API with:

  • Strong input validation via Pydantic models
  • Database persistence with SQLModel (SQLAlchemy under the hood)
  • Clean error handling (404s, validation errors)
  • Pagination and filtering
  • Practical curl examples

You can use this as a starter template for side projects or internal tools.

Project setup

Create a folder and install dependencies:

mkdir fastapi-tasks cd fastapi-tasks python -m venv .venv # macOS/Linux: source .venv/bin/activate # Windows: # .venv\Scripts\activate pip install fastapi uvicorn sqlmodel

We’ll keep everything in a single file to stay focused. Create main.py.

Define models and database session

We’ll use SQLite for local development. SQLModel makes it easy to define “table models” (for persistence) and “schema models” (for input/output). This prevents accidental writes to fields you don’t want clients controlling.

from datetime import datetime from typing import Optional from fastapi import FastAPI, HTTPException, Query from sqlmodel import Field, SQLModel, Session, create_engine, select app = FastAPI(title="Tasks API", version="1.0.0") # --- Database setup --- DATABASE_URL = "sqlite:///./tasks.db" engine = create_engine(DATABASE_URL, echo=False) # --- Models --- class TaskBase(SQLModel): title: str = Field(min_length=1, max_length=200) description: Optional[str] = Field(default=None, max_length=2000) is_done: bool = False priority: int = Field(default=3, ge=1, le=5) # 1 (high) .. 5 (low) class Task(TaskBase, table=True): id: Optional[int] = Field(default=None, primary_key=True) created_at: datetime = Field(default_factory=datetime.utcnow) class TaskCreate(TaskBase): pass class TaskUpdate(SQLModel): title: Optional[str] = Field(default=None, min_length=1, max_length=200) description: Optional[str] = Field(default=None, max_length=2000) is_done: Optional[bool] = None priority: Optional[int] = Field(default=None, ge=1, le=5) class TaskRead(TaskBase): id: int created_at: datetime def init_db() -> None: SQLModel.metadata.create_all(engine) @app.on_event("startup") def on_startup() -> None: init_db()

Key ideas:

  • Task is the DB table model (has id and created_at).
  • TaskCreate is what clients are allowed to send when creating a task.
  • TaskUpdate uses optional fields so you can PATCH only what you need.
  • TaskRead is what you return to clients.

Create a task (POST)

This endpoint validates input automatically. If the client sends invalid types or out-of-range values (like priority: 99), FastAPI returns a 422 with details.

@app.post("/tasks", response_model=TaskRead, status_code=201) def create_task(payload: TaskCreate) -> Task: task = Task.model_validate(payload) # copy validated fields into Task with Session(engine) as session: session.add(task) session.commit() session.refresh(task) return task

Try it:

uvicorn main:app --reload
curl -X POST "http://127.0.0.1:8000/tasks" \ -H "Content-Type: application/json" \ -d '{"title":"Ship MVP","description":"Release v1 to users","priority":2}'

Open the interactive docs at /docs to test endpoints without leaving the browser.

List tasks with pagination and filtering (GET)

Junior devs often implement “list everything” endpoints that become slow as data grows. Start with pagination early.

This endpoint supports:

  • limit and offset for pagination
  • is_done to filter completed vs pending tasks
  • Sorting by newest first
@app.get("/tasks", response_model=list[TaskRead]) def list_tasks( limit: int = Query(default=20, ge=1, le=100), offset: int = Query(default=0, ge=0), is_done: Optional[bool] = None, ) -> list[Task]: stmt = select(Task) if is_done is not None: stmt = stmt.where(Task.is_done == is_done) stmt = stmt.order_by(Task.created_at.desc()).offset(offset).limit(limit) with Session(engine) as session: return list(session.exec(stmt).all())

Try it:

curl "http://127.0.0.1:8000/tasks?limit=10&offset=0" curl "http://127.0.0.1:8000/tasks?is_done=false"

Get one task (GET /tasks/{id}) with clean 404s

When an ID doesn’t exist, return a 404 with a helpful message.

@app.get("/tasks/{task_id}", response_model=TaskRead) def get_task(task_id: int) -> Task: with Session(engine) as session: task = session.get(Task, task_id) if not task: raise HTTPException(status_code=404, detail="Task not found") return task

Update a task safely (PATCH)

PATCH should update only the fields the client provides. A common bug is overwriting fields with null accidentally. SQLModel/Pydantic can help by excluding unset values.

@app.patch("/tasks/{task_id}", response_model=TaskRead) def update_task(task_id: int, payload: TaskUpdate) -> Task: with Session(engine) as session: task = session.get(Task, task_id) if not task: raise HTTPException(status_code=404, detail="Task not found") updates = payload.model_dump(exclude_unset=True) for key, value in updates.items(): setattr(task, key, value) session.add(task) session.commit() session.refresh(task) return task

Try it:

curl -X PATCH "http://127.0.0.1:8000/tasks/1" \ -H "Content-Type: application/json" \ -d '{"is_done": true}'

Delete a task (DELETE)

For deletes, a 204 No Content response is common. Here we’ll return a small confirmation payload to keep it beginner-friendly.

@app.delete("/tasks/{task_id}") def delete_task(task_id: int) -> dict: with Session(engine) as session: task = session.get(Task, task_id) if not task: raise HTTPException(status_code=404, detail="Task not found") session.delete(task) session.commit() return {"deleted": True, "task_id": task_id}

Common “production-ish” improvements

  • Move config to environment variables. For example, read DATABASE_URL from os.environ so you can swap SQLite for Postgres in deployment.

  • Add migrations. create_all() is fine for demos, but real projects use migrations (often with Alembic) to manage schema changes safely.

  • Split code into modules. A typical layout is models.py, db.py, routers/, and schemas.py.

  • Add indexes for query speed. If you frequently filter by is_done or sort by created_at, indexing those columns helps at scale.

  • Introduce auth. Even simple APIs usually need auth (API keys, OAuth2, or session-based auth depending on the app).

Quick sanity checks you should run

Before calling an API “done,” confirm these behaviors:

  • Invalid payloads return 422 with clear validation details (try {"priority": 999}).

  • Missing resources return 404 (try /tasks/999999).

  • Pagination works and does not return the entire table by default.

  • PATCH updates only provided fields (doesn’t wipe fields you omit).

Next steps

You now have a working CRUD API with validation, a real database, and practical patterns you can reuse. From here, try adding:

  • A GET /health endpoint for deployments and uptime checks

  • Unit tests with pytest and FastAPI’s TestClient

  • PostgreSQL in Docker for local dev

  • Structured logging and request IDs for debugging

If you want, I can provide a version with a proper folder structure (routers, services, schemas) and automated tests so it’s closer to what you’d push to a real repo.

Leave a Reply

Your email address will not be published. Required fields are marked *