Implementing FastAPI from Scratch Using Only Pure Python: A Comprehensive Guide
FastAPI, a modern, high-performance web framework for building APIs with Python 3.7+, is known for its ease of use, speed, and automatic data validation. While FastAPI itself relies on libraries like Pydantic and Starlette, understanding its core principles and how it functions under the hood can be incredibly valuable. This article guides you through building a simplified FastAPI-like framework from scratch using only pure Python, helping you grasp the fundamentals of web framework design and the power of Python’s built-in libraries.
Table of Contents
- Introduction: Why Build a Framework from Scratch?
- Understanding the Basics: WSGI and ASGI
- What is WSGI?
- What is ASGI?
- Choosing the Right Protocol
- Setting Up the Project Structure
- Core Components: The Router
- Creating the Router Class
- Registering Routes
- Handling Different HTTP Methods (GET, POST, etc.)
- Request Handling: Parsing and Processing
- Parsing Request Headers
- Reading Request Body
- Query Parameter Handling
- Response Generation: Building HTTP Responses
- Setting Status Codes
- Setting Headers
- Returning Data as JSON
- Middleware Implementation
- Creating a Simple Middleware Class
- Applying Middleware to the Application
- Exception Handling
- Defining Custom Exceptions
- Implementing Exception Handlers
- Data Validation (Simplified)
- Creating a Basic Data Validation Class
- Validating Request Data
- Running the Application with a WSGI/ASGI Server
- Using
wsgireffor WSGI - Using
uvicornorhypercornfor ASGI
- Using
- Example Application: A Simple To-Do API
- Limitations and Next Steps
- Conclusion
1. Introduction: Why Build a Framework from Scratch?
You might wonder, why bother building a web framework from scratch when robust frameworks like FastAPI, Flask, and Django already exist? There are several compelling reasons:
- Deeper Understanding: Building a framework forces you to understand the underlying mechanics of web servers, routing, request handling, and response generation. This knowledge is invaluable for debugging, optimizing, and extending existing frameworks.
- Customization: When you control the entire stack, you can tailor the framework to your specific needs. This can be useful in niche scenarios where existing frameworks are overkill or don’t quite fit.
- Educational Exercise: It’s an excellent learning exercise to improve your Python skills and your understanding of software architecture principles.
- Debugging Skills: By understanding the inner workings, debugging becomes significantly easier. You can trace the flow of execution step-by-step.
- Appreciation for Existing Frameworks: You’ll gain a greater appreciation for the complexities handled by frameworks like FastAPI and understand why they are so powerful and efficient.
This article focuses on building a simplified framework, omitting many features of a production-ready framework for clarity and brevity. The goal is to illustrate the core concepts, not to create a fully functional alternative to FastAPI.
2. Understanding the Basics: WSGI and ASGI
Before diving into the code, it’s crucial to understand the underlying protocols that power web applications: WSGI and ASGI.
What is WSGI?
WSGI (Web Server Gateway Interface) is a standard interface between web servers (like Apache or Gunicorn) and Python web applications. It defines a simple calling convention that allows the server to pass requests to the application and receive responses in a standardized format.
In essence, a WSGI application is a callable (usually a function) that takes two arguments:
environ: A dictionary containing information about the request, such as headers, method, and URL.start_response: A callable that the application uses to set the HTTP status code and headers.
The application then returns an iterable (usually a list) of byte strings representing the response body.
What is ASGI?
ASGI (Asynchronous Server Gateway Interface) is a successor to WSGI, designed to handle asynchronous operations and long-lived connections like WebSockets. ASGI allows your application to handle multiple requests concurrently without blocking, leading to better performance and scalability.
ASGI applications are also callables, but their signature is slightly different and they operate asynchronously. They typically use async def to define asynchronous functions and leverage await to handle asynchronous operations.
Choosing the Right Protocol
For this simplified framework, we’ll focus on ASGI as it’s more modern and supports both synchronous and asynchronous operations. We will implement a minimal ASGI application capable of handling HTTP requests.
3. Setting Up the Project Structure
Let’s create a basic project structure to organize our code:
my_framework/
├── app.py # Main application file
├── router.py # Routing logic
├── request.py # Request handling
├── response.py # Response generation
├── middleware.py # Middleware implementation
└── exceptions.py # Custom exceptions
4. Core Components: The Router
The router is responsible for mapping incoming HTTP requests to the appropriate handler functions. It’s the heart of our framework.
Creating the Router Class
Create a file named router.py and add the following code:
“`python
# router.py
class Router:
def __init__(self):
self.routes = {}
def add_route(self, path: str, methods: list, handler):
if path not in self.routes:
self.routes[path] = {}
for method in methods:
self.routes[path][method.upper()] = handler
def get(self, path: str):
def decorator(handler):
self.add_route(path, [“GET”], handler)
return handler
return decorator
def post(self, path: str):
def decorator(handler):
self.add_route(path, [“POST”], handler)
return handler
return decorator
def put(self, path: str):
def decorator(handler):
self.add_route(path, [“PUT”], handler)
return handler
return decorator
def delete(self, path: str):
def decorator(handler):
self.add_route(path, [“DELETE”], handler)
return handler
return decorator
def patch(self, path: str):
def decorator(handler):
self.add_route(path, [“PATCH”], handler)
return handler
return decorator
def route(self, path: str, methods: list):
def decorator(handler):
self.add_route(path, methods, handler)
return handler
return decorator
async def handle_request(self, scope, receive, send):
method = scope[“method”]
path = scope[“path”]
if path in self.routes and method in self.routes[path]:
handler = self.routes[path][method]
await handler(scope, receive, send)
else:
await self.default_response(scope, receive, send)
async def default_response(self, scope, receive, send):
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”,
})
“`
Registering Routes
The add_route method registers a handler function for a specific path and HTTP method. The convenience methods (get, post, put, delete, patch) provide a more declarative way to define routes using decorators.
Example usage:
“`python
# In app.py (example, assuming you’ve imported the Router class)
from router import Router
router = Router()
@router.get(“/”)
async def index_handler(scope, receive, send):
await send({
“type”: “http.response.start”,
“status”: 200,
“headers”: [
[b”content-type”, b”text/plain”],
],
})
await send({
“type”: “http.response.body”,
“body”: b”Hello, world!”,
})
@router.post(“/items”)
async def create_item_handler(scope, receive, send):
# Handle the request body to create an item
await send({
“type”: “http.response.start”,
“status”: 201,
“headers”: [
[b”content-type”, b”text/plain”],
],
})
await send({
“type”: “http.response.body”,
“body”: b”Item Created”,
})
“`
Handling Different HTTP Methods (GET, POST, etc.)
The Router class provides decorators for common HTTP methods. Each decorator calls add_route with the appropriate method name to register the handler.
5. Request Handling: Parsing and Processing
Handling incoming requests involves parsing headers, reading the body, and extracting query parameters. Create a file named request.py for this purpose.
“`python
# request.py
import json
from urllib.parse import urlparse, parse_qs
class Request:
def __init__(self, scope, receive):
self.scope = scope
self.receive = receive
self._body = None
self._parsed_body = None
self._query_params = None
@property
def method(self) -> str:
return self.scope[“method”]
@property
def path(self) -> str:
return self.scope[“path”]
@property
def headers(self) -> dict:
headers = {}
for name, value in self.scope[“headers”]:
headers[name.decode(“utf-8”)] = value.decode(“utf-8″)
return headers
async def body(self) -> bytes:
if self._body is None:
body = b””
more_body = True
while more_body:
message = await self.receive()
body += message.get(“body”, b””)
more_body = message.get(“more_body”, False)
self._body = body
return self._body
async def json(self) -> dict:
if self._parsed_body is None:
body = await self.body()
try:
self._parsed_body = json.loads(body.decode(“utf-8”))
except json.JSONDecodeError:
self._parsed_body = {} # Or raise an exception
return self._parsed_body
@property
def query_params(self) -> dict:
if self._query_params is None:
query_string = self.scope.get(“query_string”, b””).decode(“utf-8”)
self._query_params = parse_qs(query_string)
return self._query_params
“`
Parsing Request Headers
The headers property extracts headers from the scope dictionary provided by ASGI.
Reading Request Body
The body method asynchronously reads the request body from the receive callable. It accumulates the body in chunks until the more_body flag is False. The json method then parses the body as JSON.
Query Parameter Handling
The query_params property parses the query string from the scope dictionary using urllib.parse.
Usage example:
“`python
# In your route handler
from request import Request
async def my_handler(scope, receive, send):
request = Request(scope, receive)
data = await request.json()
print(f”Received data: {data}”)
# … process the data …
“`
6. Response Generation: Building HTTP Responses
Generating HTTP responses involves setting the status code, headers, and body. Create a file named response.py:
“`python
# response.py
import json
class Response:
def __init__(self, content: str = “”, status_code: int = 200, headers: dict = None, media_type: str = “text/plain”):
self.content = content
self.status_code = status_code
self.headers = headers or {}
self.media_type = media_type
async def __call__(self, scope, receive, send):
headers = [[b”content-type”, self.media_type.encode(“utf-8”)]]
for name, value in self.headers.items():
headers.append([name.encode(“utf-8”), value.encode(“utf-8”)])
await send({
“type”: “http.response.start”,
“status”: self.status_code,
“headers”: headers,
})
await send({
“type”: “http.response.body”,
“body”: self.content.encode(“utf-8″),
})
class JSONResponse(Response):
def __init__(self, content: dict, status_code: int = 200, headers: dict = None):
super().__init__(
content=json.dumps(content),
status_code=status_code,
headers=headers,
media_type=”application/json”,
)
“`
Setting Status Codes
The status_code attribute sets the HTTP status code of the response.
Setting Headers
The headers attribute is a dictionary that allows you to set custom HTTP headers.
Returning Data as JSON
The JSONResponse class is a specialized response class that automatically serializes a dictionary to JSON and sets the Content-Type header to application/json.
Usage example:
“`python
# In your route handler
from response import Response, JSONResponse
async def my_handler(scope, receive, send):
return JSONResponse(content={“message”: “Hello, world!”}, status_code=200)(scope, receive, send)
async def another_handler(scope, receive, send):
return Response(content=”Plain text response”, status_code=200)(scope, receive, send)
“`
7. Middleware Implementation
Middleware allows you to intercept and process requests and responses globally. This is useful for tasks like authentication, logging, and request modification. Create a file named middleware.py:
“`python
# middleware.py
class Middleware:
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send):
# Pre-processing logic (before the request reaches the handler)
print(“Middleware: Request received”)
async def wrapped_send(message):
# Post-processing logic (before the response is sent)
if message[“type”] == “http.response.start”:
print(“Middleware: Response status:”, message[“status”])
await send(message)
await self.app(scope, receive, wrapped_send)
“`
Creating a Simple Middleware Class
The Middleware class takes the application as an argument in its constructor. The __call__ method is the entry point for the middleware. It receives the scope, receive, and send callables.
Applying Middleware to the Application
To apply middleware, you wrap the application instance with the middleware class:
“`python
# In app.py
from middleware import Middleware
# … your router and route definitions …
app = router.handle_request # or whatever you named the method to handle requests
app = Middleware(app)
“`
The middleware intercepts the request before it reaches the handler, executes its pre-processing logic, calls the next application in the chain (in this case, the route handler), and then executes its post-processing logic before sending the response.
8. Exception Handling
Robust exception handling is crucial for building reliable applications. Create a file named exceptions.py:
“`python
# exceptions.py
class HTTPException(Exception):
def __init__(self, status_code: int, detail: str = None):
self.status_code = status_code
self.detail = detail
“`
Defining Custom Exceptions
The HTTPException class is a custom exception that represents an HTTP error. It takes a status_code and an optional detail message as arguments.
Implementing Exception Handlers
You can implement global exception handlers to catch exceptions and return appropriate HTTP responses:
“`python
# In app.py
from exceptions import HTTPException
from response import JSONResponse
async def app(scope, receive, send):
try:
# … your route handling logic …
await router.handle_request(scope, receive, send)
except HTTPException as e:
await JSONResponse(content={“detail”: e.detail}, status_code=e.status_code)(scope, receive, send)
except Exception as e:
# Handle unexpected errors
await JSONResponse(content={“detail”: “Internal Server Error”}, status_code=500)(scope, receive, send)
“`
This example catches HTTPException and returns a JSON response with the error details. It also catches other exceptions and returns a generic “Internal Server Error” response.
9. Data Validation (Simplified)
Data validation ensures that incoming data conforms to the expected format and constraints. This example provides a very basic implementation.
Creating a Basic Data Validation Class
“`python
# In app.py or a separate file
class ValidationError(Exception):
def __init__(self, message):
self.message = message
class Validator:
def __init__(self, schema: dict):
self.schema = schema
def validate(self, data: dict):
for field, expected_type in self.schema.items():
if field not in data:
raise ValidationError(f”Missing field: {field}”)
if not isinstance(data[field], expected_type):
raise ValidationError(f”Invalid type for field {field}. Expected {expected_type}, got {type(data[field])}”)
return data
“`
Validating Request Data
Usage within a route handler:
“`python
from response import JSONResponse
from request import Request
async def create_item_handler(scope, receive, send):
request = Request(scope, receive)
try:
data = await request.json()
schema = {“name”: str, “description”: str, “price”: (int, float)}
validated_data = Validator(schema).validate(data)
# Process validated data here
await JSONResponse(content={“message”: “Item created”, “data”: validated_data}, status_code=201)(scope, receive, send)
except ValidationError as e:
await JSONResponse(content={“detail”: e.message}, status_code=400)(scope, receive, send)
except Exception as e:
await JSONResponse(content={“detail”: “Invalid JSON”}, status_code=400)(scope, receive, send)
“`
This example defines a schema for the request data and validates that the required fields are present and have the correct types. If validation fails, it raises a ValidationError and returns a 400 Bad Request response.
10. Running the Application with a WSGI/ASGI Server
Now that we have the basic framework components, let’s run the application using a WSGI/ASGI server.
Using wsgiref for WSGI
While our framework is primarily ASGI, we can use wsgiref (Python’s built-in WSGI server) for simple testing. This requires adapting our ASGI application to a WSGI-compatible one.
First, install asgiref (if you don’t have it already):
pip install asgirefThen, modify app.py:
“`python
# app.py
import asyncio
from wsgiref.simple_server import make_server
from asgiref.wsgi import WsgiToAsgi
from router import Router
router = Router()
@router.get(“/”)
async def index_handler(scope, receive, send):
await send({
“type”: “http.response.start”,
“status”: 200,
“headers”: [
[b”content-type”, b”text/plain”],
],
})
await send({
“type”: “http.response.body”,
“body”: b”Hello, world!”,
})
app = router.handle_request
asgi_app = app # This is already an ASGI app, but we’re showing the variable for clarity
# Adapt ASGI app to WSGI using asgiref
wsgi_app = WsgiToAsgi(asgi_app)
if __name__ == “__main__”:
with make_server(“”, 8000, wsgi_app) as httpd:
print(“Serving on port 8000…”)
httpd.serve_forever()
“`
Run this script using python app.py and access http://localhost:8000 in your browser.
Using uvicorn or hypercorn for ASGI
For proper ASGI support and better performance, use uvicorn or hypercorn:
pip install uvicorn
# or
pip install hypercornModify your app.py to be minimal:
“`python
# app.py
from router import Router
router = Router()
@router.get(“/”)
async def index_handler(scope, receive, send):
await send({
“type”: “http.response.start”,
“status”: 200,
“headers”: [
[b”content-type”, b”text/plain”],
],
})
await send({
“type”: “http.response.body”,
“body”: b”Hello, world!”,
})
app = router.handle_request
“`
Then, run the application using:
uvicorn app:app --reload
# or
hypercorn app:app --reloadAccess http://localhost:8000 in your browser. The --reload flag enables automatic reloading on code changes.
11. Example Application: A Simple To-Do API
Let’s create a simple To-Do API using our framework:
“`python
# app.py
import asyncio
import json
from router import Router
from request import Request
from response import JSONResponse, Response
from exceptions import HTTPException
router = Router()
todos = []
todo_id_counter = 1
@router.get(“/todos”)
async def list_todos(scope, receive, send):
return JSONResponse(content=todos)(scope, receive, send)
@router.post(“/todos”)
async def create_todo(scope, receive, send):
global todo_id_counter
request = Request(scope, receive)
try:
data = await request.json()
if not isinstance(data, dict) or “task” not in data:
raise HTTPException(status_code=400, detail=”Invalid todo data. ‘task’ field required.”)
new_todo = {“id”: todo_id_counter, “task”: data[“task”], “completed”: False}
todos.append(new_todo)
todo_id_counter += 1
return JSONResponse(content=new_todo, status_code=201)(scope, receive, send)
except HTTPException as e:
return JSONResponse(content={“detail”: e.detail}, status_code=e.status_code)(scope, receive, send)
except Exception as e:
return JSONResponse(content={“detail”: “Invalid JSON data”}, status_code=400)(scope, receive, send)
@router.get(“/todos/{todo_id}”)
async def get_todo(scope, receive, send):
try:
todo_id = int(scope[“path”].split(“/”)[-1])
todo = next((t for t in todos if t[“id”] == todo_id), None)
if todo:
return JSONResponse(content=todo)(scope, receive, send)
else:
raise HTTPException(status_code=404, detail=”Todo not found”)
except ValueError:
return JSONResponse(content={“detail”: “Invalid todo ID”}, status_code=400)(scope, receive, send)
except HTTPException as e:
return JSONResponse(content={“detail”: e.detail}, status_code=e.status_code)(scope, receive, send)
@router.put(“/todos/{todo_id}”)
async def update_todo(scope, receive, send):
request = Request(scope, receive)
try:
todo_id = int(scope[“path”].split(“/”)[-1])
data = await request.json()
if not isinstance(data, dict) or “task” not in data:
raise HTTPException(status_code=400, detail=”Invalid todo data. ‘task’ field required.”)
todo_index = next((i for i, t in enumerate(todos) if t[“id”] == todo_id), None)
if todo_index is None:
raise HTTPException(status_code=404, detail=”Todo not found”)
else:
todos[todo_index][“task”] = data[“task”]
return JSONResponse(content=todos[todo_index])(scope, receive, send)
except ValueError:
return JSONResponse(content={“detail”: “Invalid todo ID”}, status_code=400)(scope, receive, send)
except HTTPException as e:
return JSONResponse(content={“detail”: e.detail}, status_code=e.status_code)(scope, receive, send)
except Exception as e:
return JSONResponse(content={“detail”: “Invalid JSON data”}, status_code=400)(scope, receive, send)
@router.delete(“/todos/{todo_id}”)
async def delete_todo(scope, receive, send):
try:
todo_id = int(scope[“path”].split(“/”)[-1])
todo_index = next((i for i, t in enumerate(todos) if t[“id”] == todo_id), None)
if todo_index is None:
raise HTTPException(status_code=404, detail=”Todo not found”)
else:
del todos[todo_index]
return Response(status_code=204)(scope, receive, send)
except ValueError:
return JSONResponse(content={“detail”: “Invalid todo ID”}, status_code=400)(scope, receive, send)
except HTTPException as e:
return JSONResponse(content={“detail”: e.detail}, status_code=e.status_code)(scope, receive, send)
app = router.handle_request
“`
This example demonstrates:
- Listing all to-dos (GET /todos)
- Creating a new to-do (POST /todos)
- Retrieving a specific to-do (GET /todos/{todo_id})
- Updating a to-do (PUT /todos/{todo_id})
- Deleting a to-do (DELETE /todos/{todo_id})
- Basic error handling using custom exceptions.
12. Limitations and Next Steps
This simplified framework has several limitations:
- Limited Functionality: It lacks many features of a production-ready framework, such as:
- Automatic data validation (using Pydantic, for example)
- Dependency injection
- Authentication and authorization
- Testing framework
- More robust exception handling
- Background tasks
- WebSockets support
- Static file serving
- Security: It doesn’t include security features like CSRF protection or input sanitization.
- Performance: It’s not optimized for performance and may not scale well under heavy load.
- No Documentation: This implementation is not documented, making it harder to maintain and extend.
Next steps for improving the framework could include:
- Implementing more robust data validation using a library like Pydantic.
- Adding support for dependency injection.
- Implementing authentication and authorization mechanisms.
- Adding a testing framework.
- Optimizing performance using techniques like caching and connection pooling.
- Adding WebSocket support.
13. Conclusion
Building a web framework from scratch is a challenging but rewarding exercise. It provides a deep understanding of the underlying principles of web application development and helps you appreciate the complexities handled by existing frameworks like FastAPI. While this simplified framework is not intended for production use, it serves as a valuable learning tool for aspiring web developers and software architects.
“`
