Request lifecycle
When a request arrives, FastrAPI processes it through several stages:
Request object
FastrAPI provides a Request object for accessing request data:
from fastrapi import FastrAPI, Request
app = FastrAPI()
@app.get ( "/info" )
def get_info ( request : Request):
return {
"client" : request.client,
"headers" : dict (request.headers),
"path_params" : request.path_params,
"query_params" : request.query_params,
"cookies" : request.cookies,
}
Request implementation
The Request object is a Rust struct exposed to Python via PyO3:
// src/request.rs
#[pyclass(name = "Request" , module = "fastrapi.request" )]
#[derive( Clone )]
pub struct PyRequest {
#[pyo3(get)]
pub scope : Py < PyAny >,
#[pyo3(get)]
pub receive : Py < PyAny >,
#[pyo3(get)]
pub send : Py < PyAny >,
// Cache for the body if it has been read once
_body : Arc < Mutex < Option < Vec < u8 >>>>,
_is_consumed : Arc < Mutex < bool >>,
}
Async request handling
FastrAPI supports both sync and async handlers. Async handlers are executed on Tokio’s runtime:
@app.get ( "/sync" )
def sync_handler ():
# Runs in thread pool
return { "type" : "sync" }
@app.get ( "/async" )
async def async_handler ():
# Runs on Tokio runtime
return { "type" : "async" }
Handler execution
FastrAPI detects whether your function is async at decorator time:
// src/pydantic.rs
let is_async = inspect
. call_method1 ( "iscoroutinefunction" , ( func ,)) ?
. extract :: < bool >() ? ;
This information is stored in the RouteHandler and used to execute the function correctly:
// src/py_handlers.rs
pub async fn run_py_handler_with_params (
rt_handle : tokio :: runtime :: Handle ,
route_key : Arc < str >,
path_params : HashMap < String , String >,
query_params : HashMap < String , String >,
payload : Option < serde_json :: Value >,
) -> Response {
rt_handle
. spawn_blocking ( move || {
Python :: attach ( | py | {
let handler = ROUTES . get ( route_key . as_ref ()) ? ;
// Call Python function
let result = if handler . needs_kwargs {
py_func . call ((), Some ( kwargs ))
} else {
py_func . call0 ()
};
// Convert to HTTP response
convert_response_by_type ( py , & result , handler . response_type)
})
})
. await
}
Reading request body
The request body can be accessed as raw bytes or parsed as JSON:
@app.post ( "/upload" )
async def upload ( request : Request):
# Read raw body
body = await request.body()
return { "size" : len (body)}
@app.post ( "/json" )
async def json_data ( request : Request):
# Parse as JSON
data = await request.json()
return { "received" : data}
Body caching
FastrAPI caches the request body after the first read:
#[pymethods]
impl PyRequest {
fn body <' py >( & self , py : Python <' py >) -> PyResult < Bound <' py , PyAny >> {
// Check cache first
{
let cache_guard = cache . lock () . unwrap ();
if let Some ( cached ) = &* cache_guard {
return Ok ( PyBytes :: new ( py , cached ) . into_any () . unbind ());
}
}
// Read from stream if not cached
// ...
}
}
This allows multiple parts of your code to read the body without consuming the stream multiple times.
FastrAPI automatically extracts and converts parameters:
Path parameters
@app.get ( "/users/ {user_id} " )
def get_user ( user_id : int ):
# user_id is automatically converted to int
return { "user_id" : user_id}
Query parameters
@app.get ( "/search" )
def search ( q : str , limit : int = 10 ):
# q is required, limit is optional with default
return { "query" : q, "limit" : limit}
Body parameters with Pydantic
from pydantic import BaseModel
class User ( BaseModel ):
name: str
email: str
age: int
@app.post ( "/users" )
def create_user ( user : User):
# Pydantic validates the request body
return {
"name" : user.name,
"email" : user.email,
"age" : user.age
}
FastrAPI validates request data using Pydantic before calling your handler:
// src/pydantic.rs
pub fn apply_body_and_validation (
py : Python ,
handler : & RouteHandler ,
payload : & serde_json :: Value ,
kwargs : & PyDict ,
) -> Result <(), Response > {
for ( param_name , validator ) in & handler . param_validators {
let py_value = pythonize ( py , payload ) ? ;
// Call Pydantic validator
match validator . call1 ( py , ( py_value ,)) {
Ok ( validated ) => {
kwargs . set_item ( param_name , validated ) ? ;
}
Err ( e ) => {
// Return 422 validation error
return Err ( validation_error_response ( py , e ));
}
}
}
Ok (())
}
Error handling
If a handler raises an exception, FastrAPI returns a 500 error:
@app.get ( "/error" )
def error_handler ():
raise ValueError ( "Something went wrong" )
# Returns: 500 Internal Server Error
Currently, FastrAPI shows Rust error messages in development. Production-ready error handling is planned for future releases.
Fast path optimization
FastrAPI detects simple handlers that don’t need parameter processing:
# Fast path: no parameters
@app.get ( "/" )
def root ():
return { "Hello" : "World" }
# Slow path: has parameters
@app.get ( "/users/ {user_id} " )
def get_user ( user_id : int ):
return { "user_id" : user_id}
Fast-path routes skip validation, parameter extraction, and dependency resolution for maximum performance:
let is_fast_path = path_param_names . is_empty ()
&& query_param_names . is_empty ()
&& body_param_names . is_empty ()
&& dependencies . is_empty ();
Thread pool execution
Python handlers run in a dedicated thread pool to avoid blocking the Tokio runtime:
static PYTHON_RUNTIME : Lazy < tokio :: runtime :: Runtime > = Lazy :: new ( || {
tokio :: runtime :: Builder :: new_multi_thread ()
. worker_threads ( num_cpus :: get () . max ( 4 ) . min ( 16 ))
. thread_name ( "python-handler" )
. enable_all ()
. build ()
. expect ( "Failed to create Python runtime" )
});
This runtime scales from 4 to 16 threads based on available CPU cores.
Complete example
Here’s a comprehensive request handling example:
from fastrapi import FastrAPI, Request
from pydantic import BaseModel
app = FastrAPI()
class Item ( BaseModel ):
name: str
price: float
description: str | None = None
@app.get ( "/" )
def root ():
# Fast path: no parameters
return { "message" : "Welcome" }
@app.get ( "/items/ {item_id} " )
def get_item ( item_id : int , request : Request):
# Access request metadata
return {
"item_id" : item_id,
"client" : request.client,
"headers" : dict (request.headers)
}
@app.get ( "/search" )
def search ( q : str , limit : int = 10 , offset : int = 0 ):
# Query parameters
return {
"query" : q,
"limit" : limit,
"offset" : offset
}
@app.post ( "/items" )
def create_item ( item : Item):
# Pydantic validation
return {
"name" : item.name,
"price" : item.price,
"description" : item.description
}
@app.post ( "/upload" )
async def upload_file ( request : Request):
# Read raw body
body = await request.body()
return { "size" : len (body)}
@app.post ( "/webhook" )
async def webhook ( request : Request):
# Parse JSON
data = await request.json()
return { "received" : data}
if __name__ == "__main__" :
app.serve( "127.0.0.1" , 8080 )
Use fast-path routes when possible
Routes without parameters are significantly faster. If you have high-traffic endpoints that don’t need parameters, keep them simple.
Prefer async handlers for I/O
Use async def for handlers that make database queries, HTTP requests, or other I/O operations.
Cache request body if needed multiple times
The request body is automatically cached after the first read, so you can safely call await request.body() multiple times.
Use Pydantic for validation
Pydantic validation happens in Rust-land for better performance than manual validation in Python.
Next steps
Response types Learn about different response formats
Dependency injection Understand how to use dependencies