FastAPI ist im Wesentlichen ein API-Wrapper für Starlette. Um FastAPI vollständig zu verstehen, muss man zuerst Starlette verstehen.

1. ASGI-Protokoll

Uvicorn interagiert über eine gemeinsame Schnittstelle mit ASGI-Anwendungen. Eine Anwendung kann über Uvicorn Informationen senden und empfangen, indem sie den folgenden Code implementiert:

async def app(scope, receive, send):
    # Die einfachste ASGI-Anwendung
    assert scope['type'] == 'http'
    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!',
    })

if __name__ == "__main__":
    # Uvicorn-Dienst
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=5000, log_level="info")

2. Starlette

Um Starlette mit Uvicorn zu starten, verwenden Sie den folgenden Code:

from starlette.applications import Starlette
from starlette.middleware.gzip import GZipMiddleware

app: Starlette = Starlette()

@app.route("/")
def demo_route() -> None: pass

@app.websocket_route("/")
def demo_websocket_route() -> None: pass

@app.add_exception_handlers(404)
def not_found_route() -> None: pass

@app.on_event("startup")
def startup_event_demo() -> None: pass

@app.on_event("shutdown")
def shutdown_event_demo() -> None: pass

app.add_middleware(GZipMiddleware)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=5000)

Dieser Code initialisiert Starlette, registriert Routen, Ausnahmere handler, Ereignisse und Middleware und übergibt sie dann an uvicorn.run. Die Methode uvicorn.run sendet Anfragedaten, indem sie die call-Methode von Starlette aufruft.

Analysieren wir zunächst die Initialisierung von Starlette:

class Starlette:
    def __init__(
        self,
        debug: bool = False,
        routes: typing.Sequence[BaseRoute] = None,
        middleware: typing.Sequence[Middleware] = None,
        exception_handlers: typing.Dict[
            typing.Union[int, typing.Type[Exception]], typing.Callable
        ] = None,
        on_startup: typing.Sequence[typing.Callable] = None,
        on_shutdown: typing.Sequence[typing.Callable] = None,
        lifespan: typing.Callable[["Starlette"], typing.AsyncGenerator] = None,
    ) -> None:
        """
        :param debug: Entscheiden Sie, ob die Debug-Funktion aktiviert werden soll.
        :param route: Eine Liste von Routen, die HTTP- und WebSocket-Dienste bereitstellen.
        :param middleware: Eine Liste von Middleware, die auf jede Anfrage angewendet wird.
        :param exception_handler: Ein Wörterbuch, das Ausnahmekennrufe speichert, wobei HTTP-Statuscodes als Schlüssel und Callback-Funktionen als Werte dienen.
        :on_startup: Callback-Funktionen, die beim Start aufgerufen werden.
        :on_shutdown: Callback-Funktionen, die beim Herunterfahren aufgerufen werden.
        :lifespan: Die Lebensdauerfunktion in ASGI.
        """

        # Wenn Lifespan übergeben wird, können on_startup und on_shutdown nicht übergeben werden
        # Da Starlette im Wesentlichen on_start_up und on_shutdown in eine Lebensdauer für Uvicorn-Aufrufe umwandelt
        assert lifespan is None or (
            on_startup is None and on_shutdown is None
        ), "Verwenden Sie entweder 'lifespan' oder 'on_startup'/'on_shutdown', nicht beides."

        # Variablen initialisieren
        self._debug = debug
        self.state = State()
        self.router = Router(
            routes, on_startup=on_startup, on_shutdown=on_shutdown, lifespan=lifespan
        )
        self.exception_handlers = (
            {} if exception_handlers is None else dict(exception_handlers)
        )
        self.user_middleware = [] if middleware is None else list(middleware)
        # Middleware erstellen
        self.middleware_stack = self.build_middleware_stack()

Wie aus dem Code ersichtlich ist, erfüllt die Initialisierung bereits die meisten Anforderungen. Es gibt jedoch eine Funktion zum Erstellen von Middleware, die weiter analysiert werden muss:

class Starlette:
    def build_middleware_stack(self) -> ASGIApp:
        debug = self.debug
        error_handler = None
        exception_handlers = {}

        # Ausnahmebehandlungs-Callbacks parsen und in error_handler und exception_handlers speichern
        # Nur der HTTP-Statuscode 500 wird in error_handler gespeichert
        for key, value in self.exception_handlers.items():
            if key in (500, Exception):
                error_handler = value
            else:
                exception_handlers[key] = value

        # Verschiedene Arten von Middleware sortieren
        # Die erste Ebene ist ServerErrorMiddleware, die den Fehlerstapel ausgeben kann, wenn eine Ausnahme gefunden wird, oder im Debug-Modus eine Fehlerseite anzeigt, um das Debugging zu erleichtern.
        # Die zweite Ebene ist die Benutzermiddleware-Ebene, auf der alle benutzerregistrierten Middleware gespeichert werden.
        # Die dritte Ebene ist ExceptionMiddleware, die die Ausnahmebehandlungsebene ist und alle Ausnahmen behandelt, die während der Routenausführung ausgelöst werden.
        middleware = (
            [Middleware(ServerErrorMiddleware, handler=error_handler, debug=debug)]
            + self.user_middleware
            + [
                Middleware(
                    ExceptionMiddleware, handlers=exception_handlers, debug=debug
                )
            ]
        )

        # Schließlich die Middleware in die App laden
        app = self.router
        for cls, options in reversed(middleware):
            # cls ist die Middleware-Klasse selbst und options sind die Parameter, die wir übergeben
            # Es ist ersichtlich, dass die Middleware selbst auch eine ASGI APP ist und das Laden von Middleware wie das Verschachteln einer ASGI APP in eine andere ist, wie eine Matrjoschka-Puppe.
            app = cls(app=app, **options)

        # Da die Middleware verschachtelt geladen wird und der Aufruf über `call_next` erfolgt, um die obere ASGI APP aufzurufen, wird die umgekehrte Reihenfolgemethode verwendet.
        return app

