Posted on Aug 6

Build a Mini-FastAPI from Scratch: Learn ASGI & Routing Internals

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!