# Create the virtual environment python -m venv venv # Activate on Windows (CMD) venv\Scripts\activate # Activate on Windows (PowerShell) venv\Scripts\Activate.ps1 # Activate on macOS / Linux source venv/bin/activate
# Install FastAPI with uvicorn pip install fastapi uvicorn # Verify the installation pip list | grep -i fastapi # Run the server with auto-reload uvicorn main:app --reload
(venv) prefix in your terminal confirms it is active. To deactivate it, just run deactivate.
from fastapi import FastAPI app = FastAPI()
fastapi dev main.py for development with auto-reload, and fastapi run main.py for production. Both use uvicorn under the hood.
A FastAPI app is an app = FastAPI() object decorated with one path operation per endpoint: @app.get("/items"), @app.post(...), @app.put(...), @app.patch(...), @app.delete(...), each binding an HTTP method and path to the function right below it. Group related endpoints in an APIRouter (with a shared prefix and tags) and pull them in with app.include_router(router).
@app.get("/") def read_root(): return {"msg": "Hello FastAPI!"} @app.get("/items/{item_id}") def read_item(item_id: int, q: str | None = None): return {"item_id": item_id, "q": q} @app.post("/items/") def create_item(item: dict): return {"item": item} @app.put("/items/{item_id}") def update_item(item_id: int, item: dict): return {"item_id": item_id, "updated": item} @app.patch("/items/{item_id}") def partial_update_item(item_id: int, item: dict): return {"item_id": item_id, "patched": item} @app.delete("/items/{item_id}") def delete_item(item_id: int): return {"item_id": item_id, "deleted": True}
@app.get("/users/{user_id}") def get_user(user_id: int, active: bool = True): return {"user_id": user_id, "active": active}
from fastapi import FastAPI, APIRouter app = FastAPI(title="My API", version="1.0") router = APIRouter(prefix="/v1", tags=["items"]) # group related routes app.include_router(router) # mount them on the app
Declare a pydantic.BaseModel and use it as a parameter type to accept and validate a JSON request body. FastAPI parses the bytes, checks every field (Field(gt=0), required vs defaulted), and hands your function a typed object — or returns a 422 listing exactly which field failed. Set response_model=ItemOut on the decorator to shape and filter what goes back out.
from pydantic import BaseModel class Item(BaseModel): name: str price: float in_stock: bool = True @app.post("/items/") def create_item(item: Item): return item
from pydantic import BaseModel, Field class ItemIn(BaseModel): name: str price: float = Field(gt=0) # validated: price must be > 0 tags: list[str] = [] class ItemOut(BaseModel): id: int name: str price: float @app.post("/items", response_model=ItemOut, status_code=201) def create_item(item: ItemIn): # "secret" is stripped by response_model return {"id": 1, "name": item.name, "price": item.price, "secret": "hidden"}
Wrap hints in Annotated[int, Path(ge=1)] or Annotated[str | None, Query(max_length=50)] to add constraints. Any value that fails is rejected with a 422 before your code runs.
from fastapi import Query, Path from typing import Annotated @app.get("/search/") def search( q: Annotated[str, Query(min_length=3, max_length=50)], page: Annotated[int, Query(ge=1)] = 1, ): return {"q": q, "page": page} @app.get("/users/{user_id}") def get_user(user_id: Annotated[int, Path(ge=1)]): return {"user_id": user_id} # GET /search/?q=ab -> 422 (min_length=3) # GET /users/0 -> 422 (fails ge=1) # GET /users/3?page=5 -> 200
Dependency injection lets a route declare what it needs as Annotated[T, Depends(provider)]. FastAPI calls the provider and injects the result. A provider that uses yield gives setup-and-teardown. Listing dependencies=[Depends(guard)] on the decorator runs a check (like auth) without passing a value.
from fastapi import FastAPI, Depends, HTTPException, Query from typing import Annotated app = FastAPI() def common_params(q: str = "", skip: int = 0, limit: int = 10): return {"q": q, "skip": skip, "limit": limit} @app.get("/items/") def read_items(commons: Annotated[dict, Depends(common_params)]): return commons # Setup & teardown with yield def get_db(): db = connect() try: yield db # injected into the handler finally: db.close() # teardown, always runs # Route-level guard (no value passed) def verify_token(x_token: Annotated[str | None, Query()] = None): if x_token != "secret": raise HTTPException(status_code=401, detail="bad token") @app.get("/admin", dependencies=[Depends(verify_token)]) def admin(): return {"ok": True}
from fastapi import Depends, HTTPException from fastapi.security import OAuth2PasswordBearer oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") @app.get("/users/me") def read_users_me(token: str = Depends(oauth2_scheme)): return {"token": token}
from fastapi.security.api_key import APIKeyHeader api_key_header = APIKeyHeader(name="X-API-Key") @app.get("/secure-data/") def secure_data(api_key: str = Depends(api_key_header)): if api_key != "secret123": raise HTTPException(403, "Unauthorized") return {"data": "Top secret!"}
Set the happy-path code with status_code= on the decorator. Signal a failure by raising HTTPException(status_code=..., detail=...). Validation failures become a structured 422 automatically, and @app.exception_handler(...) lets you convert your own exception types.
from fastapi import FastAPI, HTTPException, status from fastapi.responses import JSONResponse app = FastAPI() @app.post("/items", status_code=status.HTTP_201_CREATED) def make_item(): return {"ok": True} @app.get("/items/{item_id}") def read_item(item_id: int): if item_id == 0: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="not found", headers={"WWW-Authenticate": "Bearer"}, ) return {"item_id": item_id} @app.exception_handler(ValueError) def handle(request, exc): return JSONResponse(status_code=400, content={"detail": str(exc)})
from fastapi import Cookie, Header @app.get("/cookies/") def get_cookie(session_id: str = Cookie(None)): return {"session_id": session_id} @app.get("/headers/") def get_header(user_agent: str = Header(None)): return {"user_agent": user_agent}
Write async def for awaitable I/O; write plain def for blocking work and FastAPI offloads it to a thread pool. Queue work to run after the response with BackgroundTasks, and manage startup/shutdown with a lifespan context manager (the old @app.on_event decorators are deprecated).
from fastapi import BackgroundTasks def write_log(msg: str): with open("log.txt", "a") as f: f.write(msg + "\n") @app.post("/log/") def log_message(msg: str, bg: BackgroundTasks): bg.add_task(write_log, msg) return {"status": "scheduled"}
from sqlalchemy import create_engine, Column, Integer, String from sqlalchemy.orm import declarative_base, sessionmaker # PostgreSQL database connection DATABASE_URL = ( "postgresql+psycopg://" "almacen_user:tu_contraseña@localhost:5432/almacen" ) # Create database engine engine = create_engine( DATABASE_URL, echo=True ) # Create session factory SessionLocal = sessionmaker( bind=engine, autoflush=False, autocommit=False ) # Create declarative base Base = declarative_base() # Define User model class User(Base): __tablename__ = "users" id = Column( Integer, primary_key=True, index=True ) name = Column( String, index=True, nullable=False ) # Create database tables Base.metadata.create_all(bind=engine)
pip install psycopg[binary] para el driver PostgreSQL. El parámetro echo=True imprime las queries SQL en consola, útil para depuración.
Your types generate an OpenAPI 3.1 schema at /openapi.json with no extra work. FastAPI serves Swagger UI at /docs and ReDoc at /redoc. Enrich docs with tags, summary, and per-field Field(description=..., examples=[...]).
from fastapi.middleware.cors import CORSMiddleware app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], )
from fastapi import FastAPI from pydantic import BaseModel, Field app = FastAPI(docs_url="/api-docs", redoc_url=None) # relocate Swagger, drop ReDoc class Item(BaseModel): name: str = Field(examples=["Ada"], description="The item name") @app.get("/items", tags=["items"], summary="List items") def list_items(): ...
from fastapi.testclient import TestClient client = TestClient(app) def test_root(): response = client.get("/") assert response.status_code == 200 assert response.json() == {"msg": "Hello FastAPI!"}
TestClient requires pip install httpx.
# Development: auto-reload fastapi dev main.py # Production: no reload, binds 0.0.0.0:8000 fastapi run main.py # Direct uvicorn invocation uvicorn main:app --reload --port 8000 # Scale with worker processes uvicorn main:app --workers 4
import uvicorn if __name__ == "__main__": uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True)
Key FastAPI calls
| Command | What it does | Area |
|---|---|---|
app = FastAPI() | Create the application object | Routes |
@app.get("/path") | Bind GET (also post/put/patch/delete) | Routes |
APIRouter(prefix=..., tags=...) | Group routes, then include_router | Routes |
item_id: int (in path) | Typed, validated path parameter | Params |
Annotated[int, Path(ge=1)] | Constrain a path parameter | Validation |
item: ItemIn (a BaseModel) | Parse and validate the JSON body | Models |
response_model=ItemOut | Shape and filter the response | Models |
Annotated[T, Depends(provider)] | Inject a dependency | Depends |
OAuth2PasswordBearer(tokenUrl=...) | JWT token auth scheme | Security |
APIKeyHeader(name="X-API-Key") | API key auth from header | Security |
raise HTTPException(404, ...) | Return a client or server error | Errors |
UploadFile = File(...) | Accept a file upload | Upload |
Cookie(None) / Header(None) | Read a cookie / header value | Cookies |
BackgroundTasks.add_task(fn) | Run work after responding | Async |
FastAPI(lifespan=lifespan) | App startup and shutdown | Async |
@app.middleware("http") | Run code on every request | Middleware |
CORSMiddleware | Enable cross-origin requests | CORS |
TestClient(app) | In-process test client | Testing |
/docs, /redoc, /openapi.json | Auto docs and schema | Docs |
fastapi dev main.py | Serve with auto-reload | Run |
uvicorn main:app --workers 4 | Scale with worker processes | Run |
Where each parameter is read from
| Declared as | Comes from | Example |
|---|---|---|
In the path {...} | URL path | @app.get("/items/{id}") |
| Plain function arg | Query string | q: str = None → ?q=... |
A BaseModel arg | JSON request body | item: ItemIn |
Annotated[..., Body()] | A single body field | q: Annotated[str, Body()] |
Annotated[..., Depends(...)] | A dependency provider | db: Annotated[Session, Depends(get_db)] |
Header(None) | A request header | user_agent: str = Header(None) |
Cookie(None) | A cookie value | session_id: str = Cookie(None) |
File(...) | Multipart upload | file: UploadFile = File(...) |
Form(...) | Form data | username: str = Form(...) |
HTTP status code classes
| Class | Range | Meaning | Color |
|---|---|---|---|
| 2xx | 200–299 | Success | green |
| 3xx | 300–399 | Redirect | amber |
| 4xx | 400–499 | Client error (their request) | red |
| 5xx | 500–599 | Server error (your handler) | red |
| 422 | (a 4xx) | Validation failed (FastAPI built it) | red |
async def for I/O-bound tasks (database queries, HTTP calls). Plain def is offloaded to a threadpool automatically.response_model to filter and validate outputs — keeps responses lean and avoids leaking internal fields.BackgroundTasks for simple jobs, or Celery / ARQ for distributed queues.--workers N to uvicorn in production to scale across CPU cores.response_class=ORJSONResponse with pip install orjson for faster JSON serialization on large payloads.{item_id} declared item_id: int coerces /items/42 to the int 42; /items/abc is rejected with a 422 before your handler runs.Annotated[...] for params. The legacy default-value spelling still works but is no longer the recommended style.response_model shapes the way out. Extra fields are stripped, so a secret never leaks to the client.lifespan, not on_event. The @app.on_event("startup") and "shutdown" decorators are deprecated.model_dump(), model_validate(), and ConfigDict./openapi.json, Swagger UI at /docs, ReDoc at /redoc.