FastAPI Introduction
Before you start
You need: OOP (Article 08), databases and APIs (Article 10).
Time: about 55 minutes, plus the practice.
Learning objective
Build a small database-backed REST API with validated request and response models, correct status codes and useful errors.
Topics
- Setup and the first route
- Path, query and body parameters
- Pydantic models
- Status codes and responses
- Dependency injection
- Errors
- Database access
- Configuration
- Automatic documentation
Setup
pip install "fastapi[standard]"
# src/school/api/main.py
from fastapi import FastAPI
app = FastAPI(
title="NexCoding School Portal API",
description="Student, exam and fee management",
version="1.0.0"
)
@app.get("/health")
def health() -> dict[str, str]:
return {"status": "healthy"}
fastapi dev src/school/api/main.py
Open http://127.0.0.1:8000/docs — interactive documentation, generated from the code, with a working "Try it out" for every endpoint.
That documentation is generated from type hints. Nothing is written by hand, and it cannot drift from the implementation — which is FastAPI's main advantage.
fastapi dev reloads on save. For production, fastapi run or uvicorn behind a process manager.
Parameters
from fastapi import FastAPI, Query, Path
app = FastAPI()
@app.get("/api/students/{public_id}")
def get_student(public_id: UUID) -> StudentResponse:
...
@app.get("/api/students")
def search_students(
term: str | None = Query(default=None, max_length=100),
class_name: str | None = Query(default=None, alias="className"),
page: int = Query(default=1, ge=1),
page_size: int = Query(default=20, ge=1, le=100)
) -> PagedResponse[StudentResponse]:
...
A parameter named in the path is a path parameter; everything else is a query parameter. No attributes needed for the common case.
The type hint does the conversion and validation. public_id: UUID means a non-UUID value returns 422 before your function runs.
ge=1, le=100 on page_size is a real control — without it a client requests pageSize=1000000 and takes the API down.
alias="className" accepts the camelCase name JavaScript clients send while keeping snake_case in Python.
Pydantic models
from datetime import date
from decimal import Decimal
from pydantic import BaseModel, Field, field_validator
class StudentCreateRequest(BaseModel):
name: str = Field(min_length=2, max_length=100)
roll_number: str = Field(pattern=r"^NCA-\d{4}-\d{4}$", alias="rollNumber")
class_name: str = Field(max_length=10, alias="className")
section: str = Field(pattern=r"^[ABC]$")
date_of_birth: date = Field(alias="dateOfBirth")
parent_name: str = Field(max_length=100, alias="parentName")
parent_phone: str = Field(pattern=r"^[6-9]\d{9}$", alias="parentPhone")
address: str | None = Field(default=None, max_length=250)
model_config = {"populate_by_name": True}
@field_validator("date_of_birth")
@classmethod
def check_age(cls, value: date) -> date:
age = (date.today() - value).days // 365
if not 3 <= age <= 25:
raise ValueError("Date of birth gives an age outside the accepted range")
return value
class StudentResponse(BaseModel):
public_id: UUID = Field(serialization_alias="publicId")
name: str
roll_number: str = Field(serialization_alias="rollNumber")
class_name: str = Field(serialization_alias="className")
section: str
parent_phone: str = Field(serialization_alias="parentPhone")
model_config = {"from_attributes": True}
Request and response models are separate, and neither is a database entity. Three reasons:
- Over-posting. A model containing
school_idorstatuslets a client set them — moving a student to another school. - Accidental disclosure. Returning the entity means adding a
password_hashcolumn publishes it to every client on the next deployment. - Coupling. A database rename becomes a breaking API change.
from_attributes = True lets a response model be built from an object rather than a dict:
return StudentResponse.model_validate(student)
Validation is automatic. An invalid body returns 422 with a field-level error list before the function runs:
{
"detail": [
{
"type": "string_pattern_mismatch",
"loc": ["body", "rollNumber"],
"msg": "String should match pattern '^NCA-\\d{4}-\\d{4}$'",
"input": "12345"
}
]
}
Cross-field rules use a model validator:
from pydantic import model_validator
class ExamResultRequest(BaseModel):
student_id: int
marks_obtained: Decimal | None = None
is_absent: bool = False
@model_validator(mode="after")
def check_marks_or_absent(self) -> "ExamResultRequest":
if self.is_absent and self.marks_obtained is not None:
raise ValueError("An absent student cannot have marks")
if not self.is_absent and self.marks_obtained is None:
raise ValueError("Enter marks, or mark the student absent")
return self
That is the absent-versus-zero rule enforced at the API boundary — the same rule the database CHECK constraint enforces underneath.
Status codes
from fastapi import status
@app.post("/api/students", status_code=status.HTTP_201_CREATED)
def create_student(request: StudentCreateRequest, response: Response) -> StudentResponse:
student = service.create(request)
response.headers["Location"] = f"/api/students/{student.public_id}"
return StudentResponse.model_validate(student)
@app.put("/api/students/{public_id}", status_code=status.HTTP_204_NO_CONTENT)
def update_student(public_id: UUID, request: StudentUpdateRequest) -> None:
...
@app.delete("/api/students/{public_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_student(public_id: UUID) -> None:
...
| Code | When |
|---|---|
| 200 | GET, or an update returning the result |
| 201 | POST succeeded — include Location |
| 204 | Successful PUT or DELETE with no body |
| 400 | A domain rule was broken |
| 401 | Not authenticated |
| 403 | Authenticated, not permitted |
| 404 | Not found |
| 409 | Conflict — duplicate |
| 422 | Validation failed — FastAPI's default |
204 for an update or delete with nothing to return, not 200 with an empty object. A JavaScript client calling .json() on a 200 with no body gets a parse error.
FastAPI uses 422 rather than 400 for request validation. That is deliberate — 400 stays available for domain failures — and it surprises clients expecting 400, so document it.
Dependency injection
from typing import Annotated
from fastapi import Depends
def get_connection() -> Iterator[Connection]:
connection = create_connection(settings.database_url)
try:
yield connection
finally:
connection.close()
ConnectionDep = Annotated[Connection, Depends(get_connection)]
@app.get("/api/students")
def search_students(connection: ConnectionDep, term: str | None = None) -> list[StudentResponse]:
...
A dependency is a function; FastAPI resolves it per request. Code after yield runs as teardown, even when the endpoint raises — which is what guarantees the connection is closed.
The Annotated alias avoids repeating Depends(...) in every signature.
Dependencies chain:
def get_current_user(
connection: ConnectionDep,
token: Annotated[str, Depends(oauth2_scheme)]
) -> CurrentUser:
user = decode_and_load(token, connection)
if user is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or expired token",
headers={"WWW-Authenticate": "Bearer"}
)
return user
CurrentUserDep = Annotated[CurrentUser, Depends(get_current_user)]
@app.get("/api/students")
def search_students(user: CurrentUserDep, connection: ConnectionDep) -> list[StudentResponse]:
return service.search(connection, school_id=user.school_id)
school_id comes from the authenticated user, never from the request. A school_id query parameter is a request from the client to choose whose data to read — the same rule as every other track.
For a role check, a dependency with no return value:
def require_role(*roles: str):
def check(user: CurrentUserDep) -> None:
if user.role not in roles:
raise HTTPException(status.HTTP_403_FORBIDDEN, "Not permitted")
return check
@app.delete("/api/students/{public_id}", dependencies=[Depends(require_role("Admin"))])
def delete_student(public_id: UUID) -> None:
...
Errors
from fastapi import HTTPException
@app.get("/api/students/{public_id}")
def get_student(public_id: UUID, user: CurrentUserDep, connection: ConnectionDep) -> StudentResponse:
student = repository.get_by_public_id(connection, user.school_id, public_id)
if student is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, "Student not found")
return StudentResponse.model_validate(student)
Return 404, not 403, for another tenant's record. A 403 confirms the record exists; 404 reveals nothing. The tenant filter in the query does both jobs at once.
Handling domain exceptions once, globally:
from fastapi.responses import JSONResponse
@app.exception_handler(DuplicateRollNumberError)
def handle_duplicate(request: Request, error: DuplicateRollNumberError) -> JSONResponse:
return JSONResponse(
status_code=status.HTTP_409_CONFLICT,
content={"detail": str(error), "rollNumber": error.roll_number}
)
@app.exception_handler(Exception)
def handle_unexpected(request: Request, error: Exception) -> JSONResponse:
logger.exception("Unhandled error on %s %s", request.method, request.url.path)
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
content={"detail": "An unexpected error occurred."}
)
A 500 must never return the exception message. It can name tables, columns, file paths and library versions. Log the detail; return a generic message.
Adding a correlation id makes the generic message actionable:
@app.middleware("http")
async def add_correlation_id(request: Request, call_next):
correlation_id = request.headers.get("X-Correlation-Id", str(uuid4()))
with logging_context(correlation_id=correlation_id):
response = await call_next(request)
response.headers["X-Correlation-Id"] = correlation_id
return response
The user quotes the id in a support ticket and someone finds the exact request in the logs.
Database access
from fastapi import APIRouter
router = APIRouter(prefix="/api/students", tags=["students"])
@router.get("")
def search_students(
user: CurrentUserDep,
connection: ConnectionDep,
term: str | None = Query(default=None, max_length=100),
page: int = Query(default=1, ge=1),
page_size: int = Query(default=20, ge=1, le=100)
) -> PagedResponse[StudentResponse]:
result = repository.search(connection, user.school_id, term, page, page_size)
return PagedResponse(
items=[StudentResponse.model_validate(s) for s in result.items],
total_count=result.total_count,
page=page,
page_size=page_size
)
@router.post("", status_code=status.HTTP_201_CREATED)
def create_student(
request: StudentCreateRequest,
user: CurrentUserDep,
connection: ConnectionDep
) -> StudentResponse:
student = service.create(connection, user.school_id, request)
return StudentResponse.model_validate(student)
app.include_router(router)
APIRouter splits endpoints across files. tags groups them in the documentation.
The endpoint calls a service; the service calls a repository. The endpoint handles HTTP concerns — status codes, models, the current user — and nothing else. That layering is what lets the same service run from a scheduled job or a console script.
Sync versus async
@app.get("/api/students")
def search_students(...) -> list[StudentResponse]: # runs in a threadpool
...
@app.get("/api/students")
async def search_students(...) -> list[StudentResponse]: # runs on the event loop
...
A def endpoint runs in a threadpool, so blocking database calls are fine. An async def endpoint runs on the event loop, and a blocking call inside it stalls the entire server:
# Wrong — blocks every other request
@app.get("/api/students")
async def search_students():
return blocking_database_query()
Rule: async def only with await-able libraries throughout. With pyodbc or sqlite3, use plain def — FastAPI handles it correctly.
Configuration
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
database_url: str
jwt_secret: str
jwt_expiry_minutes: int = 60
cors_origins: list[str] = []
model_config = {"env_file": ".env"}
settings = Settings()
Settings() raises at startup if anything required is missing. A misconfigured deployment then fails immediately with a message naming the setting, rather than on the first request that needs it.
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"]
)
Never allow_origins=["*"]. It opens the API to every site, and it is incompatible with allow_credentials=True.
CORS is a browser restriction enforced on the server's response. It cannot be fixed in the frontend, and it does not appear in Postman — which is why "it works in Postman" is not evidence the API is fine.
Never commit .env. A credential in Git history is permanent, and the fix is rotating it.
Documentation
@router.post("", status_code=status.HTTP_201_CREATED,
summary="Create a student",
responses={
409: {"description": "The roll number is already in use"},
422: {"description": "Validation failed"}
})
def create_student(request: StudentCreateRequest, ...) -> StudentResponse:
"""Create a student in the caller's school.
The school is taken from the authenticated user, not the request body.
"""
The docstring becomes the endpoint description; responses documents the failures the type hints cannot express.
| URL | Gives |
|---|---|
/docs | Swagger UI, interactive |
/redoc | ReDoc, reference style |
/openapi.json | The OpenAPI schema |
/openapi.json generates typed clients for TypeScript, C# and others — so a frontend can be kept in step automatically.
Disable the documentation in production unless it is protected:
app = FastAPI(docs_url=None if settings.is_production else "/docs", redoc_url=None)
A public /docs is a published map of your API.
Testing
from fastapi.testclient import TestClient
client = TestClient(app)
def test_health_returns_healthy():
response = client.get("/health")
assert response.status_code == 200
assert response.json() == {"status": "healthy"}
def test_invalid_roll_number_is_rejected():
response = client.post("/api/students", json={
"name": "Ravi Kumar",
"rollNumber": "12345",
"className": "10th",
"section": "A",
"dateOfBirth": "2009-05-14",
"parentName": "Suresh Kumar",
"parentPhone": "9951510727"
})
assert response.status_code == 422
assert any(e["loc"] == ["body", "rollNumber"] for e in response.json()["detail"])
app.dependency_overrides[get_connection] = lambda: fake_connection
app.dependency_overrides[get_current_user] = lambda: CurrentUser(school_id=1, role="Admin")
dependency_overrides substitutes any dependency in a test — a fake database, a fixed user — with no change to the endpoint. That substitutability is what dependency injection is for.
Clear the overrides afterwards, or they leak into later tests.
Errors you will hit
| Message | Cause | Fix |
|---|---|---|
422 Unprocessable Entity | Request failed Pydantic validation | Read the detail array — it names the field |
ImportError: email-validator is not installed | EmailStr needs an extra | pip install pydantic[email] |
| Route never matches | Path parameter type mismatch, or route order | More specific routes first |
AssertionError: A parameter-less dependency | Dependency not called correctly | Depends(get_db) not Depends(get_db()) |
| Docs page is empty | App not reloaded, or wrong module path | Restart uvicorn |
Read the detail array in a 422. It names the field and the rule, and most "FastAPI rejects my request" reports end there.
Common mistakes
- Entities used as request or response models
school_idaccepted from the client- No
lebound onpage_size - 200 with an empty body where 204 belongs
- 403 for another tenant's record
- Exception messages returned from a 500
async defwith a blocking database callallow_origins=["*"]- Trying to fix CORS in the frontend
.envcommitted- Configuration read with
os.getenvand no startup validation /docspublic in production- Cross-field rules split across endpoints instead of a model validator
- No
Locationheader on a 201 - Business logic in the endpoint function
Practice
The course exercise is build GET/POST endpoints, and the assignment is a simple API with validation.
- Create an app with a
/healthendpoint. Open/docsand call it. - Add
GET /api/students/{public_id}typed asUUID. Pass a non-UUID and confirm the 422. - Add search with
term,pageandpage_size, withgeandlebounds. Requestpage_size=1000000and confirm the rejection. - Remove the
lebound and repeat. Record the response time. - Write
StudentCreateRequestwith patterns and aliases. POST an invalid roll number and read thedetailarray. - Write
StudentResponseseparately withfrom_attributes. Add apassword_hashfield to your entity and confirm it is not returned. - Use the entity as the response model instead and confirm it is.
- Accept
school_idin the request body. POST with a different school's id and confirm the record is created in the wrong school. - Move it to a
get_current_userdependency and confirm the client can no longer choose. - Add a
model_validatorfor the absent/marks rule. Test both violations. - Return 201 with a
Locationheader. Follow it. - Return 200 with an empty body from a delete, call it from JavaScript with
.json(), then change it to 204. - Add a global handler for a domain exception mapping to 409.
- Raise an unexpected exception and confirm the response contains no detail. Find the full traceback in the log.
- Write an
async defendpoint calling a blocking query. Fire twenty concurrent requests and measure. Change it todefand compare. - Configure settings with
pydantic-settings. Remove a required value and confirm the startup failure. - Set
allow_origins=["*"]withallow_credentials=Trueand record the error. - Write tests with
TestClientanddependency_overridesfor the current user.
Exercises 6, 8 and 15 correspond to accidental disclosure, a tenant breach and a stalled server.
You can now
- Build a validated REST API with FastAPI
- Define request and response models with Pydantic
- Read a 422 response and fix the request
- Use dependency injection for the database session
- Return correct status codes
Review questions
- Why must request and response models be separate from entities?
- Why does FastAPI return 422 rather than 400 for validation?
- What happens when a blocking call runs inside an
async defendpoint? - Why must
school_idcome from the authenticated user?
Next: Guided Python project