Skip to main content

Routing and Path Operations

Organize your API by grouping related path operations into routers and defining endpoints with specific HTTP methods, streaming capabilities, and WebSocket support.

Define Path Operations​

Define endpoints directly on the FastAPI application instance using decorators for HTTP methods like @app.get(), @app.post(), @app.put(), and @app.delete().

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

@app.post("/items/")
async def create_item(item: Item):
    return item

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

Group Routes with APIRouter​

Use APIRouter to organize your application into multiple files or logical groups. You can apply shared prefixes, tags, and dependencies to all routes within a router.

# routers/users.py
from fastapi import APIRouter, Depends

router = APIRouter(
    prefix="/users",
    tags=["users"],
    responses={404: {"description": "Not found"}},
)

@router.get("/")
async def read_users():
    return [{"username": "Rick"}, {"username": "Morty"}]

Include Routers in the Main App​

Connect your routers to the main FastAPI application using app.include_router().

# main.py
from fastapi import FastAPI
from .routers import users, items

app = FastAPI()

app.include_router(users.router)
app.include_router(
    items.router,
    prefix="/items",
    tags=["items"],
)

Stream Responses​

This codebase supports automatic streaming for JSON Lines (JSONL) and Server-Sent Events (SSE) based on return type annotations and response classes.

Stream JSON Lines (JSONL)​

To stream data as application/jsonl, return an AsyncIterable or Iterable from your path operation. FastAPI will automatically serialize each yielded item and append a newline.

from typing import AsyncIterable
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str

@app.get("/items/stream")
async def stream_items() -> AsyncIterable[Item]:
    items = [Item(name="Plumbus"), Item(name="Portal Gun")]
    for item in items:
        yield item

Stream Server-Sent Events (SSE)​

Use EventSourceResponse as the response_class to stream events. You can yield plain objects (which are JSON-encoded) or ServerSentEvent objects for fine-grained control over event fields like event, id, and retry.

from fastapi import FastAPI
from fastapi.responses import EventSourceResponse
from fastapi.sse import ServerSentEvent

app = FastAPI()

@app.get("/events", response_class=EventSourceResponse)
async def stream_events():
    yield ServerSentEvent(data="hello", event="greeting", id="1")
    yield {"msg": "plain data is also supported"}

Manage WebSocket Connections​

Define WebSocket endpoints using the @app.websocket() decorator. These endpoints integrate with the dependency injection system.

from fastapi import FastAPI, WebSocket, Depends

app = FastAPI()

async def get_cookie_or_token(websocket: WebSocket):
    return websocket.cookies.get("session")

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket, session: str = Depends(get_cookie_or_token)):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Session {session} sent: {data}")

Troubleshooting​

Strict Content-Type Checking​

By default, strict_content_type is enabled. If a client sends a JSON body without a Content-Type: application/json header, the request will fail. You can disable this at the app or router level:

# Disable globally
app = FastAPI(strict_content_type=False)

# Or disable for a specific router
router = APIRouter(strict_content_type=False)

Path Prefix Constraints​

When using APIRouter prefixes:

  • Prefixes must start with / (e.g., /items).
  • Prefixes must not end with /.
  • If you include a router, the combination of the router prefix and the route path cannot result in an empty path.

Response Validation Errors​

If your path operation returns data that does not match the response_model (or the type in AsyncIterable[Type]), FastAPI will raise a ResponseValidationError (HTTP 500). Ensure your return values strictly adhere to your Pydantic models.