Nachdem die Middleware erstellt wurde, ist die Initialisierung abgeschlossen, und die Methode uvicorn.run ruft die Methode call auf:

class Starlette:
    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        scope["app"] = self
        await self.middleware_stack(scope, receive, send)

Diese Methode ist einfach. Sie setzt die App im Anfragefluss durch scope für nachfolgende Aufrufe und startet dann die Anforderungsverarbeitung durch Aufrufen von middleware_stack. Aus dieser Methode und der Middleware-Initialisierung ist ersichtlich, dass die Middleware in Starlette ebenfalls eine ASGI APP ist (es ist auch ersichtlich, dass die Route am unteren Ende des Aufrufstapels eine ASGI APP ist). Gleichzeitig gibt Starlette die Ausnahmebehandlung an die Middleware ab, was bei anderen Webanwendungs-Frameworks selten zu sehen ist. Es ist ersichtlich, dass Starlette so konzipiert ist, dass jede Komponente so weit wie möglich eine ASGI APP ist.

2. Middleware

Wie oben erwähnt, ist Middleware in Starlette eine ASGI APP. Daher muss jede Middleware in Starlette eine Klasse sein, die der folgenden Form entspricht:

class BaseMiddleware:
    def __init__(self, app: ASGIApp) -> None:
        pass

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        pass

In starlette.middleware gibt es viele Middleware-Implementierungen, die dieser Anforderung entsprechen. Dieses Kapitel behandelt jedoch nicht alle Middleware, sondern wählt nur einige repräsentative aus, die von der Route am weitesten entfernt analysiert werden sollen.

2.1. Ausnahmebehandlungs-Middleware - ExceptionMiddleware

Die erste ist die ExceptionMiddleware. Benutzer interagieren nicht direkt mit dieser Middleware (daher ist sie nicht in starlette.middleware platziert), sondern interagieren indirekt über die folgende Methode mit ihr:

@app.app_exception_handlers(404)
def not_found_route() -> None: pass

Wenn der Benutzer diese Methode verwendet, hängt Starlette die Callback-Funktion im entsprechenden Wörterbuch ein, wobei der HTTP-Statuscode als Schlüssel und die Callback-Funktion als Wert dient. Wenn ExceptionMiddleware feststellt, dass die Routenanforderungsverarbeitung eine Ausnahme aufweist, kann sie die entsprechende Callback-Funktion über den HTTP-Statuscode der Ausnahmereaktion finden, die Anfrage und Ausnahme an die vom Benutzer montierte Callback-Funktion übergeben und schließlich das Ergebnis der Benutzer-Callback-Funktion an die vorherige ASGI APP zurückgeben. Darüber hinaus unterstützt ExceptionMiddleware auch die Ausnahme-Registrierung. Wenn die von der Route ausgelöste Ausnahme mit der registrierten Ausnahme übereinstimmt, wird der entsprechende Callback für diese Ausnahme-Registrierung aufgerufen. Der Quellcode und die Kommentare dieser Klasse lauten wie folgt:

