FastAPI は本質的に Starlette の API ラッパーです FastAPI を完全に理解するには、まず Starlette を理解する必要があります。

1. ASGI プロトコル

Uvicorn は共通のインターフェースを通じて ASGI アプリケーションと対話します。アプリケーションは、次のコードを実装することで Uvicorn を介して情報を送受信できます。

async def app(scope, receive, send):
    # 最もシンプルな ASGI アプリケーション
    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 サービス
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=5000, log_level="info")

2. Starlette

Uvicorn で Starlette を起動するには、次のコードを使用します。

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)

このコードは、Starlette を初期化し、ルート、例外ハンドラー、イベント、ミドルウェアを登録してから、uvicorn.run に渡します。uvicorn.run メソッドは、Starlette の call メソッドを呼び出すことでリクエストデータを送信します。

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: デバッグ機能の有効化を決定します。
        :param route: HTTP および WebSocket サービスを提供するルートのリスト。
        :param middleware: 各リクエストに適用されるミドルウェアのリスト。
        :param exception_handler: 例外コ​​ールバックを格納する辞書。HTTP ステータス コードをキー、コールバック関数を値とします。
        :on_startup: 起動時に呼び出されるコールバック関数。
        :on_shutdown: シャットダウン時に呼び出されるコールバック関数。
        :lifespan: ASGI でのライフスパン関数。
        """

        # lifespan が渡された場合、on_startup と on_shutdown は渡せません
        # Starlette は本質的に on_start_up と on_shutdown を lifespan に変換して Uvicorn が呼び出せるようにします
        assert lifespan is None or (
            on_startup is None and on_shutdown is None
        ), "'lifespan' または 'on_startup'/'on_shutdown' のいずれかを使用し、両方を同時に使用しないでください。"

        # 変数の初期化
        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)
        # ミドルウェアの構築
        self.middleware_stack = self.build_middleware_stack()

コードからわかるように、初期化はすでにほとんどの要件を満たしています。ただし、ミドルウェアを構築するための機能があり、さらに分析が必要です。

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

        # 例外処理コ​​ールバックを解析し、error_handler および exception_handlers に格納します
        # HTTP ステータス コード 500 のみが error_handler に格納されます
        for key, value in self.exception_handlers.items():
            if key in (500, Exception):
                error_handler = value
            else:
                exception_handlers[key] = value

        # 異なるタイプのミドルウェアの順序付け
        # 最初のレイヤーは ServerErrorMiddleware で、例外が見つかった場合にエラー スタックを印刷したり、デバッグ モードでエラー ページを表示してデバッグしやすくします。
        # 2 番目のレイヤーはユーザーミドルウェアレイヤーで、ユーザーによって登録されたすべてのミドルウェアが格納されます。
        # 3 番目のレイヤーは ExceptionMiddleware で、例外処理レイヤーであり、ルート実行中に発生したすべての例外を処理します。
        middleware = (
            [Middleware(ServerErrorMiddleware, handler=error_handler, debug=debug)]
            + self.user_middleware
            + [
                Middleware(
                    ExceptionMiddleware, handlers=exception_handlers, debug=debug
                )
            ]
        )

        # 最後に、ミドルウェアをアプリにロードします
        app = self.router
        for cls, options in reversed(middleware):
            # cls はミドルウェアクラス自体であり、options は私たちが渡すパラメータです
            # ミドルウェア自体も ASGI APP であり、ミドルウェアのロードは、入れ子になった ASGI APP のように、マトリョーシカ人形のように見えます。
            app = cls(app=app, **options)

        # ミドルウェアは入れ子方式でロードされ、`call_next` を介して呼び出されて上位の ASGI APP を呼び出すため、逆順の方法が使用されます。
        return app

ミドルウェアを構築した後、初期化が完了し、uvicorn.run メソッドが call メソッドを呼び出します。

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

このメソッドはシンプルです。scope を介してリクエストフローにアプリを設定し、後続の呼び出しのために、middleware_stack を呼び出してリクエスト処理を開始します。このメソッドとミドルウェアの初期化から、Starlette のミドルウェアも ASGI APP であり（ルートも呼び出しスタックの最下位で ASGI APP であることがわかります）、同時に Starlette は例外処理をミドルウェアに委譲しています。これは他の Web アプリケーションフレームワークではまれに見られます。Starlette は、各コンポーネントが可能な限り ASGI APP になるように設計されていることがわかります。

2. ミドルウェア

前述のように、Starlette ではミドルウェアは ASGI APP です。したがって、Starlette のすべてのミドルウェアは、次の形式を満たすクラスである必要があります。

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

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

starlette.middleware には、この要件を満たす多くのミドルウェア実装があります。ただし、この章ではすべてのミドルウェアをカバーせず、ルートから最も遠いものから最も近いものまで、代表的なものをいくつか選択して分析します。

2.1. 例外処理ミドルウェア - ExceptionMiddleware

最初が ExceptionMiddleware です。ユーザーはこのミドルウェア（starlette.middleware に配置されていない）を直接操作しませんが、次のメソッドを通じて間接的に操作します。

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

ユーザーがこのメソッドを使用すると、Starlette はコールバック関数を対応する辞書にハングさせます。HTTP ステータス コードをキー、コールバック関数を値とします。 ExceptionMiddleware がルート リクエスト処理で例外を検出すると、例外応答の HTTP ステータス コードを通じて対応するコールバック関数を見つけ、リクエストと例外をユーザーがマウントしたコールバック関数に渡し、最終的にユーザーのコールバック関数の結果を前の ASGI APP に返します。 さらに、ExceptionMiddleware は例外登録もサポートしています。ルートによってスローされた例外が登録された例外と一致すると、その例外登録に対応するコールバック関数が呼び出されます。 このクラスのソースコードとコメントは次のとおりです。

class ExceptionMiddleware:
    def __init__(
        self,
        app: ASGIApp, handlers: dict = None, debug: bool = False
    ) -> None:
        self.app = app
        self.debug = debug  # TODO: デバッグが設定されている場合、404 ケースを処理する必要があります。
        # Starlette は HTTP ステータス コードと例外型をサポートしています
        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:
        # Starlette アプリメソッドを通じてユーザーがマウントした例外コ​​ールバックは、最終的にこのメソッドを通じてクラスの _status_handlers または _exception_handler にマウントされます。
        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]:
        # 登録された例外に関連するコールバック関数を検索し、mro を介して例外に対応するコールバック関数を見つけます
        # 
        # ユーザーは基底クラスをマウントすることがあり、マウントされた例外の subsequent subclasses も基底クラスに登録されたコールバックを呼び出します。
        # 例：ユーザーが基底クラスを登録し、その後、ユーザー例外とシステム例外の 2 つの例外があり、両方ともこの基底クラスを継承しています。
        # 後で関数がユーザー例外またはシステム例外をスローすると、基底クラスに登録された対応するコールバックが実行されます。
        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:
        # おなじみの ASGI 呼び出しメソッド
        if scope["type"]!= "http":
            # WebSocket リクエストはサポートされていません
            await self.app(scope, receive, send)
            return

        # 同じ応答で複数の例外が発生するのを防ぎます
        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:
            # 次の ASGI APP を呼び出します
            await self.app(scope, receive, sender)
        except Exception as exc:
            handler = None

            if isinstance(exc, HTTPException):
                # HTTPException の場合、登録された HTTP コールバック辞書で検索します
                handler = self._status_handlers.get(exc.status_code)

            if handler is None:
                # 通常の例外の場合、例外コールバック辞書で検索します
                handler = self._lookup_exception_handler(exc)

            if handler is None:
                # 対応する例外が見つからない場合、それを上位にスローします
                raise exc from None

            # 1 つの応答につき 1 つの例外のみを処理します
            if response_started:
                msg = "Caught handled exception, but response already started."
                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)
            # コールバック関数によって生成された応答でリクエストを処理します
            await response(scope, receive, sender)

2.2. ユーザーミドルウェア

次はユーザーミドルウェアで、最もよく接触するミドルウェアです。starlette.middleware を使用する場合、通常は BaseHTTPMiddleware というミドルウェアを継承し、次のコードに基づいて拡張します。

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

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

リクエスト前の前処理を実行したい場合は、before ブロックに関連コードを記述します。リクエスト後の後処理を実行したい場合は、after ブロックにコードを記述します。使い方は非常に簡単で、同じスコープにあります。これは、このメソッドの変数でコンテキストや動的変数を通じて伝搬する必要がないことを意味します（Django や Flask のミドルウェア実装に触れたことがあるなら、Starlette の実装の優雅さがわかるでしょう）。 実装を見てみましょう。コードは非常にシンプルで、約 60 行ですが、コメントはたくさんあります。

class BaseHTTPMiddleware:
    def __init__(self, app: ASGIApp, dispatch: DispatchFunction = None) -> None:
        # 次のレベルの ASGI アプリを割り当てます
        self.app = app
        # ユーザーが dispatch を渡した場合、ユーザーが渡した関数を使用します。それ以外の場合は、独自の dispatch を使用します。
        # 一般に、ユーザーは BaseHTTPMiddleware を継承し、dispatch メソッドを書き換えます
        self.dispatch_func = self.dispatch if dispatch is None else dispatch

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        """
        ASGI 標準関数シグネチャを持つ関数で、ASGI リクエストがここから入力されることを表します。
        """
        if scope["type"]!= "http":
            # タイプが http でない場合、ミドルウェアは渡されません（つまり、WebSocket はサポートされていません）。
            # WebSocket をサポートするには、この方法でミドルウェアを実装するのは非常に困難です。Rap フレームワークを実装したとき、WebSocket ライクなトラフィックのミドルウェア処理を実現するためにいくつかの関数を犠牲にしました。
            await self.app(scope, receive, send)
            return

        # scope からリクエストオブジェクトを生成します
        request = Request(scope, receive=receive)
        # dispatch ロジック、つまりユーザーの処理ロジックに入ります
        # このロジックから取得した応答は、実際には call_next 関数によって生成され、dispatch 関数は渡しの役割のみを果たします。
        response = await self.dispatch_func(request, self.call_next)
        # 生成された応答に従って、上位レイヤーに応答データを返します
        await response(scope, receive, send)

    async def call_next(self, request: Request) -> Response:
        loop = asyncio.get_event_loop()
        # キューの生産と消費モデルを通じて次のレベルのメッセージを取得します
        queue: "asyncio.Queue[typing.Optional[Message]]" = asyncio.Queue()

        scope = request.scope
        # request.receive オブジェクトを通じて uvicorn の receive オブジェクトを渡します
        # ここで使用される receive オブジェクトは、uvicorn によって初期化された receive オブジェクトです
        receive = request.receive
        send = queue.put

        async def coro() -> None:
            try:
                await self.app(scope, receive, send)
            finally:
                # この put 操作は、get 側がブロックされないことを保証します
                await queue.put(None)

        # loop.create_task を通じて、別のコルーチンで次の ASGI APP を実行します
        task = loop.create_task(coro())
        # 次の ASGI APP の返信を待ちます
        message = await queue.get()
        if message is None:
            # 取得した値が空の場合、次の ASGI APP が応答を返さなかったことを意味し、エラーが発生した可能性があります。
            # task.result() を呼び出すことで、コルーチンに例外がある場合、コルーチンのエラーがスローされます。
            task.result()
            # 例外がスローされない場合、ユーザーエラー（空の応答を返すなど）が原因である可能性があります。
            # この時点でクライアントに応答を返すことはできないため、例外を作成して subsequent 500 応答の生成を容易にする必要があります。
            raise RuntimeError("No response returned.")

        # ASGI が応答を処理するとき、それは複数のステップで行われます。通常、上記の queue.get は応答を取得する最初のステップです。
        assert message["type"] == "http.response.start"

        async def body_stream() -> typing.AsyncGenerator[bytes, None]:
            # その他の処理は body_stream 関数に委譲されます
            # このメソッドは単にデータ ストリームを返し続けます
            while True:
                message = await queue.get()
                if message is None:
                    break
                assert message["type"] == "http.response.body"
                yield message.get("body", b"")
            task.result()

        # body_stream 関数を Response メソッドに配置します
        # 応答自体も ASGI APP に似たクラスです。
        ...

2.3. ServerErrorMiddleware

ServerErrorMiddleware は ExceptionMiddleware に非常に似ています（そのため、この部分はさらに詳しく説明しません）。全体的なロジックはほとんど同じです。しかし、ExceptionMiddleware がルーティング例外のキャッチと処理を担当するのに対し、ServerErrorMiddleware は主に、正当な HTTP 応答が常に返されることを保証するためのフォールバック対策として機能します。

ServerErrorMiddleware の間接的な呼び出し関数は ExceptionMiddleware と同じです。ただし、登録された HTTP ステータス コードが 500 の場合にのみ、ServerErrorMiddleware にコールバックが登録されます。

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

ServerErrorMiddleware は ASGI APP の最上位レベルにあります。フォールバック例外の処理タスクを引き受けます。それが行う必要があることはシンプルです。次のレベルの ASGI APP の処理中に例外が発生した場合、フォールバック ロジックに入ります。

• 1. デバッグが有効な場合、デバッグ ページを返します。
• 2. 登録されたコールバックがある場合、登録されたコールバックを実行します。
• 3. 上記のいずれでもない場合、500 応答を返します。

3. ルート

Starlette では、ルートは 2 つの部分に分かれています。1 つは、私がリアルアプリのルーターと呼んでいるもので、ミドルウェアの下のレベルにあります。ルートの検索と照合、アプリケーションの起動とシャットダウン処理など、ミドルウェア以外のほとんどすべてを担当します。もう 1 つは、ルーターに登録されたルートで構成されます。

3.1. ルーター

ルーターは簡単です。主な責任は、ルートのロードと照合です。ルートをロードする部分を除いたソースコードとコメントを次に示します。

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:
        # Starlette の初期化からの情報をロードします
        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()

        # 初期化された lifespan が空の場合、on_startup と on_shutdown を lifespan に変換します
        self.lifespan_context = default_lifespan if lifespan is None else lifespan

    async def not_found(self, scope: Scope, receive: Receive, send: Send) -> None:
        """ルートが一致しない場合に実行されるロジック"""
        if scope["type"] == "websocket":
            # WebSocket の一致に失敗しました
            websocket_close = WebSocketClose()
            await websocket_close(scope, receive, send)
            return

        # starlette アプリケーション内で実行されている場合、例外をスローして、設定可能な例外ハンドラーが応答の返却を処理できるようにします。プレーン ASGI アプリの場合は、応答を返します。
        if "app" in scope:
            # starlette.applications の __call__ メソッドで、starlette が自身を scope に格納していることがわかります。
            # ここで例外をスローすると、ServerErrorMiddleware によってキャプチャされる可能性があります。
            raise HTTPException(status_code=404)
        else:
            # Starlette 以外からの呼び出しの場合は、直接エラーを返します
            response = PlainTextResponse("Not Found", status_code=404)
        await response(scope, receive, send)

    async def lifespan(self, scope: Scope, receive: Receive, send: Send) -> None:
        """
        ASGI ライフスパンメッセージを処理し、アプリケーションの起動およびシャットダウンイベントを管理できるようにします。
        """
        # ライフスパン実行ロジック。実行時、Starlette は ASGI サーバーと通信します。しかし、現時点では、このコードにはまだ開発されていない機能がある可能性があります。
        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:
        """
        Router クラスのメインエントリポイント。
        """
        # ルートの照合と実行のメイン関数
        # 現在、http、websocket、lifespan タイプのみがサポートされています
        assert scope["type"] in ("http", "websocket", "lifespan")

        # scope にルーターを初期化します
        if "router" not in scope:
            scope["router"] = self

        if scope["type"] == "lifespan":
            # ライフスパンロジックを実行します
            await self.lifespan(scope, receive, send)
            return

        partial = None

        # ルートの照合を実行します
        for route in self.routes:
            match, child_scope = route.matches(scope)
            if match == Match.FULL:
                # 完全一致（URLとメソッドの両方が一致）の場合
                # 通常のルート処理を実行します
                scope.update(child_scope)
                await route.handle(scope, receive, send)
                return
            elif match == Match.PARTIAL and partial is None:
                # 部分一致（URLは一致するが、メソッドは一致しない）の場合
                # 値を保持して一致を続行します
                partial = route
                partial_scope = child_scope

        if partial is not None:
            # 部分一致のルートがある場合も実行を続行しますが、ルートは HTTP メソッドエラーを返します
            scope.update(partial_scope)
            await partial.handle(scope, receive, send)
            return

        if scope["type"] == "http" and self.redirect_slashes and scope["path"]!= "/":
            # 一致しない状況、リダイレクトを判断します
            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:
                    # 再度一致します。結果が空でない場合、リダイレクト応答を送信します
                    redirect_url = URL(scope=redirect_scope)
                    response = RedirectResponse(url=str(redirect_url))
                    await response(scope, receive, send)
                    return

    # 上記のどのプロセスもヒットしなかった場合、ルートが見つからなかったことを意味します。この時点で、デフォルトのルートが実行され、デフォルトのデフォルト ルートは 404 not found です。
    await self.default(scope, receive, send)

ルーターのコードは非常にシンプルであることがわかります。ほとんどのコードは `call` メソッドに集中していますが、ルートをクエリするために複数のトラバースがあり、各ルートは一致を判断するために正規表現を実行します。一部の人は、この実行速度が遅いと考えるかもしれません。私も以前はそう思っていましたが、ルートツリーを実装して置き換えました（詳細は `route_trie.py` を参照してください）。しかし、パフォーマンステストの後、ルートの数が 50 を超えない場合、ループマッチングのパフォーマンスはルートツリーよりも優れていることがわかりました。100 を超えない場合、両者は同等です。そして、通常の状況では、指定するルートの数は 100 を超えません。そのため、このルートのマッチングパフォーマンスについて心配する必要はありません。それでも心配な場合は、`Mount` を使用してルートをグループ化することで、マッチング数を減らすことができます。

### 3.2. その他のルート
`Mount` は `BaseRoute` を継承しており、`HostRoute`、`WebSocketRoute` などの他のルートも同様です。これらのルートはわずかな実装の違い（主に初期化、ルートマッチング、逆引き）がありますが、同様のメソッドを提供します。まず `BaseRoute` を見てみましょう。
```python
class BaseRoute:
    def matches(self, scope: Scope) -> typing.Tuple[Match, Scope]:
        # 標準的なマッチング関数のシグネチャです。各 Route は (Match, Scope) タプルを返す必要があります。
        # Match には 3 つのタイプがあります。
        #   NONE: 一致しない。
        #   PARTIAL: 部分一致（URL は一致するが、メソッドは一致しない）。
        #   FULL: 完全一致（URL とメソッドの両方が一致）。
        # Scope は基本的に次の形式を返しますが、Mount はさらに多くのコンテンツを返します。
        #   {"endpoint": self.endpoint, "path_params": path_params}
        raise NotImplementedError()  # pragma: no cover

    def url_path_for(self, name: str, **path_params: str) -> URLPath:
        # 名前に従って逆引きを生成します
        raise NotImplementedError()  # pragma: no cover

    async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
        # Router によって一致された後に呼び出すことができる関数
        raise NotImplementedError()  # pragma: no cover

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        """
        ルートは、スタンドアロン ASGI アプリとして単独で使用できます。
        これらはほとんど常に Router 内で使用されますが、一部のツールや最小限のアプリケーションに役立つ可能性があります。
        """
        # ルートがスタンドアロン ASGI APP として呼び出された場合、マッチングを実行し、自身で応答します
        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)

BaseRoute は多くの機能を提供せず、他のルートはそれを拡張して機能を追加することがわかります。

• Route: 標準的な HTTP ルートです。HTTP URL と HTTP メソッドを通じてルートのマッチングを担当し、その後 HTTP ルートを呼び出すメソッドを提供します。
• WebSocketRoute: 標準的な WebSocket ルートです。HTTP URL に従ってルートをマッチングし、starlette.websocket の WebSocket を通じてセッションを生成し、対応する関数に渡します。
• Mount: ルートのネストされたカプセル化です。そのマッチング方法は URL のプレフィックスマッチングであり、ルールに適合する次のレベルの ASGI APP にリクエストを転送します。その次のレベルの ASGI APP が Router の場合、呼び出しチェーンは次のようになる可能性があります。Router->Mount->Router->Mount->Router->Route。Mount を使用すると、ルートをグループ化でき、マッチング速度を向上させることができます。推奨されます。さらに、リクエストを他の ASGI APP に分散することもできます。たとえば、Starlette->ASGI Middleware->Mount->Other Starlette->...
• Host: ユーザーリクエストの Host に従って、対応する ASGI APP（Route、Mount、ミドルウェアなど）にリクエストを分散します。

4. その他のコンポーネント

上記からわかるように、Starlette のほとんどのコンポーネントは ASGI APP として設計されており、互換性が非常に高いです。これによりパフォーマンスが多少犠牲になりますが、互換性は非常に強力です。他のコンポーネントも多かれ少なかれ ASGI APP として設計されています。他のコンポーネントを紹介する前に、まず Starlette の全体的なプロジェクト構造を見てみましょう。

├── middleware                       # ミドルウェア
├── applications.py                  # 起動アプリケーション
├── authentication.py                # 認証関連
├── background.py                    # バックグラウンドタスクをカプセル化し、応答が返された後に実行されます
├── concurrency.py                   # いくつかの小さな asyncio 関連のカプセル化。新しいバージョンでは、anyio ライブラリが直接使用されます。
├── config.py                        # 設定
├── convertors.py                    # いくつかの型変換メソッド
├── datastructures.py                # Url、Header、Form、QueryParam、State など、いくつかのデータ構造
├── endpoints.py                     # cbv とより高度な WebSocket カプセル化をサポートするルート
├── exceptions.py                    # 例外処理
├── formparsers.py                   # Form、File などの解析
├── graphql.py                       # GraphQL 関連の処理を担当
├── __init__.py
├── py.typed                         # Starlette に必要な型ヒント
├── requests.py                      # リクエスト、ユーザーがデータを取得するために使用
├── responses.py                     # 応答、ヘッダーとクッキーの初期化、異なる Response クラスに従った応答データの生成を担当し、クラス ASGI 呼び出しインターフェースを備えています。このインターフェースは ASGI プロトコルを Uvicorn サービスに送信します。送信後、バックグラウンドタスクがあれば、完了まで実行されます。
├── routing.py                       # ルーティング
├── schemas.py                       # OpenApi 関連スキーマ
├── staticfiles.py                   # 静的ファイル
├── status.py                        # HTTP ステータスコード
├── templating.py                    # Jinja に基づくテンプレート応答
├── testclient.py                    # テストクライアント
├── types.py                         # タイプ
└── websockets.py                    # WebSocket

上記のファイルは多数ありますが、いくつかの簡単なものは省略します。

4.1. Request

Request クラスは非常にシンプルです。HttpConnection を継承しています。このクラスは主に ASGI プロトコルから渡された Scope を解析して、URL やメソッドなどの情報を抽出します。また、Request クラスはリクエストデータの読み取りとデータ（HTTP 1.1 はサーバーがクライアントにデータをプッシュすることをサポートしています）の返却の機能を追加します。その中で、データの読み取りはコア機能である stream に依存します。そのソースコードは次のとおりです。

async def stream(self) -> typing.AsyncGenerator[bytes, None]:
    # すでに読み取られている場合、キャッシュからデータを取得します
    if hasattr(self, "_body"):
        yield self._body
        yield b""
        return

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

    self._stream_consumed = True
    while True:
        # ASGI コンテナの受信ループから継続的にデータを取得します
        message = await self._receive()
        if message["type"] == "http.request":
            body = message.get("body", b"")
            if body:
                # データが空でなければ返します
                yield body
            if not message.get("more_body", False):
                # すべてのボディデータが取得されたことを意味します
                break
        elif message["type"] == "http.disconnect":
            # クライアントとの接続が閉じられたことを意味します
            self._is_disconnected = true
            # 例外をスローします。 `await request.body()` または `await request.json()` を呼び出すユーザーは例外をスローします。
            raise ClientDisconnect()
    # 終端を示すために空のバイトを返します
    yield b""

この実装はシンプルですが、小さなバグがあります。Nginx やその他の Web サービスに詳しい人なら、一般的な中間サーバーはボディデータを処理せず、単に転送するだけであることを知っています。ASGI も同様です。URL とヘッダーを処理した後、Uvicorn は ASGI APP を呼び出し、send と receive オブジェクトを下位に渡します。これらの 2 つのオブジェクトは複数の ASGI APP を通過し、ユーザーが関数で使用するルート ASGI APP に到達します。したがって、Request が受け取る receive オブジェクトは Uvicorn によって生成されます。そして receive のデータソースは asyncio.Queue キューから来ます。ミドルウェアの分析から、各 ASGI APP は scope と receive に基づいて Request オブジェクトを生成するため、各レイヤーの ASGI APP の Request オブジェクトは一貫性がありません。ミドルウェアが Request オブジェクトを呼び出してボディを読み取ると、receive を介してキューのデータを事前に消費し、後続の ASGI APP が Request オブジェクトを通じてボディデータを読み取れなくなります。この問題のサンプルコードは次のとおりです。

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)

実行してリクエストの結果を確認します。

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

上記のように、結果は false です。これは、request.body の実行がタイムアウトしたことを意味します。receive キューがすでに空で、データを取得できないためです。タイムアウトがない場合、このリクエストはハングアップします。

この問題を解決するには、まず Request がどのようにボディを取得するかを見てみましょう。ユーザーはボディを複数回取得でき、データは同じであるため、実装のアイデアは、取得後にデータをキャッシュすることです。このアイデアに従いましょう。データは receive を介して取得されるため、データを読み取った後に receive 関数を構築できます。この関数は ASGI 通信プロトコルに似たデータを返し、完全なボディデータ（Request.stream のボディ取得の構築を満たす）を持っています。コードは次のとおりです。

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

その後、ASGI APP のいずれかのレイヤーがボディデータを取得する必要がある場合、この関数を呼び出してボディデータを取得でき、後続の ASGI APP がボディデータを取得する能力に影響を与えません。

5. 要約

これまでに、Starlette のいくつかの重要な機能コードを分析してきました。Starlette は優れた設計思想を持つ優れたライブラリです。将来的に独自のフレームワークを記述するのに役立つため、Starlette のソース コードを自分で読むことをお勧めします。

Leapcell: Web ホスティング、非同期タスク、Redis 用の次世代サーバーレス プラットフォーム

最後に、FastAPI をデプロイするのに最適なプラットフォームである Leapcell を紹介します。

1. マルチ言語サポート

• JavaScript、Python、Go、または Rust で開発します。

2. 無制限のプロジェクトを無料でデプロイ

• 使用量のみを支払います — リクエスト数や料金はありません。

3. 圧倒的なコスト効率

• 従量課金制でアイドル料金はありません。
• 例：$25 で 694 万リクエストを 60ms の平均応答時間でサポートします。

4. ストリーム化された開発者エクスペリエンス

• 直感的な UI で簡単なセットアップ。
• 完全自動化された CI/CD パイプラインと GitOps 統合。
• リアルタイムメトリクスとロギングで実践的な洞察を提供。

5. 簡単なスケーラビリティと高性能

• 自動スケーリングで高同時実行性を容易に処理。
• 運用オーバーヘッドゼロ — 構築に集中するだけ。

ドキュメントでさらに詳しく見る！

Leapcell Twitter: https://x.com/LeapcellHQ