Skip to main content
Published / updated

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_id or status lets a client set them — moving a student to another school.
  • Accidental disclosure. Returning the entity means adding a password_hash column 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:
...
CodeWhen
200GET, or an update returning the result
201POST succeeded — include Location
204Successful PUT or DELETE with no body
400A domain rule was broken
401Not authenticated
403Authenticated, not permitted
404Not found
409Conflict — duplicate
422Validation 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.

URLGives
/docsSwagger UI, interactive
/redocReDoc, reference style
/openapi.jsonThe 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

MessageCauseFix
422 Unprocessable EntityRequest failed Pydantic validationRead the detail array — it names the field
ImportError: email-validator is not installedEmailStr needs an extrapip install pydantic[email]
Route never matchesPath parameter type mismatch, or route orderMore specific routes first
AssertionError: A parameter-less dependencyDependency not called correctlyDepends(get_db) not Depends(get_db())
Docs page is emptyApp not reloaded, or wrong module pathRestart 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_id accepted from the client
  • No le bound on page_size
  • 200 with an empty body where 204 belongs
  • 403 for another tenant's record
  • Exception messages returned from a 500
  • async def with a blocking database call
  • allow_origins=["*"]
  • Trying to fix CORS in the frontend
  • .env committed
  • Configuration read with os.getenv and no startup validation
  • /docs public in production
  • Cross-field rules split across endpoints instead of a model validator
  • No Location header 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.

  1. Create an app with a /health endpoint. Open /docs and call it.
  2. Add GET /api/students/{public_id} typed as UUID. Pass a non-UUID and confirm the 422.
  3. Add search with term, page and page_size, with ge and le bounds. Request page_size=1000000 and confirm the rejection.
  4. Remove the le bound and repeat. Record the response time.
  5. Write StudentCreateRequest with patterns and aliases. POST an invalid roll number and read the detail array.
  6. Write StudentResponse separately with from_attributes. Add a password_hash field to your entity and confirm it is not returned.
  7. Use the entity as the response model instead and confirm it is.
  8. Accept school_id in the request body. POST with a different school's id and confirm the record is created in the wrong school.
  9. Move it to a get_current_user dependency and confirm the client can no longer choose.
  10. Add a model_validator for the absent/marks rule. Test both violations.
  11. Return 201 with a Location header. Follow it.
  12. Return 200 with an empty body from a delete, call it from JavaScript with .json(), then change it to 204.
  13. Add a global handler for a domain exception mapping to 409.
  14. Raise an unexpected exception and confirm the response contains no detail. Find the full traceback in the log.
  15. Write an async def endpoint calling a blocking query. Fire twenty concurrent requests and measure. Change it to def and compare.
  16. Configure settings with pydantic-settings. Remove a required value and confirm the startup failure.
  17. Set allow_origins=["*"] with allow_credentials=True and record the error.
  18. Write tests with TestClient and dependency_overrides for 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

  1. Why must request and response models be separate from entities?
  2. Why does FastAPI return 422 rather than 400 for validation?
  3. What happens when a blocking call runs inside an async def endpoint?
  4. Why must school_id come from the authenticated user?

Next: Guided Python project