class ExceptionMiddleware:
    def __init__(
        self,
        app: ASGIApp, handlers: dict = None, debug: bool = False
    ) -> None:
        self.app = app
        self.debug = debug  # TODO: Wir sollten 404-Fälle behandeln, wenn debug gesetzt ist.
        # Starlette unterstützt sowohl HTTP-Statuscodes als auch Exception-Typen
        self._status_handlers = {}  # type: typing.Dict[int, typing.Callable]
        self._exception_handlers = {
            HTTPException: self.http_exception
        }  # type: typing.Dict[typing.Type[Exception], typing.Callable]
        if handlers is not None:
            for key, value in handlers.items():
                self.add_exception_handler(key, value)

    def add_exception_handler(
        self,
        exc_class_or_status_code: typing.Union[int, typing.Type[Exception]],
        handler: typing.Callable,
    ) -> None:
        # Die von der Starlette-App-Methode montierten Ausnahme-Callbacks werden über diese Methode schließlich in _status_handlers oder _exception_handler der Klasse montiert.
        if isinstance(exc_class_or_status_code, int):
            self._status_handlers[exc_class_or_status_code] = handler
        else:
            assert issubclass(exc_class_or_status_code, Exception)
            self._exception_handlers[exc_class_or_status_code] = handler

    def _lookup_exception_handler(
        self,
        exc: Exception
    ) -> typing.Optional[typing.Callable]:
        # Callback-Funktion im Zusammenhang mit der registrierten Ausnahme suchen, entsprechende Callback-Funktion für die Ausnahme über mro finden
        # 
        # Der Benutzer kann eine Basisklasse montieren, und nachfolgende Unterklassen der montierten Ausnahme rufen auch den Callback auf, der für die Basisklasse registriert wurde.
        # Zum Beispiel registriert der Benutzer eine Basisklasse, und dann gibt es zwei Ausnahmen, Benutzer-Ausnahme und System-Ausnahme, die beide von dieser Basisklasse erben.
        # Wenn die Funktion später die Benutzer-Ausnahme oder die System-Ausnahme auslöst, wird der entsprechende Callback, der für die Basisklasse registriert wurde, ausgeführt.
        for cls in type(exc).__mro__:
            if cls in self._exception_handlers:
                return self._exception_handlers[cls]
        return None

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        # Die vertraute ASGI-Aufrufmethode
        if scope["type"]!= "http":
            # Unterstützt keine WebSocket-Anfragen
            await self.app(scope, receive, send)
            return

        # Verhindern, dass mehrere Ausnahmen in derselben Antwort auftreten
        response_started = False

        async def sender(message: Message) -> None:
            nonlocal response_started

            if message["type"] == "http.response.start":
                response_started = True
            await send(message)

        try:
            # Nächste ASGI APP aufrufen
            await self.app(scope, receive, sender)
        except Exception as exc:
            handler = None

            if isinstance(exc, HTTPException):
                # Wenn es sich um eine HTTPException handelt, suchen Sie sie im registrierten HTTP-Callback-Wörterbuch
                handler = self._status_handlers.get(exc.status_code)

            if handler is None:
                # Wenn es sich um eine normale Ausnahme handelt, suchen Sie sie im Ausnahmekennruf-Wörterbuch
                handler = self._lookup_exception_handler(exc)

            if handler is None:
                # Wenn die entsprechende Ausnahme nicht gefunden werden kann, werfen Sie sie nach oben
                raise exc from None

            # Nur eine Ausnahme pro Antwort behandeln
            if response_started:
                msg = "Behandelte Ausnahme abgefangen, aber die Antwort wurde bereits gestartet."
                raise RuntimeError(msg) from exc

            request = Request(scope, receive=receive)
            if asyncio.iscoroutinefunction(handler):
                response = await handler(request, exc)
            else:
                response = await run_in_threadpool(handler, request, exc)
            # Die Anfrage mit der vom Callback generierten Antwort verarbeiten
            await response(scope, receive, sender)

2.2. Benutzermiddleware

Als nächstes kommt die Benutzermiddleware, die Middleware, mit der wir am meisten in Berührung kommen. Bei der Verwendung von starlette.middleware erben wir normalerweise von einer Middleware namens BaseHTTPMiddleware und erweitern basierend auf dem folgenden Code:

class DemoMiddleware(BaseHTTPMiddleware):
    def __init__(
        self,
        app: ASGIApp,
    ) -> None:
        super(DemoMiddleware, self).__init__(app)

    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
        # Zuvor
        response: Response = await call_next(request)
        # Danach
        return response

Wenn Sie eine Vorverarbeitung vor der Anfrage durchführen möchten, schreiben Sie den entsprechenden Code im Block before. Wenn Sie nach der Anfrage verarbeiten möchten, schreiben Sie Code im Block after. Die Verwendung ist sehr einfach und sie befinden sich im selben Gültigkeitsbereich, was bedeutet, dass die Variablen in dieser Methode nicht über Kontext oder dynamische Variablen weitergegeben werden müssen (wenn Sie sich mit der Middleware-Implementierung in Django oder Flask befasst haben, werden Sie die Eleganz der Starlette-Implementierung verstehen). Mal sehen, wie es implementiert ist. Der Code ist sehr einfach, etwa 60 Zeilen, aber mit vielen Kommentaren:

