Eine vereinfachte FastAPI von Grund auf neu erstellen: ASGI und Core Routing verstehen

Einführung: Warum dieses Rad neu erfinden?

Wenn wir über asynchrone Python-Web-Frameworks sprechen, ist FastAPI zweifellos der hellste Stern der letzten Jahre. Es hat breite Anerkennung für seine beeindruckende Leistung, die automatische Generierung von API-Dokumentationen und die Unterstützung von Type Hints gefunden. Aber haben Sie sich jemals gefragt: Welcher Zauber steckt hinter diesem leistungsstarken Framework?

Heute werden wir eine vereinfachte Version von FastAPI von Grund auf neu erstellen und uns dabei auf das Verständnis zweier Kernkonzepte konzentrieren: das ASGI-Protokoll und das Routing-System. Indem wir es mit unseren eigenen Händen konstruieren, werden Sie die Funktionsprinzipien moderner asynchroner Web-Frameworks verstehen. Dies hilft Ihnen nicht nur, FastAPI besser zu nutzen, sondern ermöglicht es Ihnen auch, die Ursache schnell zu erkennen, wenn Probleme auftreten.

Was ist ASGI? Warum ist es fortschrittlicher als WSGI?

Bevor wir mit dem Programmieren beginnen, müssen wir ASGI (Asynchronous Server Gateway Interface) verstehen – das Fundament, das es FastAPI ermöglicht, eine hochleistungsfähige asynchrone Verarbeitung zu erreichen.

Einschränkungen von WSGI

Wenn Sie Django oder Flask verwendet haben, haben Sie wahrscheinlich von WSGI (Web Server Gateway Interface) gehört. WSGI ist eine synchrone Schnittstellenspezifikation zwischen Python-Webanwendungen und Servern, hat aber offensichtliche Mängel:

• Kann nur eine Anfrage gleichzeitig bearbeiten, keine Nebenläufigkeit
• Unterstützt keine langlebigen Verbindungen (wie WebSocket)
• Kann die Vorteile von asynchronem I/O nicht voll ausschöpfen

Vorteile von ASGI

ASGI wurde entwickelt, um diese Probleme zu lösen:

• Vollständig asynchron, unterstützt die gleichzeitige Verarbeitung mehrerer Anfragen
• Kompatibel mit WebSocket und HTTP/2
• Ermöglicht Middleware die Arbeit in asynchronen Umgebungen
• Unterstützt asynchrone Ereignisse während des gesamten Anfragelebenszyklus

Einfach ausgedrückt definiert ASGI eine Standardschnittstelle, die es asynchronen Webanwendungen ermöglicht, mit Servern (wie Uvicorn) zu kommunizieren. Als Nächstes werden wir einen minimalistischen ASGI-Server implementieren.

Schritt 1: Implementieren eines einfachen ASGI-Servers

Eine ASGI-Anwendung ist im Wesentlichen ein aufrufbares Objekt (Funktion oder Klasse), das drei Parameter empfängt: Scope, Receive und Send.

# asgi_server.py
import socket
import asyncio
import json
from typing import Callable, Awaitable, Dict, Any

