FastAPI Tutorial: Build a REST API in Python Step by Step
Build a complete task API with FastAPI: typed request and response models, a database session per request, proper status codes, and tests that need no server.
A typical hand-written endpoint spends most of its lines on chores. It parses JSON, checks that fields exist, converts types, builds error responses, and formats output. The actual business rule is three lines in the middle. FastAPI removes the chores: you declare the shape of the input and output, and the framework does the rest.
FastAPI is a Python web framework for building APIs. This FastAPI tutorial produces a small service called taskapi, with endpoints to create, list, read, update, and delete tasks. A REST API exposes resources, here tasks, at URLs, and uses HTTP methods such as GET and POST to act on them.
You need Python 3.11. The code targets FastAPI 0.100 to 0.103, which use Pydantic 2. Two background ideas help. A Pydantic model is a class that validates data against type annotations, covered in depth in Pydantic v2 models and validators. An ORM session is a short-lived workspace for database changes, explained in SQLAlchemy 2.0 sessions and select(). This article shows all the code you need for both.
My position: FastAPI’s value is its type-driven contract, not its benchmark numbers. Your database and your own code decide how fast the API is. The framework decides how much boilerplate you maintain.
Step 1: install FastAPI and the supporting packages
python --version
mkdir taskapi
cd taskapi
python -m venv .venv
source .venv/bin/activate # macOS and Linux
.venv\Scripts\Activate.ps1 # Windows PowerShell
python -m pip install "fastapi>=0.100,<0.104" "uvicorn[standard]" "sqlalchemy>=2.0,<2.1" httpx pytest
FastAPI is the framework. Uvicorn is the server that runs it. SQLAlchemy talks to the database, and SQLite ships with Python. httpx and pytest are for the tests.
Step 2: write the application
The whole service fits in one file. Save it as main.py. The sections that follow explain each part.
from collections.abc import Iterator
from contextlib import asynccontextmanager
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException, Query, status
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy import String, create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, sessionmaker
DATABASE_URL = 'sqlite:///tasks.db'
engine = create_engine(DATABASE_URL, connect_args={'check_same_thread': False})
SessionLocal = sessionmaker(engine, expire_on_commit=False)
# --- database model -------------------------------------------------------
class Base(DeclarativeBase):
pass
class TaskRow(Base):
__tablename__ = 'task'
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str] = mapped_column(String(200))
done: Mapped[bool] = mapped_column(default=False)
# --- API schemas ----------------------------------------------------------
class TaskCreate(BaseModel):
model_config = ConfigDict(extra='forbid', str_strip_whitespace=True)
title: str = Field(min_length=3, max_length=200)
class TaskUpdate(BaseModel):
model_config = ConfigDict(extra='forbid', str_strip_whitespace=True)
title: str | None = Field(default=None, min_length=3, max_length=200)
done: bool | None = None
class TaskRead(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
title: str
done: bool
# --- application and dependencies -----------------------------------------
@asynccontextmanager
async def lifespan(app: FastAPI):
Base.metadata.create_all(engine)
yield
engine.dispose()
app = FastAPI(title='taskapi', version='0.1.0', lifespan=lifespan)
def get_session() -> Iterator[Session]:
with SessionLocal() as session:
yield session
SessionDep = Annotated[Session, Depends(get_session)]
def get_task_or_404(task_id: int, session: SessionDep) -> TaskRow:
task = session.get(TaskRow, task_id)
if task is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, detail=f'task {task_id} not found')
return task
TaskDep = Annotated[TaskRow, Depends(get_task_or_404)]
# --- endpoints ------------------------------------------------------------
@app.post('/tasks', response_model=TaskRead, status_code=status.HTTP_201_CREATED)
def create_task(payload: TaskCreate, session: SessionDep) -> TaskRow:
task = TaskRow(title=payload.title)
session.add(task)
session.commit()
return task
@app.get('/tasks', response_model=list[TaskRead])
def list_tasks(
session: SessionDep,
done: bool | None = None,
limit: Annotated[int, Query(ge=1, le=100)] = 20,
offset: Annotated[int, Query(ge=0)] = 0,
) -> list[TaskRow]:
statement = select(TaskRow).order_by(TaskRow.id)
if done is not None:
statement = statement.where(TaskRow.done.is_(done))
return list(session.scalars(statement.limit(limit).offset(offset)))
@app.get('/tasks/{task_id}', response_model=TaskRead)
def read_task(task: TaskDep) -> TaskRow:
return task
@app.patch('/tasks/{task_id}', response_model=TaskRead)
def update_task(payload: TaskUpdate, task: TaskDep, session: SessionDep) -> TaskRow:
changes = payload.model_dump(exclude_unset=True, exclude_none=True)
for field, value in changes.items():
setattr(task, field, value)
session.commit()
return task
@app.delete('/tasks/{task_id}', status_code=status.HTTP_204_NO_CONTENT)
def delete_task(task: TaskDep, session: SessionDep) -> None:
session.delete(task)
session.commit()
Step 3: run it and try the endpoints
uvicorn main:app --reload
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: Application startup complete.
main:app means “the object named app in main.py“. The --reload flag restarts the server when you save a file. Use it only during development.
Open http://127.0.0.1:8000/docs in a browser. FastAPI generates an interactive page from your code, where you can send requests without any other tool. From a second terminal on macOS or Linux, you can also use curl:
curl -X POST http://127.0.0.1:8000/tasks \
-H "Content-Type: application/json" \
-d '{"title": "Write docs"}'
{"id":1,"title":"Write docs","done":false}
curl -X PATCH http://127.0.0.1:8000/tasks/1 \
-H "Content-Type: application/json" \
-d '{"done": true}'
{"id":1,"title":"Write docs","done":true}
curl "http://127.0.0.1:8000/tasks?done=true&limit=10"
[{"id":1,"title":"Write docs","done":true}]
On Windows, quoting JSON in a shell is awkward, so use the /docs page instead.
How the pieces work
Parameters decided by type
FastAPI reads each endpoint’s signature and decides where every parameter comes from.
| Parameter | FastAPI treats it as | Example in the code |
|---|---|---|
Name appears in the path, such as {task_id} |
Path parameter | task_id: int |
| A Pydantic model | JSON request body | payload: TaskCreate |
| A simple type not in the path | Query parameter | done, limit, offset |
Annotated with Depends(...) |
Dependency, computed by FastAPI | session: SessionDep |
Each one is validated and converted before your function runs. /tasks/abc never reaches get_task_or_404, because abc is not an integer. limit=500 is rejected by Query(le=100). Your function body can therefore assume its arguments are valid.
Separate models for one resource
The code defines four classes for tasks, and each has a separate job. TaskRow is the database table. TaskCreate is what a client may send when creating. TaskUpdate makes every field optional for partial updates. TaskRead is what the API returns.
The separation is a security measure, not ceremony. A client cannot set id or done at creation, because TaskCreate has no such fields and extra='forbid' rejects unknown ones. A column you add to the table later stays private until you add it to TaskRead.
The response_model argument tells FastAPI to convert the returned TaskRow into a TaskRead. from_attributes=True lets Pydantic read the ORM object’s attributes.
Dependencies: one session per request
A dependency is a function that FastAPI calls for you, passing the result into your endpoint. get_session opens a session and yields it. When the response is finished, FastAPI resumes the function, and the with block closes the session, even if the endpoint raised an error.
request arrives
|
v
get_session() opens a Session and yields it
|
v
get_task_or_404() receives the SAME session, loads the task or raises 404
|
v
update_task() receives the same session and the loaded task
|
v
response sent, then get_session() resumes and closes the Session
Here is the detail that makes this pattern work, and the documentation mentions it only in passing. Within one request, FastAPI calls each dependency once and reuses the result. update_task asks for a session directly and also through get_task_or_404, and both receive the same object. Without that caching, the task would be loaded in one session and committed in another, and the update would be lost.
The SessionDep and TaskDep aliases keep signatures short. get_task_or_404 also removes repetition: three endpoints need “load this task or return 404”, and none of them contains that logic.
Status codes and errors
Use the status code that matches the outcome. Creation returns 201 Created. Deletion returns 204 No Content with an empty body. To signal a failure, raise HTTPException:
curl -i http://127.0.0.1:8000/tasks/99
HTTP/1.1 404 Not Found
content-type: application/json
{"detail":"task 99 not found"}
Validation failures need no code from you. FastAPI returns 422 Unprocessable Entity with one entry per problem:
curl -X POST http://127.0.0.1:8000/tasks \
-H "Content-Type: application/json" \
-d '{"title": "ab"}'
{"detail":[{"type":"string_too_short","loc":["body","title"],"msg":"String should have at least 3 characters","input":"ab","ctx":{"min_length":3},"url":"https://errors.pydantic.dev/2.3/v/string_too_short"}]}
The loc field tells the client exactly which part of the request was wrong.
Partial updates with PATCH
payload.model_dump(exclude_unset=True, exclude_none=True) returns only the fields the client actually sent, minus explicit nulls. A request with {"done": true} changes done and leaves the title alone. Without exclude_unset, the missing title would arrive as None and overwrite the stored value.
Startup and shutdown with lifespan
The lifespan function runs once when the server starts, up to the yield, and once when it stops. Here it creates the table and later releases database connections. create_all is fine for a tutorial. A real service manages schema changes with a migration tool such as Alembic.
Step 4: test the API
FastAPI ships a TestClient that calls your application in-process, with no server and no network. The tests replace the database dependency with an in-memory SQLite database, so each test starts clean. If pytest fixtures are new to you, a fixture is a function that prepares something a test needs. Save this as test_main.py.
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
from main import Base, app, get_session
@pytest.fixture
def client():
engine = create_engine(
'sqlite://',
connect_args={'check_same_thread': False},
poolclass=StaticPool,
)
Base.metadata.create_all(engine)
TestSession = sessionmaker(engine, expire_on_commit=False)
def override_get_session():
with TestSession() as session:
yield session
app.dependency_overrides[get_session] = override_get_session
yield TestClient(app)
app.dependency_overrides.clear()
engine.dispose()
def test_create_and_read_task(client):
created = client.post('/tasks', json={'title': 'Write docs'})
assert created.status_code == 201
assert created.json() == {'id': 1, 'title': 'Write docs', 'done': False}
assert client.get('/tasks/1').json()['title'] == 'Write docs'
def test_short_title_is_rejected(client):
response = client.post('/tasks', json={'title': 'ab'})
assert response.status_code == 422
assert response.json()['detail'][0]['loc'] == ['body', 'title']
def test_missing_task_returns_404(client):
assert client.get('/tasks/99').status_code == 404
def test_patch_updates_only_sent_fields(client):
client.post('/tasks', json={'title': 'Write docs'})
response = client.patch('/tasks/1', json={'done': True})
assert response.json() == {'id': 1, 'title': 'Write docs', 'done': True}
def test_filter_by_done(client):
client.post('/tasks', json={'title': 'First task'})
client.post('/tasks', json={'title': 'Second task'})
client.patch('/tasks/2', json={'done': True})
response = client.get('/tasks', params={'done': 'true'})
assert [task['id'] for task in response.json()] == [2]
python -m pytest -q
..... [100%]
5 passed in 0.31s
app.dependency_overrides is the reason to use dependencies in the first place. The endpoints ask for “a session” and do not care where it comes from, so a test can supply a different one without patching anything. StaticPool keeps a single connection, which an in-memory SQLite database needs in order to keep its data. The fixture creates TestClient(app) without a with block, so the lifespan function does not run and no tasks.db file appears during tests.
The myth: async def makes your endpoints faster
FastAPI is an async framework, so many tutorials write every endpoint as async def. People then assume that async is the fast option. It is the right option only under one condition.
FastAPI treats the two forms differently. An async def endpoint runs directly on the asyncio event loop, the single thread that serves every request. A plain def endpoint runs in a thread pool, so a blocking call inside it holds up one thread and nothing else.
# WRONG: a blocking database call inside async def freezes the whole server
@app.get('/tasks/{task_id}')
async def read_task(task_id: int, session: SessionDep):
return session.get(TaskRow, task_id) # blocks the event loop
# RIGHT: plain def, so FastAPI runs it in a worker thread
@app.get('/tasks/{task_id}')
def read_task(task_id: int, session: SessionDep):
return session.get(TaskRow, task_id)
In the wrong version, every request waits while one request talks to the database. Under load, throughput collapses, and no error appears anywhere. This tutorial uses a synchronous SQLAlchemy session, so every endpoint is a plain def.
My rule for choosing: write async def only when every I/O call in the function is awaited through an async library. If even one call blocks, use def. A mixed endpoint is slower than either pure form.
Troubleshooting
Error loading ASGI app. Could not import module "main". Runuvicornfrom the folder that containsmain.py, and check themodule:objectspelling.SQLite objects created in a thread can only be used in that same thread. You removedcheck_same_thread=False. Sync endpoints run in a thread pool, so SQLite needs that setting.- Every request returns 422. The body is not valid JSON, or the
Content-Type: application/jsonheader is missing. Read thelocfield in the response. - A field is missing from the response.
response_modelfilters output to its declared fields. Add the field toTaskRead. Address already in use. Another process holds port 8000. Stop it, or pass--port 8001.
How real systems structure FastAPI services
- Routers per resource. Larger applications split endpoints into modules with
APIRouter, such asrouters/tasks.py, and include them in the main application. - Thin endpoints. An endpoint validates input, calls a service function, and shapes the response. Business rules live in plain functions that tests can call without HTTP.
- Settings from the environment. The database URL and secrets come from a settings object loaded at startup, never from constants in the code.
- Migrations, not create_all. Schema changes run as versioned migrations during deployment.
- Several worker processes. In production, the service runs as multiple server processes behind a process manager or container platform, so that one busy worker does not stall the API.
A mistake I have seen in production is a FastAPI service where every endpoint was async def and every database call used a synchronous driver. It passed all tests and handled light traffic well. At a few dozen concurrent users, response times jumped from tens of milliseconds to several seconds, because each query blocked the event loop for everyone. Removing the word async from the endpoint definitions restored normal latency the same day.
Choosing sync or async endpoints: a decision framework
- Is your database driver synchronous? Use
defendpoints. This covers most projects that start with a standard SQLAlchemy session. - Do you call other services with an async client, and use an async database driver? Use
async defandawaitevery call. - Does one endpoint need a blocking library inside async code? Make that endpoint a plain
def, or push the call to a thread. - Does the endpoint do heavy computation? Neither form helps. Move the work to a background worker or a separate process.
- Unsure? Use
def. It is the safe default, and it is rarely the bottleneck.
When NOT to use FastAPI
- Server-rendered sites with an admin interface. If you need templates, forms, authentication, and an admin panel out of the box, a full-stack framework such as Django gives you those on day one.
- A script that exposes one internal function. A queue consumer or a scheduled job may be simpler than running an HTTP service at all.
- Teams not using type hints. FastAPI’s benefits come from annotations. Without them, you keep the framework’s complexity and lose its validation and documentation.
Common mistakes
- Blocking calls inside async def. One slow query stalls every request on the server. Latency spikes appear on unrelated endpoints.
- Returning ORM objects without a response model. Every column goes to the client, including ones you meant to keep private.
- One model for input and output. Clients can set server-owned fields such as
id, and password hashes can leak into responses. - A global session. A session shared across requests mixes transactions between users. Use a dependency that yields a session per request.
- Running with –reload in production. The reloader watches the file system and restarts on changes. It wastes resources and is not built for production traffic.
- Using model_dump() for PATCH without exclude_unset. Fields the client did not send become
Noneand erase stored values.
Key takeaways
- Declare inputs and outputs with types, and FastAPI validates, converts, and documents them.
- Keep separate models for create, update, read, and the database table.
- Provide the database session through a
yielddependency, one per request. - Dependencies are resolved once per request, so nested dependencies share the same session.
- Use
defendpoints with synchronous libraries, andasync defonly with fully async I/O. - Raise
HTTPExceptionfor expected failures, and return precise status codes. - Test with
TestClientanddependency_overridesagainst an in-memory database.
FAQ
What is FastAPI used for?
FastAPI is a Python framework for building web APIs. It uses type hints to validate requests, serialize responses, and generate interactive API documentation automatically.
How do I run a FastAPI app?
Install FastAPI and Uvicorn, then run uvicorn main:app --reload, where main is your file name and app is the FastAPI object. The API is served at http://127.0.0.1:8000.
Should I use async def or def in FastAPI?
Use def when the endpoint calls blocking libraries, such as a synchronous database driver. Use async def only when all I/O inside the endpoint is awaited through async libraries.
How do I connect FastAPI to a database?
Create a SQLAlchemy engine and session factory, write a dependency that yields a session, and declare that dependency in each endpoint. FastAPI opens and closes one session per request.
How do I test a FastAPI application?
Use fastapi.testclient.TestClient with pytest. Override the database dependency through app.dependency_overrides, so that tests run against an in-memory database.
Declare the contract, keep the endpoints thin
FastAPI works best when the types carry the contract and the functions carry only the logic. Separate models guard what enters and leaves. Dependencies supply what each request needs and make tests simple. Choose def or async def by what your libraries actually do.
Rule of thumb: if an endpoint is longer than a dozen lines, the missing piece is usually a model, a dependency, or a service function.