class BaseHTTPMiddleware:
    def __init__(self, app: ASGIApp, dispatch: DispatchFunction = None) -> None:
        # Die ASGI-App der nächsten Ebene zuweisen
        self.app = app
        # Wenn der Benutzer dispatch übergibt, wird die vom Benutzer übergebene Funktion verwendet, Andernfalls wird ihr eigenes dispatch verwendet
        # Im Allgemeinen erben Benutzer von BaseHTTPMiddleware und überschreiben die dispatch-Methode
        self.dispatch_func = self.dispatch if dispatch is None else dispatch

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        """
        Eine Funktion mit der ASGI-Standardfunktionssignatur, die darstellt, dass ASGI-Anfragen von hier eintreten.
        """
        if scope["type"]!= "http":
            # Wenn der Typ nicht http ist, wird die Middleware nicht übergeben (d. h. WebSocket wird nicht unterstützt)
            # Um WebSocket zu unterstützen, ist es sehr schwierig, die Middleware auf diese Weise zu implementieren. Als ich das Rap-Framework implementierte, opferte ich einige Funktionen, um die Middleware-Verarbeitung für WebSocket-ähnlichen Traffic zu erreichen.
            await self.app(scope, receive, send)
            return

        # Ein Request -Objekt aus scope erstellen
        request = Request(scope, receive=receive)
        # In die Dispatch -Logik eintreten, d. h. die eigene Verarbeitungslogik des Benutzers
        # Die von dieser Logik erhaltene Antwort wird tatsächlich durch die call_next -Funktion generiert, und die dispatch -Funktion spielt nur die Rolle der Weitergabe.
        response = await self.dispatch_func(request, self.call_next)
        # Daten gemäß der generierten Antwort an die obere Ebene zurückgeben
        await response(scope, receive, send)

    async def call_next(self, request: Request) -> Response:
        loop = asyncio.get_event_loop()
        # Die nächste Nachrichten-Ebene über das Queue-Produktions- und Verbrauchsmodell erhalten
        queue: "asyncio.Queue[typing.Optional[Message]]" = asyncio.Queue()

        scope = request.scope
        # Die send-Funktion von uvicorn wird über das request.receive-Objekt übergeben
        # Das hier verwendete receive-Objekt ist immer noch das von uvicorn initialisierte receive-Objekt
        receive = request.receive
        send = queue.put

        async def coro() -> None:
            try:
                await self.app(scope, receive, send)
            finally:
                # Diese Put-Operation stellt sicher, dass die Get-Seite nicht blockiert wird
                await queue.put(None)

        # Die nächste ASGI APP in einem anderen Coroutine über loop.create_task ausführen
        task = loop.create_task(coro())
        # Auf die Rückgabe der nächsten ASGI APP warten
        message = await queue.get()
        if message is None:
            # Wenn der erhaltene Wert leer ist, bedeutet dies, dass die nächste ASGI APP keine Antwort zurückgegeben hat und möglicherweise ein Fehler aufgetreten ist.
            # Durch den Aufruf von task.result() wird der Fehler der Coroutine ausgelöst, wenn die Coroutine eine Ausnahme aufweist.
            task.result()
            # Wenn keine Ausnahme ausgelöst wird, kann dies auf Benutzerfehler zurückzuführen sein, z. B. die Rückgabe einer leeren Antwort.
            # Zu diesem Zeitpunkt ist es unmöglich, eine Antwort an den Client zurückzugeben, daher muss eine Ausnahme erstellt werden, um die nachfolgende Generierung einer 500-Antwort zu erleichtern.
            raise RuntimeError("Keine Antwort zurückgegeben.")

        # Wenn ASGI die Antwort verarbeitet, geschieht dies in mehreren Schritten. Normalerweise ist der obige queue.get der erste Schritt, um die Antwort zu erhalten.
        assert message["type"] == "http.response.start"

        async def body_stream() -> typing.AsyncGenerator[bytes, None]:
            # Andere Verarbeitungen werden an die body_stream -Funktion übergeben
            # Diese Methode gibt einfach den Datenstrom zurück
            while True:
                message = await queue.get()
                if message is None:
                    break
                assert message["type"] == "http.response.body"
                yield message.get("body", b"")
            task.result()

        # Die body_stream -Funktion in die Response-Methode einfügen
        # Die Antwort selbst ist auch eine Klasse, die einer ASGI APP ähnelt.

2.3. ServerErrorMiddleware

ServerErrorMiddleware ist der ExceptionMiddleware sehr ähnlich (daher wird dieser Teil nicht weiter ausgeführt). Ihre Gesamtlogik ist größtenteils dieselbe. Während ExceptionMiddleware jedoch für die Erfassung und Behandlung von Routing-Ausnahmen zuständig ist, dient ServerErrorMiddleware hauptsächlich als Fallback-Maßnahme, um sicherzustellen, dass immer eine legitime HTTP-Antwort zurückgegeben wird.

Die indirekte Aufruffunktion von ServerErrorMiddleware ist dieselbe wie die von ExceptionMiddleware. Nur wenn der registrierte HTTP-Statuscode 500 ist, wird der Callback in ServerErrorMiddleware registriert:

@app.exception_handlers(500)
def not_found_route() -> None: pass

ServerErrorMiddleware befindet sich auf der obersten Ebene der ASGI APP. Sie übernimmt die Aufgabe der Behandlung von Fallback-Ausnahmen. Was sie tun muss, ist einfach: Wenn während der Verarbeitung der ASGI APP der nächsten Ebene ein Fehler auftritt, greift sie auf die Fallback-Logik zurück:

