FastAPI Request Processing

Introduction

When you build web applications with FastAPI, understanding how requests are processed is essential. Request processing is the mechanism by which FastAPI receives, parses, validates, and makes available the data sent from clients to your application.

In this tutorial, we'll explore how FastAPI handles incoming HTTP requests, how to access different types of request data, and how to customize request processing to suit your application's needs.

How FastAPI Processes Requests

When a client sends an HTTP request to your FastAPI application, a sequence of operations occurs:

1. Route matching: FastAPI matches the request path to a defined path operation
2. Dependency resolution: Any dependencies required by the path operation are resolved
3. Parameter extraction: Data is extracted from the request (path, query, headers, body)
4. Data validation: The extracted data is validated against the defined models
5. Function execution: Your path operation function is called with the validated data
6. Response generation: The return value is converted to an HTTP response

Let's dive into each aspect of request processing.

Accessing Request Data

FastAPI provides several ways to access data from incoming requests. Let's explore each method:

Path Parameters

Path parameters are parts of the URL path that are variable and are captured as function parameters.

python

from fastapi import FastAPI

app = FastAPI()

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

In this example, if a request comes to /items/42, FastAPI will:

• Extract 42 from the URL path
• Convert it to an integer (as specified by the type hint)
• Pass it to the function as item_id
• Return {"item_id": 42}

Query Parameters

Query parameters are the key-value pairs that appear after the ? in a URL.

python

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

For a request to /items/?skip=20&limit=30, FastAPI will:

• Extract the query parameters skip=20 and limit=30
• Convert them to integers
• Call the function with skip=20 and limit=30
• Return {"skip": 20, "limit": 30}

If the URL doesn't include these parameters, the default values (0 and 10) are used.

Request Body

To receive JSON request bodies, you can define a Pydantic model and use it as a parameter:

python

from fastapi import FastAPI
from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

app = FastAPI()

@app.post("/items/")
async def create_item(item: Item):
    item_dict = item.dict()
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict

When a POST request is made to /items/ with a JSON body, FastAPI will:

• Read the request body as JSON
• Validate the data against the Item model
• Create an instance of Item with the data
• Pass that instance to your function

Example request body:

json

{
    "name": "Laptop",
    "price": 999.99,
    "tax": 100.00
}

Example response:

json

{
    "name": "Laptop",
    "description": null,
    "price": 999.99,
    "tax": 100.0,
    "price_with_tax": 1099.99
}

Form Data

To handle form data, you'll need to install python-multipart and use Form:

python

from fastapi import FastAPI, Form

app = FastAPI()

@app.post("/login/")
async def login(username: str = Form(...), password: str = Form(...)):
    return {"username": username}

File Uploads

FastAPI makes it easy to handle file uploads:

python

from fastapi import FastAPI, File, UploadFile

app = FastAPI()

@app.post("/files/")
async def create_file(file: bytes = File(...)):
    return {"file_size": len(file)}

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile):
    contents = await file.read()
    return {"filename": file.filename, "size": len(contents)}

Headers

You can access request headers in a similar way:

python

from fastapi import FastAPI, Header

app = FastAPI()

@app.get("/items/")
async def read_items(user_agent: str = Header(None)):
    return {"User-Agent": user_agent}

The Request Object

Sometimes, you may need to access the raw request object. FastAPI provides access to this through the Request class:

python

from fastapi import FastAPI, Request

app = FastAPI()

@app.get("/info")
async def get_request_info(request: Request):
    client_host = request.client.host
    method = request.method
    url = request.url
    headers = request.headers
    query_params = request.query_params
    
    return {
        "client_host": client_host,
        "method": method,
        "url": str(url),
        "headers": dict(headers),
        "query_params": dict(query_params)
    }

This is particularly useful when you need information about the request that isn't easily accessible through FastAPI's parameter injection.

Request Validation

One of FastAPI's most powerful features is automatic request validation:

python

from fastapi import FastAPI, Query
from typing import Optional

app = FastAPI()

@app.get("/items/")
async def read_items(
    q: Optional[str] = Query(
        None, 
        min_length=3, 
        max_length=50, 
        regex="^[a-zA-Z]+$"
    )
):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

