Leapcell: The Best of Serverless Web Hosting
Building a Simplified FastAPI from Scratch: Understanding ASGI and Core Routing
Introduction: Why Reinvent This Wheel?
When we talk about Python asynchronous web frameworks, FastAPI is undoubtedly the brightest star in recent years. It has gained widespread acclaim for its impressive performance, automatic API documentation generation, and type hint support. But have you ever wondered: what magic lies behind this powerful framework?
Today, we'll build a simplified version of FastAPI from scratch, focusing on understanding two core concepts: the ASGI protocol and the routing system. By constructing it with our own hands, you'll grasp the working principles of modern asynchronous web frameworks. This won't just help you use FastAPI better—it'll enable you to quickly identify the root cause when problems arise.
What is ASGI? Why is it More Advanced than WSGI?
Before we start coding, we need to understand ASGI (Asynchronous Server Gateway Interface)—the foundation that allows FastAPI to achieve high-performance asynchronous processing.
Limitations of WSGI
If you've used Django or Flask, you've probably heard of WSGI (Web Server Gateway Interface). WSGI is a synchronous interface specification between Python web applications and servers, but it has obvious flaws:
- Can only handle one request at a time, no concurrency
- Doesn't support long-lived connections (like WebSocket)
- Can't fully leverage the advantages of asynchronous I/O
Advantages of ASGI
ASGI was created to solve these problems:
- Fully asynchronous, supporting concurrent processing of multiple requests
- Compatible with WebSocket and HTTP/2
- Allows middleware to work in asynchronous environments
- Supports asynchronous events throughout the request lifecycle
Simply put, ASGI defines a standard interface that allows asynchronous web applications to communicate with servers (like Uvicorn). Next, we'll implement a minimalist ASGI server.
Step 1: Implement a Basic ASGI Server
An ASGI application is essentially a callable object (function or class) that receives three parameters: scope, receive, and send.
# asgi_server.py
import socket
import asyncio
import json
from typing import Callable, Awaitable, Dict, Any
# ASGI application type definition
ASGIApp = Callable[[Dict[str, Any], Callable[[], Awaitable[Dict]]], Awaitable[None]]
class ASGIServer:
def __init__(self, host: str = "127.0.0.1", port: int = 8000):
self.host = host
self.port = port
self.app: ASGIApp = self.default_app # Default application
async def default_app(self, scope: Dict[str, Any], receive: Callable, send: Callable):
"""Default application: returns 404 response"""
if scope["type"] == "http":
await send({
"type": "http.response.start",
"status": 404,
"headers": [(b"content-type", b"text/plain")]
})
await send({
"type": "http.response.body",
"body": b"Not Found"
})
async def handle_connection(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
"""Handles new connections, parses HTTP requests and passes to ASGI application"""
data = await reader.read(1024)
request = data.decode().split("\r\n")
method, path, _ = request[0].split()
# Build ASGI scope
scope = {
"type": "http",
"method": method,
"path": path,
"headers": []
}
# Parse request headers
for line in request[1:]:
if line == "":
break
key, value = line.split(":", 1)
scope["headers"].append((key.strip().lower().encode(), value.strip().encode()))
# Define receive and send methods
async def receive() -> Dict:
"""Simulates receiving messages (simplified version)"""
return {"type": "http.request", "body": b""}
async def send(message: Dict):
"""Sends response to client"""
if message["type"] == "http.response.start":
status = message["status"]
status_line = f"HTTP/1.1 {status} OK\r\n"
headers = "".join([f"{k.decode()}: {v.decode()}\r\n" for k, v in message["headers"]])
writer.write(f"{status_line}{headers}\r\n".encode())
if message["type"] == "http.response.body":
writer.write(message["body"])
await writer.drain()
writer.close()
# Call ASGI application
await self.app(scope, receive, send)
async def run(self):
"""Starts the server"""
server = await asyncio.start_server(
self.handle_connection, self.host, self.port
)
print(f"Server running on http://{self.host}:{self.port}")
async with server:
await server.serve_forever()
# Run the server
if __name__ == "__main__":
server = ASGIServer()
asyncio.run(server.run())
This simplified ASGI server can handle basic HTTP requests and return responses. Test it out: after running the script, visit http://127.0.0.1:8000 and you'll see "Not Found" because we haven't defined any routes yet.
Step 2: Implement the Routing System
One of FastAPI's most intuitive features is its elegant route definition, like:
@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
return {"item_id": item_id, "q": q}
Let's implement similar routing functionality.
Routing Core Component Design
We need three core components:
- Router: Manages all routing rules
- Decorators: @get, @post, etc., for registering routes
- Path matching: Handles dynamic path parameters (like /items/{item_id})
# router.py
from typing import Callable, Awaitable, Dict, Any, List, Tuple, Pattern
import re
from functools import wraps
# Route type definition
RouteHandler = Callable[[Dict[str, Any]], Awaitable[Dict[str, Any]]]
class Route:
def __init__(self, path: str, methods: List[str], handler: RouteHandler):
self.path = path
self.methods = [m.upper() for m in methods]
self.handler = handler
self.path_pattern, self.param_names = self.compile_path(path)
def compile_path(self, path: str) -> Tuple[Pattern, List[str]]:
"""Converts path to regular expression and extracts parameter names"""
param_names = []
pattern = re.sub(r"{(\w+)}", lambda m: (param_names.append(m.group(1)), r"(\w+)")[1], path)
return re.compile(f"^{pattern}$"), param_names
def match(self, path: str, method: str) -> Tuple[bool, Dict[str, Any]]:
"""Matches path and method, returns parameters"""
if method not in self.methods:
return False, {}
match = self.path_pattern.match(path)
if not match:
return False, {}
params = dict(zip(self.param_names, match.groups()))
return True, params
class Router:
def __init__(self):
self.routes: List[Route] = []
def add_route(self, path: str, methods: List[str], handler: RouteHandler):
"""Adds a route"""
self.routes.append(Route(path, methods, handler))
def route(self, path: str, methods: List[str]):
"""Route decorator"""
def decorator(handler: RouteHandler):
self.add_route(path, methods, handler)
@wraps(handler)
async def wrapper(*args, **kwargs):
return await handler(*args, **kwargs)
return wrapper
return decorator
# Shortcut methods
def get(self, path: str):
return self.route(path, ["GET"])
def post(self, path: str):
return self.route(path, ["POST"])
async def handle(self, scope: Dict[str, Any], receive: Callable) -> Dict[str, Any]:
"""Handles requests, finds matching route and executes it"""
path = scope["path"]
method = scope["method"]
for route in self.routes:
matched, params = route.match(path, method)
if matched:
# Parse query parameters
query_params = self.parse_query_params(scope)
# Merge path parameters and query parameters
request_data = {** params, **query_params}
# Call handler function
return await route.handler(request_data)
# No route found
return {"status": 404, "body": {"detail": "Not Found"}}
def parse_query_params(self, scope: Dict[str, Any]) -> Dict[str, Any]:
"""Parses query parameters (simplified version)"""
# In actual ASGI, query parameters are in scope["query_string"]
query_string = scope.get("query_string", b"").decode()
params = {}
if query_string:
for pair in query_string.split("&"):
if "=" in pair:
key, value = pair.split("=", 1)
params[key] = value
return params
Integrating Routing with the ASGI Server
Now we need to modify our ASGI server to use our routing system:
# Add routing support to ASGIServer class
class ASGIServer:
def __init__(self, host: str = "127.0.0.1", port: int = 8000):
self.host = host
self.port = port
self.router = Router() # Instantiate router
self.app = self.asgi_app # Use routing-enabled ASGI application
async def asgi_app(self, scope: Dict[str, Any], receive: Callable, send: Callable):
"""ASGI application with routing functionality"""
if scope["type"] == "http":
# Handle request
response = await self.router.handle(scope, receive)
status = response.get("status", 200)
body = json.dumps(response.get("body", {})).encode()
# Send response
await send({
"type": "http.response.start",
"status": status,
"headers": [(b"content-type", b"application/json")]
})
await send({
"type": "http.response.body",
"body": body
})
Step 3: Implement Parameter Parsing and Type Conversion
One of FastAPI's highlights is its automatic parameter parsing and type conversion. Let's implement this feature:
# Add type conversion to Router's handle method
async def handle(self, scope: Dict[str, Any], receive: Callable) -> Dict[str, Any]:
# ... previous code ...
if matched:
# Parse query parameters
query_params = self.parse_query_params(scope)
# Merge path parameters and query parameters
raw_data = {** params, **query_params}
# Get parameter type annotations from handler function
handler_params = route.handler.__annotations__
# Type conversion
request_data = {}
for key, value in raw_data.items():
if key in handler_params:
target_type = handler_params[key]
try:
# Attempt type conversion
request_data[key] = target_type(value)
except (ValueError, TypeError):
return {
"status": 400,
"body": {"detail": f"Invalid type for {key}, expected {target_type}"}
}
else:
request_data[key] = value
# Call handler function
return await route.handler(request_data)
Now our framework can automatically convert parameters to the types specified by the function annotations!
Step 4: Implement Request Body Parsing (POST Support)
Next, we'll add support for POST request bodies, enabling JSON data parsing:
# Add request body parsing to Router
async def handle(self, scope: Dict[str, Any], receive: Callable) -> Dict[str, Any]:
# ... previous code ...
# If it's a POST request, parse the request body
request_body = {}
if method == "POST":
# Get request body from receive
message = await receive()
if message["type"] == "http.request" and "body" in message:
try:
request_body = json.loads(message["body"].decode())
except json.JSONDecodeError:
return {
"status": 400,
"body": {"detail": "Invalid JSON"}
}
# Merge all parameters
raw_data = {** params, **query_params,** request_body}
# ... type conversion and handler function call ...
Step 5: Build a Complete Example Application
Now we can use our framework just like FastAPI:
# main.py
from asgi_server import ASGIServer
import asyncio
# Create server instance (includes router)
app = ASGIServer()
router = app.router
# Define routes
@router.get("/")
async def root():
return {"message": "Hello, World!"}
@router.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
return {"item_id": item_id, "q": q}
@router.post("/items/")
async def create_item(name: str, price: float):
return {"item": {"name": name, "price": price, "id": 42}}
# Run the application
if __name__ == "__main__":
asyncio.run(app.run())
Test this application:
- Visit http://127.0.0.1:8000 → Get welcome message
- Visit http://127.0.0.1:8000/items/42?q=test → Get response with parameters
- Send a POST request to http://127.0.0.1:8000/items/ with {"name": "Apple", "price": 1.99} → Get the created item
Differences from FastAPI and Optimization Directions
Our simplified version implements FastAPI's core functionality, but the real FastAPI has many advanced features:
- Dependency injection system: FastAPI's dependency injection is very powerful, supporting nested dependencies, global dependencies, etc.
- Automatic documentation: FastAPI can automatically generate Swagger and ReDoc documentation
- More data type support: Including Pydantic model validation, form data, file uploads, etc.
- Middleware system: More complete middleware support
- WebSocket support: Full implementation of ASGI's WebSocket specification
- Asynchronous database tools: Deep integration with tools like SQLAlchemy
Summary: What Have We Learned?
Through this hands-on practice, we've understood:
- The basic working principles of the ASGI protocol: the three elements of scope, receive, and send
- The core of the routing system: path matching, parameter parsing, and handler function mapping
- How type conversion is implemented: using function annotations for automatic conversion
- The request handling process: the complete lifecycle from receiving a request to returning a response
This knowledge applies not only to FastAPI but also to all ASGI frameworks (like Starlette, Quart, etc.). When you encounter problems using these frameworks, recalling the simplified version we built today will help resolve many confusions.
Finally, remember: the best way to learn is through hands-on practice. Try extending our simplified framework—like adding dependency injection or more complete error handling. This will take your understanding of web frameworks to the next level!
Leapcell: The Best of Serverless Web Hosting
Finally, here's a platform ideal for deploying Python services: Leapcell
🚀 Build with Your Favorite Language
Develop effortlessly in JavaScript, Python, Go, or Rust.
🌍 Deploy Unlimited Projects for Free
Only pay for what you use—no requests, no charges.
⚡ Pay-as-You-Go, No Hidden Costs
No idle fees, just seamless scalability.
🔹 Follow us on Twitter: @LeapcellHQ