1. Wenn Debugging aktiviert ist, wird die Debug-Seite zurückgegeben.
1. Wenn ein registrierter Callback vorhanden ist, wird der registrierte Callback ausgeführt.
1. Wenn keine der obigen Optionen zutrifft, wird eine 500-Antwort zurückgegeben.

3. Route

In Starlette sind Routen in zwei Teile unterteilt. Ein Teil, den ich den Router der echten App nenne, befindet sich unterhalb der Middleware. Er ist hauptsächlich für fast alles in Starlette außer der Middleware zuständig, einschließlich Routing-Suche und -Abgleich, Behandlung von Anwendungsstart und -stopp usw. Der andere Teil besteht aus den Routen, die beim Router registriert sind.

3.1. Router

Der Router ist unkompliziert. Seine Hauptaufgabe ist das Laden und Abgleichen von Routen. Hier sind der Quellcode und die Kommentare, ausgenommen des Teils zum Laden von Routen:

class Router:
    def __init__(
        self,
        routes: typing.Sequence[BaseRoute] = None,
        redirect_slashes: bool = True,
        default: ASGIApp = None,
        on_startup: typing.Sequence[typing.Callable] = None,
        on_shutdown: typing.Sequence[typing.Callable] = None,
        lifespan: typing.Callable[[typing.Any], typing.AsyncGenerator] = None,
    ) -> None:
        # Informationen aus Starlette-Initialisierung laden
        self.routes = [] if routes is None else list(routes)
        self.redirect_slashes = redirect_slashes
        self.default = self.not_found if default is None else default
        self.on_startup = [] if on_startup is None else list(on_startup)
        self.on_shutdown = [] if on_shutdown is None else list(on_shutdown)

        async def default_lifespan(app: typing.Any) -> typing.AsyncGenerator:
            await self.startup()
            yield
            await self.shutdown()

        # Wenn die initialisierte Lebensdauer leer ist, konvertieren Sie on_startup und on_shutdown in Lebensdauer
        self.lifespan_context = default_lifespan if lifespan is None else lifespan

    async def not_found(self, scope: Scope, receive: Receive, send: Send) -> None:
        """Logik, die ausgeführt wird, wenn keine Route übereinstimmt"""
        if scope["type"] == "websocket":
            # WebSocket-Übereinstimmung fehlgeschlagen
            websocket_close = WebSocketClose()
            await websocket_close(scope, receive, send)
            return

        # Wenn wir uns innerhalb einer Starlette-Anwendung befinden, lösen Sie eine Ausnahme aus,
        # damit der konfigurierbare Ausnahmehandler die Antwort zurückgeben kann. Für reine ASGI-Apps geben Sie einfach die Antwort zurück.
        if "app" in scope:
            # In der __call__ -Methode von starlette.applications können wir sehen, dass Starlette sich selbst in scope speichert.
            # Nach dem Auslösen einer Ausnahme hier kann sie von ServerErrorMiddleware abgefangen werden.
            raise HTTPException(status_code=404)
        else:
            # Für Aufrufe, die nicht von Starlette stammen, direkt einen Fehler zurückgeben
            response = PlainTextResponse("Not Found", status_code=404)
        await response(scope, receive, send)

    async def lifespan(self, scope: Scope, receive: Receive, send: Send) -> None:
        """
        Verarbeitet ASGI-Lebensdauer-Nachrichten, die uns die Verwaltung von Anwendungsstart und -stopp-Ereignissen ermöglichen.
        """
        # Logik zur Lebensdauer-Ausführung. Bei der Ausführung kommuniziert Starlette mit dem ASGI-Server. Aber derzeit gibt es möglicherweise noch zu entwickelnde Funktionen in diesem Code.
        first = True
        app = scope.get("app")
        await receive()
        try:
            if inspect.isasyncgenfunction(self.lifespan_context):
                async for item in self.lifespan_context(app):
                    assert first, "Lifespan context yielded multiple times."
                    first = False
                    await send({"type": "lifespan.startup.complete"})
                    await receive()
            else:
                for item in self.lifespan_context(app):  # type: ignore
                    assert first, "Lifespan context yielded multiple times."
                    first = False
                    await send({"type": "lifespan.startup.complete"})
                    await receive()
        except BaseException:
            if first:
                exc_text = traceback.format_exc()
                await send({"type": "lifespan.startup.failed", "message": exc_text})
            raise
        else:
            await send({"type": "lifespan.shutdown.complete"})

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        """
        Der Haupteinstiegspunkt für die Router -Klasse.
        """
        # Die Hauptfunktion für den Abgleich und die Ausführung von Routen
        # Derzeit werden nur die Typen http, websocket und lifespan unterstützt
        assert scope["type"] in ("http", "websocket", "lifespan")

        # Router in scope initialisieren
        if "router" not in scope:
            scope["router"] = self

        if scope["type"] == "lifespan":
            # Lebensdauer-Logik ausführen
            await self.lifespan(scope, receive, send)
            return

        partial = None

        # Routenabgleich durchführen
        for route in self.routes:
            match, child_scope = route.matches(scope)
            if match == Match.FULL:
                # Wenn es eine vollständige Übereinstimmung ist (sowohl URL als auch Methode stimmen überein)
                # Dann normale Routenverarbeitung durchführen
                scope.update(child_scope)
                await route.handle(scope, receive, send)
                return
            elif match == Match.PARTIAL and partial is None:
                # Wenn es eine teilweise Übereinstimmung ist (URL stimmt überein, Methode nicht)
                # Dann den Wert beibehalten und den Abgleich fortsetzen
                partial = route
                partial_scope = child_scope

        if partial is not None:
            # Wenn es eine Route mit teilweisem Abgleich gibt, wird die Ausführung ebenfalls fortgesetzt, aber die Route gibt einen HTTP-Methodenfehler zurück
            scope.update(partial_scope)
            await partial.handle(scope, receive, send)
            return

        if scope["type"] == "http" and self.redirect_slashes and scope["path"]!= "/":
            # Nicht übereinstimmende Situation, Umleitung beurteilen
            redirect_scope = dict(scope)
            if scope["path"].endswith("/"):
                redirect_scope["path"] = redirect_scope["path"].rstrip("/")
            else:
                redirect_scope["path"] = redirect_scope["path"] + "/"

            for route in self.routes:
                match, child_scope = route.matches(redirect_scope)
                if match!= Match.NONE:
                    # Erneut abgleichen. Wenn das Ergebnis nicht leer ist, eine Umleitungsantwort senden
                    redirect_url = URL(scope=redirect_scope)
                    response = RedirectResponse(url=str(redirect_url))
                    await response(scope, receive, send)
                    return