This validates that the query parameter q:

• Is optional (None by default)
• Must be at least 3 characters long
• Cannot be more than 50 characters long
• Must match the regular expression (only alphabet characters)

If validation fails, FastAPI automatically returns an HTTP 422 Unprocessable Entity response with details about the validation error.

Real-world Example: Building a Simple API with Request Processing

Let's put all of this together in a more comprehensive example - a simple API for managing blog posts:

python

from fastapi import FastAPI, Path, Query, Body, HTTPException
from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime

app = FastAPI()

# Models
class PostBase(BaseModel):
    title: str = Field(..., min_length=1, max_length=100)
    content: str = Field(..., min_length=10)
    published: bool = True

class PostCreate(PostBase):
    pass

class Post(PostBase):
    id: int
    created_at: datetime
    
    class Config:
        orm_mode = True

# Fake database
posts_db = [
    Post(
        id=1,
        title="First Post",
        content="This is my first post content",
        published=True,
        created_at=datetime.now()
    ),
    Post(
        id=2,
        title="Second Post",
        content="This is my second post content",
        published=False,
        created_at=datetime.now()
    )
]

# Path operations
@app.get("/posts/", response_model=List[Post])
async def get_posts(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100),
    published: Optional[bool] = None
):
    if published is None:
        return posts_db[skip:skip+limit]
    
    filtered_posts = [post for post in posts_db if post.published == published]
    return filtered_posts[skip:skip+limit]

@app.get("/posts/{post_id}", response_model=Post)
async def get_post(post_id: int = Path(..., gt=0)):
    for post in posts_db:
        if post.id == post_id:
            return post
    raise HTTPException(status_code=404, detail="Post not found")

@app.post("/posts/", response_model=Post, status_code=201)
async def create_post(post: PostCreate = Body(...)):
    new_post = Post(
        id=len(posts_db) + 1,
        **post.dict(),
        created_at=datetime.now()
    )
    posts_db.append(new_post)
    return new_post

@app.put("/posts/{post_id}", response_model=Post)
async def update_post(
    post_id: int = Path(..., gt=0),
    post_update: PostBase = Body(...)
):
    for i, post in enumerate(posts_db):
        if post.id == post_id:
            updated_post = Post(
                id=post_id,
                **post_update.dict(),
                created_at=post.created_at
            )
            posts_db[i] = updated_post
            return updated_post
    raise HTTPException(status_code=404, detail="Post not found")

@app.delete("/posts/{post_id}", status_code=204)
async def delete_post(post_id: int = Path(..., gt=0)):
    for i, post in enumerate(posts_db):
        if post.id == post_id:
            posts_db.pop(i)
            return
    raise HTTPException(status_code=404, detail="Post not found")

This example demonstrates:

• Path parameter validation with Path
• Query parameter validation with Query
• Request body validation with Pydantic models and Body
• Different HTTP methods (GET, POST, PUT, DELETE)
• Status code customization
• Error handling with HTTPException

Summary

FastAPI's request processing is a powerful system that combines route matching, dependency injection, and data validation to create robust API endpoints. The key points to remember are:

1. Type hints drive the validation and conversion of request data
2. Pydantic models provide a way to define complex data structures for requests
3. Path, Query, Body, and other classes allow fine-tuning the validation rules
4. FastAPI automates the tedious parts of request processing like validation
5. The Request object gives you access to lower-level details when needed

Understanding request processing is fundamental to building effective FastAPI applications, as it allows you to confidently accept and process user input while ensuring data correctness.

Additional Resources

• FastAPI official documentation on request body
• FastAPI request validation
• Pydantic documentation

Exercises

1. Create a FastAPI endpoint that accepts a complex JSON object with nested fields and validates it using Pydantic models.
2. Build an endpoint that accepts both query parameters and a request body, with appropriate validations for each.
3. Create an API that accepts file uploads and performs validation on the file type and size.
4. Implement a search endpoint with multiple optional filters as query parameters.
5. Build a complete CRUD API for a resource of your choice with proper validation at all endpoints.