# ASGI-Anwendungstypdefinition
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  # Standardanwendung
    
    async def default_app(self, scope: Dict[str, Any], receive: Callable, send: Callable):
        \"\"\"Standardanwendung: gibt 404-Antwort zurück\"\"\"        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):
        \"\"\"Verarbeitet neue Verbindungen, parst HTTP-Anfragen und übergibt sie an die ASGI-Anwendung\"\"\"        data = await reader.read(1024)
        request = data.decode().split("\r\n")
        method, path, _ = request[0].split()
        
        # Erstellen des ASGI-Scopes
        scope = {
            \"type\": \"http\",
            \"method\": method,
            \"path\": path,
            \"headers\": []
        }
        
        # Parsen der Request-Header
        for line in request[1:]:
            if line == \"\":
                break
            key, value = line.split(":", 1)
            scope[\"headers\"] .append((key.strip().lower().encode(), value.strip().encode()))
        
        # Definieren der Receive- und Send-Methoden
        async def receive() -> Dict:
            \"\"\"Simuliert den Empfang von Nachrichten (vereinfachte Version)\"\"\"            return {\"type\": \"http.request\", \"body\": b\"\"}
        
        async def send(message: Dict):
            \"\"\"Sendet Antwort an den 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()
        
        # Aufrufen der ASGI-Anwendung
        await self.app(scope, receive, send)
    
    async def run(self):
        \"\"\"Startet den Server\"\"\"        server = await asyncio.start_server(
            self.handle_connection, self.host, self.port
        )
        print(f\"Server wird ausgeführt auf http://{self.host}:{self.port}\")
        async with server:
            await server.serve_forever()

# Server ausführen
if __name__ == \"__main__\":
    server = ASGIServer()
    asyncio.run(server.run())

Dieser vereinfachte ASGI-Server kann grundlegende HTTP-Anfragen verarbeiten und Antworten zurückgeben. Testen Sie es aus: Besuchen Sie nach dem Ausführen des Skripts http://127.0.0.1:8000 und Sie sehen "Not Found", da wir noch keine Routen definiert haben.

Schritt 2: Implementieren des Routing-Systems

Eine der intuitivsten Funktionen von FastAPI ist seine elegante Routendefinition, wie zum Beispiel:

@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    return {\"item_id\": item_id, \"q\": q}

Lassen Sie uns eine ähnliche Routing-Funktionalität implementieren.

Design der Routing-Kernkomponente

Wir benötigen drei Kernkomponenten:

• Router: Verwaltet alle Routing-Regeln
• Dekoratoren: @get, @post usw. zum Registrieren von Routen
• Pfadabgleich: Behandelt dynamische Pfadparameter (wie /items/{item_id})

# router.py
from typing import Callable, Awaitable, Dict, Any, List, Tuple, Pattern
import re
from functools import wraps

# Routentypdefinition
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]]:
        \"\"\"Konvertiert den Pfad in einen regulären Ausdruck und extrahiert Parameternamen\"\"\"        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]]:
        \"\"\"Gleicht Pfad und Methode ab, gibt Parameter zurück\"\"\"        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):
        \"\"\"Fügt eine Route hinzu\"\"\"        self.routes.append(Route(path, methods, handler))
    
    def route(self, path: str, methods: List[str]):
        \"\"\"Routendekorator\"\"\"        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-Methoden
    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]:
        \"\"\"Verarbeitet Anfragen, findet die passende Route und führt sie aus\"\"\"        path = scope[\"path\"]
        method = scope[\"method\"]
        
        for route in self.routes:
            matched, params = route.match(path, method)
            if matched:
                # Parsen der Query-Parameter
                query_params = self.parse_query_params(scope)
                # Zusammenführen von Pfadparametern und Query-Parametern
                request_data = {** params, **query_params}
                # Aufrufen der Handler-Funktion
                return await route.handler(request_data)
        
        # Keine Route gefunden
        return {\"status\": 404, \"body\": {\"detail\": \"Not Found\"}}
    
    def parse_query_params(self, scope: Dict[str, Any]) -> Dict[str, Any]:
        \"\"\"Parsen der Query-Parameter (vereinfachte Version)\"\"\"        # Im tatsächlichen ASGI befinden sich die Query-Parameter 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

Integrieren des Routings mit dem ASGI-Server

Jetzt müssen wir unseren ASGI-Server so ändern, dass er unser Routing-System verwendet:

# Hinzufügen von Routing-Unterstützung zur ASGIServer-Klasse
class ASGIServer:
    def __init__(self, host: str = \"127.0.0.1\", port: int = 8000):
        self.host = host
        self.port = port
        self.router = Router()  # Router instanziieren
        self.app = self.asgi_app  # Verwenden der Routing-fähigen ASGI-Anwendung
    
    async def asgi_app(self, scope: Dict[str, Any], receive: Callable, send: Callable):
        \"\"\"ASGI-Anwendung mit Routing-Funktionalität\"\"\"        if scope[\"type\"] == \"http\":
            # Anfrage verarbeiten
            response = await self.router.handle(scope, receive)
            status = response.get(\"status\", 200)
            body = json.dumps(response.get(\"body\", {})).encode()
            
            # Senden der Antwort
            await send({
                \"type\": \"http.response.start\",
                \"status\": status,
                \"headers\": [(b\"content-type\", b\"application/json\")]
            })
            await send({
                \"type\": \"http.response.body\",
                \"body\": body
            })

Schritt 3: Implementieren der Parameteranalyse und Typkonvertierung

Einer der Höhepunkte von FastAPI ist seine automatische Parameteranalyse und Typkonvertierung. Lassen Sie uns diese Funktion implementieren:

# Hinzufügen der Typkonvertierung zur Handle-Methode des Routers
async def handle(self, scope: Dict[str, Any], receive: Callable) -> Dict[str, Any]:
    # ... vorheriger Code ...
    
    if matched:
        # Parsen der Query-Parameter
        query_params = self.parse_query_params(scope)
        # Zusammenführen von Pfadparametern und Query-Parametern
        raw_data = {** params, **query_params}
        
        # Abrufen der Parameter-Typannotationen aus der Handler-Funktion
        handler_params = route.handler.__annotations__
        
        # Typkonvertierung
        request_data = {}
        for key, value in raw_data.items():
            if key in handler_params:
                target_type = handler_params[key]
                try:
                    # Versuch der Typkonvertierung
                    request_data[key] = target_type(value)
                except (ValueError, TypeError):
                    return {
                        \"status\": 400,
                        \"body\": {\"detail\": f\"Ungültiger Typ für {key}, erwartet {target_type}\"}
                    }
            else:
                request_data[key] = value
        
        # Aufrufen der Handler-Funktion
        return await route.handler(request_data)

Jetzt kann unser Framework Parameter automatisch in die von den Funktionsannotationen angegebenen Typen konvertieren!

Schritt 4: Implementieren der Analyse des Anfragetextes (POST-Unterstützung)

Als Nächstes fügen wir Unterstützung für POST-Anfragetexte hinzu und aktivieren die JSON-Datenanalyse:

# Hinzufügen der Analyse des Anfragetextes zum Router
async def handle(self, scope: Dict[str, Any], receive: Callable) -> Dict[str, Any]:
    # ... vorheriger Code ...
    
    # Wenn es sich um eine POST-Anfrage handelt, analysieren Sie den Anfragetext
    request_body = {}
    if method == \"POST\":
        # Abrufen des Anfragetextes von 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\": \"Ungültiges JSON\"}
                }
    
    # Zusammenführen aller Parameter
    raw_data = {** params, **query_params,** request_body}
    # ... Typkonvertierung und Aufruf der Handler-Funktion ...

Schritt 5: Erstellen einer vollständigen Beispielanwendung

Jetzt können wir unser Framework genauso wie FastAPI verwenden:

# main.py
from asgi_server import ASGIServer
import asyncio

# Erstellen einer Serverinstanz (einschließlich Router)
app = ASGIServer()
router = app.router

# Definieren von Routen
@router.get("/")
async def root():
    return {\"message\": \"Hallo Welt!\"}

@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}}

# Ausführen der Anwendung
if __name__ == \"__main__\":
    asyncio.run(app.run())

Testen Sie diese Anwendung:

• Besuchen Sie http://127.0.0.1:8000 → Willkommensnachricht erhalten
• Besuchen Sie http://127.0.0.1:8000/items/42?q=test → Antwort mit Parametern erhalten
• Senden Sie eine POST-Anfrage an http://127.0.0.1:8000/items/ mit {"name": "Apple", "price": 1.99} → Erstellten Artikel erhalten

Unterschiede zu FastAPI und Optimierungsrichtungen

Unsere vereinfachte Version implementiert die Kernfunktionalität von FastAPI, aber das echte FastAPI verfügt über viele erweiterte Funktionen:

• Dependency-Injection-System: Die Dependency Injection von FastAPI ist sehr leistungsstark und unterstützt verschachtelte Abhängigkeiten, globale Abhängigkeiten usw.
• Automatische Dokumentation: FastAPI kann automatisch Swagger- und ReDoc-Dokumentation generieren
• Mehr Datentypunterstützung: Einschließlich Pydantic-Modellvalidierung, Formulardaten, Dateiuploads usw.
• Middleware-System: Vollständigere Middleware-Unterstützung
• WebSocket-Unterstützung: Vollständige Implementierung der WebSocket-Spezifikation von ASGI
• Asynchrone Datenbanktools: Tiefe Integration mit Tools wie SQLAlchemy

Zusammenfassung: Was haben wir gelernt?

Durch diese praktische Übung haben wir Folgendes verstanden:

• Die grundlegenden Funktionsprinzipien des ASGI-Protokolls: die drei Elemente Scope, Receive und Send
• Der Kern des Routing-Systems: Pfadabgleich, Parameteranalyse und Handler-Funktionszuordnung
• Wie die Typkonvertierung implementiert wird: Verwenden von Funktionsannotationen zur automatischen Konvertierung
• Der Anfrageverarbeitungsprozess: Der vollständige Lebenszyklus vom Empfangen einer Anfrage bis zum Zurückgeben einer Antwort

Dieses Wissen gilt nicht nur für FastAPI, sondern auch für alle ASGI-Frameworks (wie Starlette, Quart usw.). Wenn Sie bei der Verwendung dieser Frameworks auf Probleme stoßen, wird Ihnen das Erinnern an die vereinfachte Version, die wir heute erstellt haben, helfen, viele Unklarheiten zu beseitigen.

Denken Sie schließlich daran: Der beste Weg zu lernen ist durch praktisches Üben. Versuchen Sie, unser vereinfachtes Framework zu erweitern – z. B. durch Hinzufügen von Dependency Injection oder einer vollständigeren Fehlerbehandlung. Dies wird Ihr Verständnis von Web-Frameworks auf die nächste Stufe heben!

Leapcell: Das Beste aus Serverless Webhosting

Schließlich ist hier eine Plattform, die sich ideal für die Bereitstellung von Python-Diensten eignet: Leapcell

🚀 Entwickeln Sie mit Ihrer Lieblingssprache

Entwickeln Sie mühelos in JavaScript, Python, Go oder Rust.

🌍 Stellen Sie unbegrenzt Projekte kostenlos bereit

Zahlen Sie nur für das, was Sie nutzen – keine Anfragen, keine Gebühren.

⚡ Pay-as-You-Go, keine versteckten Kosten

Keine Leerlaufgebühren, nur nahtlose Skalierbarkeit.

📖 Erkunden Sie unsere Dokumentation

🔹 Folgen Sie uns auf Twitter: @LeapcellHQ