Wenn keiner der obigen Prozesse zutrifft, bedeutet dies, dass keine Route gefunden wurde. Zu diesem Zeitpunkt wird die Standardroute ausgeführt, und die Standard-Standardroute ist 404 Not Found.

        await self.default(scope, receive, send)

Es ist ersichtlich, dass der Router-Code recht einfach ist. Der meiste Code konzentriert sich in der call-Methode. Allerdings gibt es mehrere Durchläufe, um Routen abzufragen, und jede Route führt einen regulären Ausdruck aus, um zu beurteilen, ob er übereinstimmt. Manche Leute denken vielleicht, dass diese Ausführungsgeschwindigkeit langsam ist. Das dachte ich früher auch, und dann habe ich einen Routenbaum implementiert, um ihn zu ersetzen (siehe route_trie.py für Details). Aber nach einem Leistungstest stellte ich fest, dass die Schleifenabgleichleistung besser ist als die des Routenbaums, wenn die Anzahl der Routen 50 nicht überschreitet. Wenn die Zahl 100 nicht überschreitet, sind die beiden gleichwertig. Und unter normalen Umständen überschreitet die Anzahl der von uns angegebenen Routen 100 nicht. Daher besteht kein Grund zur Sorge über die Abgleichleistung dieses Teils der Routen. Wenn Sie sich immer noch Sorgen machen, können Sie Mount verwenden, um die Routen zu gruppieren und die Abgleichgeschwindigkeit zu verringern.

3.2. Andere Routen

Mount erbt von BaseRoute, ebenso wie andere Routen wie HostRoute, WebSocketRoute. Sie bieten ähnliche Methoden mit nur geringfügigen Unterschieden in der Implementierung (hauptsächlich bei der Initialisierung, dem Routenabgleich und der Rückwärtssuche). Betrachten wir zunächst BaseRoute:

class BaseRoute:
    def matches(self, scope: Scope) -> typing.Tuple[Match, Scope]:
        # Eine standardmäßige Abgleichfunktionssignatur. Jede Route sollte ein (Match, Scope) Tupel zurückgeben.
        # Match hat 3 Typen:
        #   NONE: Keine Übereinstimmung.
        #   PARTIAL: Teilweise Übereinstimmung (URL stimmt überein, Methode nicht).
        #   FULL: Vollständige Übereinstimmung (sowohl URL als auch Methode stimmen überein).
        # Scope gibt im Grunde das folgende Format zurück, aber Mount gibt mehr Inhalte zurück:
        #   {"endpoint": self.endpoint, "path_params": path_params}
        raise NotImplementedError()  # pragma: no cover

    def url_path_for(self, name: str, **path_params: str) -> URLPath:
        # Generiert eine Rückwärtssuche nach dem Namen
        raise NotImplementedError()  # pragma: no cover

    async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
        # Die Funktion, die nach dem Abgleich durch den Router aufgerufen werden kann
        raise NotImplementedError()  # pragma: no cover

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        """
        Eine Route kann isoliert als eigenständige ASGI-App verwendet werden.
        Dies ist ein etwas erzwungener Fall, da sie fast immer innerhalb eines Routers verwendet wird,
        könnte aber für einige Tools und minimale Apps nützlich sein.
        """
        # Wenn die Route als eigenständige ASGI APP aufgerufen wird, wird sie abgeglichen und antwortet selbst
        match, child_scope = self.matches(scope)
        if match == Match.NONE:
            if scope["type"] == "http":
                response = PlainTextResponse("Not Found", status_code=404)
                await response(scope, receive, send)
            elif scope["type"] == "websocket":
                websocket_close = WebSocketClose()
                await websocket_close(scope, receive, send)
            return

        scope.update(child_scope)
        await self.handle(scope, receive, send)

Es ist ersichtlich, dass BaseRoute nicht viele Funktionen bereitstellt und andere Routen darauf aufbauen:

Route: Eine Standard-HTTP-Route. Sie ist für den Routenabgleich über die HTTP-URL und die HTTP-Methode zuständig und stellt dann eine Methode zum Aufrufen der HTTP-Route bereit.
WebSocketRoute: Eine Standard-WebSocket-Route. Sie gleicht Routen anhand der HTTP-URL ab und generiert dann eine Sitzung über das WebSocket in starlette.websocket und gibt sie an die entsprechende Funktion weiter.
Mount: Eine verschachtelte Kapselung von Routen. Ihre Abgleichmethode ist ein Präfixabgleich der URL, und sie leitet Anfragen an die ASGI APP der nächsten Ebene weiter, die den Regeln entspricht. Wenn ihre ASGI APP der nächsten Ebene ein Router ist, kann die Aufrufkette wie folgt aussehen: Router->Mount->Router->Mount->Router->Route. Durch die Verwendung von Mount können Routen gruppiert und die Abgleichgeschwindigkeit erhöht werden. Es wird empfohlen. Außerdem kann sie Anfragen an andere ASGI APPs verteilen, z. B. Starlette->ASGI Middleware->Mount->Andere Starlette->...
Host: Sie verteilt Anfragen anhand des Host in der Benutzeranfrage an die entsprechende ASGI APP, die Route, Mount, Middleware usw. sein kann.

4. Andere Komponenten

Wie oben zu sehen ist, sind die meisten Komponenten in Starlette als ASGI APPs konzipiert, was eine hohe Kompatibilität aufweist. Obwohl dies geringfügige Leistungseinbußen bedeutet, ist die Kompatibilität extrem stark. Andere Komponenten sind ebenfalls mehr oder weniger als ASGI APPs konzipiert. Bevor wir andere Komponenten vorstellen, betrachten wir zunächst die Gesamtprojektstruktur von Starlette:

├── middleware                       # Middleware
├── applications.py                  # Startup Application
├── authentication.py                # Authentifizierungsbezogen
├── background.py                    # Kapselt Hintergrundaufgaben, die nach der Rückgabe der Antwort ausgeführt werden
├── concurrency.py                   # Einige kleine asyncio-bezogene Kapselungen. In der neuen Version wird stattdessen die anyio-Bibliothek direkt verwendet.
├── config.py                        # Konfiguration
├── convertors.py                    # Einige Typkonvertierungsmethoden
├── datastructures.py                # Einige Datenstrukturen, z. B. Url, Header, Form, QueryParam, State usw.
├── endpoints.py                     # Routen, die CBV und eine etwas fortgeschrittenere WebSocket-Kapselung unterstützen
├── exceptions.py                    # Ausnahmebehandlung
├── formparsers.py                   # Parsen von Form, Datei usw.
├── graphql.py                       # Verantwortlich für die Verarbeitung von GraphQL
├── __init__.py
├── py.typed                         # Typ-Hints benötigt von Starlette
├── requests.py                      # Anfragen, damit Benutzer Daten abrufen können
├── responses.py                     # Antworten, verantwortlich für die Initialisierung von Headern und Cookies, das Generieren von Antwortdaten gemäß verschiedenen Response -Klassen und das anschließende Aufrufen der ASGI-Aufrufschnittstelle. Diese Schnittstelle sendet das ASGI-Protokoll an den Uvicorn-Dienst. Nach dem Senden werden Hintergrundaufgaben ausgeführt, bis sie abgeschlossen sind.
├── routing.py                       # Routing
├── schemas.py                       # OpenApi-bezogene Schemas
├── staticfiles.py                   # Statische Dateien
├── status.py                        # HTTP-Statuscodes
├── templating.py                    # Template-Antworten basierend auf Jinja
├── testclient.py                    # Test-Client
├── types.py                         # Typen
└── websockets.py                    # WebSocket

Es gibt viele Dateien oben, und einige einfachere werden übersprungen.

4.1. Request

Die Request-Klasse ist sehr einfach. Sie erbt von HttpConnection. Diese Klasse parst hauptsächlich die vom ASGI-Protokoll übergebene Scope, um Informationen wie die URL und die Methode zu extrahieren. Und die Request-Klasse fügt die Funktionen zum Lesen von Anfragedaten und zum Zurückgeben von Daten hinzu (HTTP 1.1 unterstützt das Pushen von Daten vom Server zum Client). Unter ihnen hängt das Lesen von Daten von einer Kernfunktion ab - stream. Sein Quellcode sieht wie folgt aus:

async def stream(self) -> typing.AsyncGenerator[bytes, None]:
    # Wenn bereits gelesen, Daten aus dem Cache abrufen
    if hasattr(self, "_body"):
        yield self._body
        yield b""
        return

    if self._stream_consumed:
        raise RuntimeError("Stream consumed")

    self._stream_consumed = True
    while True:
        # Kontinuierlich Daten aus der Empfangsschleife des ASGI-Containers abrufen
        message = await self._receive()
        if message["type"] == "http.request":
            body = message.get("body", b"")
            if body:
                # Daten zurückgeben, wenn sie nicht leer sind
                yield body
            if not message.get("more_body", False):
                # Das bedeutet, dass alle Body-Daten erhalten wurden
                break
        elif message["type"] == "http.disconnect":
            # Das bedeutet, dass die Verbindung zum Client geschlossen wurde
            self._is_disconnected = true
            # Eine Ausnahme auslösen. Benutzer, die `await request.body()` oder `await request.json()` aufrufen, lösen eine Ausnahme aus.
            raise ClientDisconnect()
    # Leere Bytes als Markierung für das Ende zurückgeben
    yield b""

Diese Implementierung ist einfach, aber es gibt einen kleinen Fehler. Wer mit Nginx oder anderen Webdiensten vertraut ist, weiß, dass allgemeine Zwischenserver keine Body-Daten verarbeiten, sondern sie nur weiterleiten. Dasselbe gilt für ASGI. Nach der Verarbeitung von URL und Header beginnt Uvicorn mit dem Aufruf der ASGI APP und leitet die Objekte send und receive nach unten weiter. Diese beiden Objekte durchlaufen mehrere ASGI APPs und erreichen die Routen-ASGI APP, die Benutzer in Funktionen verwenden können. Die receive-Objekte, die von der Request empfangen werden, werden von Uvicorn generiert. Und die Datenquelle von receive kommt aus einer asyncio.Queue-Schlange. Aus der Analyse der Middleware ergibt sich, dass jede ASGI APP ein Request-Objekt basierend auf scope und receive generiert, was bedeutet, dass die Request-Objekte jeder Schicht von ASGI APPs inkonsistent sind. Wenn die Middleware das Request-Objekt zum Lesen des Bodies aufruft, verbraucht sie die Daten in der Warteschlange im Voraus über receive, was dazu führt, dass nachfolgende ASGI APPs keine Body-Daten über das Request-Objekt lesen können. Der Beispielcode für dieses Problem lautet wie folgt:

import asyncio
from starlette.applications import Starlette
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from starlette.requests import Request
from starlette.responses import JSONResponse, Response

app: Starlette = Starlette()

class DemoMiddleware(BaseHTTPMiddleware):
    async def dispatch(
        self,
        request: Request,
        call_next: RequestResponseEndpoint,
    ) -> Response:
        print(request, await request.body())
        return await call_next(request)

app.add_middleware(DemoMiddleware)

@app.route("/")
async def demo(request: Request) -> JSONResponse:
    try:
        await asyncio.wait_for(request.body(), 1)
        return JSONResponse({"result": True})
    except asyncio.TimeoutError:
        return JSONResponse({"result": False})

if __name__ == "__main__":
    import uvicorn  # type: ignore
    uvicorn.run(app)

Führen Sie es aus und überprüfen Sie das Ergebnis der Anfrage:

-> curl http://127.0.0.1:8000
{"result":false}

Wie oben gezeigt, ist das Ergebnis false, was bedeutet, dass die Ausführung von request.body abgelaufen ist, da die receive-Warteschlange bereits leer ist und keine Daten abgerufen werden können. Ohne ein Zeitlimit würde diese Anfrage blockieren.

Um dieses Problem zu lösen, schauen wir uns zunächst an, wie Request den Body erhält. Da der Benutzer den Body mehrmals abrufen kann und die Daten gleich bleiben, ist die Implementierungsidee, die Daten nach dem Abruf zu cachen. Wir können dieser Idee folgen. Da die Daten über receive abgerufen werden, können wir nach dem Lesen der Daten eine receive-Funktion konstruieren. Diese Funktion gibt Daten zurück, die dem ASGI-Kommunikationsprotokoll ähneln, und enthält vollständige Body-Daten (passend für Request.stream zum Abrufen des Bodies). Der Code sieht wie folgt aus:

async def proxy_get_body(request: Request) -> bytes:
    async def receive() -> Message:
        return {"type": "http.request", "body": body}

    body = await request.body()
    request._receive = receive
    return body

Anschließend kann jede Ebene der ASGI APP diese Funktion aufrufen, um die Body-Daten zu erhalten, ohne die Fähigkeit nachfolgender ASGI APPs zu beeinträchtigen, die Body-Daten zu erhalten, sofern sie die Body-Daten abrufen muss.

5. Zusammenfassung

Wir haben inzwischen mehrere wichtige funktionale Codes von Starlette analysiert. Starlette ist eine ausgezeichnete Bibliothek mit einem tollen Designkonzept. Es wird empfohlen, den Starlette-Quellcode selbst zu lesen, was für das Schreiben zukünftiger eigener Frameworks hilfreich sein wird.