FastAPIの内部構造を理解したい

◾️OpenAPIでスキーマ管理

app/main.py

app = FastAPI()

fastapi/applications.py

Class FastAPI(Starlette):
・・・
   def openapi(self) -> dict[str, Any]:

◾️DI(依存注入)

・アプリ全体に適用する共通の依存関係を登録。

fastapi/applications.py

Class FastAPI(Starlette):
・・・
   def __init__():
            dependencies: Annotated[
            Sequence[Depends] | None,
            Doc(),] = None

・Dependsオブジェクトをリストとして受け取り、内部の router に引き渡す。

fastapi/params.py

@dataclass(frozen=True)
class Depends:
    dependency: Callable[..., Any] | None = None
    use_cache: bool = True
    scope: Literal["function", "request"] | None = None

fastapi/applications.py

 self.router: routing.APIRouter = routing.APIRouter(
     routes=routes,
     dependencies=dependencies
)

fastapi/routing.py

class APIRouter(routing.Router):
      def __init__():
      ~~~~~~~
      self.dependant = get_dependant(
          path=self.path_format, call=self.endpoint, scope="function"
        )  

FastAPIの起動時に、Dependant オブジェクトを作る。

fastapi/dependencies/utils.py

def get_dependant(
    *,
    path: str,
    call: Callable[..., Any],
    scope: Literal["function", "request"] | None = None,
) -> Dependant:
    dependant = Dependant(
        call=call,
        name=name,
        path=path,

models.pyに定義してある。

fastapi/dependencies/models.py

@dataclass
class Dependant:
    path_params: list[ModelField] = field(default_factory=list)
    query_params: list[ModelField] = field(default_factory=list)
    header_params: list[ModelField] = field(default_factory=list)
    cookie_params: list[ModelField] = field(default_factory=list)
    body_params: list[ModelField] = field(default_factory=list)
    ~~~~
    scope: Literal["function", "request"] | None = None

fastapi/routing.py

class _DefaultLifespan:
                       ~~~~~        
        def __init__(self, router: "APIRouter") -> None:
            self._router = router
                       ~~~~~  
            solved_result = await solve_dependencies(
            request=websocket,
            dependant=dependant,
                       ~~~~~  

リクエストが来た瞬間、この関数が動く。

fastapi/dependencies/utils.py

# utils.py 内の solve_dependencies 関数
async def solve_dependencies(..., dependant: Dependant, ...) -> SolvedDependency:
    values: dict[str, Any] = {}
    
    # 依存先を再帰的に実行して解決
    for sub_dependant in dependant.dependencies:
        solved_result = await solve_dependencies(dependant=sub_dependant, ...)
        
        # 依存関数を実行して値を手に入れる
        solved = await call(**solved_result.values)
        
        # 辞書にkeyとvalueを追加
        if sub_dependant.name:
            values[sub_dependant.name] = solved

完成した辞書を関数に叩き込む。

fastapi/applications.py

class FastAPI(Starlette):
      def __init__():
          self.router: routing.APIRouter = routing.APIRouter(
          routes=routes,
         dependencies=dependencies
)

fastapi/routing.py

class _DefaultLifespan:
      def __init__(self, router: "APIRouter") -> None:
          self._router = router
def get_request_handler():
                      ~~~
    async def app(request: Request) -> Response:
                      ~~~
    raw_response = await run_endpoint_function(
    dependant=dependant,

fastapi/routing.py

async def run_endpoint_function(
    *, dependant: Dependant, values: dict[str, Any], is_coroutine: bool
) -> Any:

◾️Pydantic(バリデーションチェック)

fastapi/applications.py

Class FastAPI(Starlette):
・・・
   def openapi(self) -> dict[str, Any]:

fastapi/applications.py

Class FastAPI(Starlette):
     def __init__():
         self.router: routing.APIRouter = routing.APIRouter(
         routes=routes,
         dependencies=dependencies
)

fastapi/routing.py

class APIRouter(routing.Router):
    def get_request_handler(
        dependant: Dependant,
        body_field: ModelField | None = None,
        status_code: int | None = None,
              ~~~
    for sub_dependant in dependant.dependencies:
              ~~~
        solved_result = await solve_dependencies(
            request=request,
            dependant=dependant, ←Pydanticの情報が詰まっている
            body=body,
)

fastapi/dependencies/utils.py

async def solve_dependencies(
    *,
    request: Request | WebSocket,
    dependant: Dependant,
           ~~~
def _validate_value_with_model_field(
    *, field: ModelField, value: Any, values: dict[str, Any], loc: tuple[str, ...]
) -> tuple[Any, list[Any]]:
    if value is None:
        if field.field_info.is_required():
            return None, [get_missing_field_error(loc=loc)]
        else:
            return deepcopy(field.default), []
    return field.validate(value, values, loc=loc)

fastapi/_compat/v2.py

class ModelField:
      def validate(
        self,
        value: Any,
        values: dict[str, Any] = {},  # noqa: B006
        *,
        loc: tuple[int | str, ...] = (),
    ) -> tuple[Any, list[dict[str, Any]]]:
        try:
            return (
                self._type_adapter.validate_python(value, from_attributes=True),
                [],
            )
        except ValidationError as exc:
            return None, _regenerate_error_with_loc(
                errors=exc.errors(include_url=False), loc_prefix=loc
            )

型が違えば errors リストに詰め込み、合っていれば辞書（solved_result.values）に追加する。

---

Starlette の役割: HTTP 通信、WebSocket、ミドルウェアのスタック管理など

fastapi/applications.py

class FastAPI(Starlette):
　　　def __init__(
        self: AppType,
        *,
        debug: Annotated[
            bool
):
　　　　　　　　〜〜〜

Middleware: リクエストが FastAPI に届く直前で、共通のログを吐いたり、ヘッダーを書き換えたりする仕組み。

starlette>applications.py

class Starlette:
        def __init__(
        self: AppType,
         ~~~
        middleware: Sequence[Middleware] | None = None,
         ~~~
        lifespan: Lifespan[AppType] | None = None,
    ) -> None:
         ~~~
        self.user_middleware = [] if middleware is None else list(middleware)
        self.middleware_stack: ASGIApp | None = None

__call__：Uvicorn などのサーバーからリクエストが届いたとき、最初に叩かれる。

starlette>applications.py

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

starlette>applications.py

    def build_middleware_stack(self) -> ASGIApp:
        debug = self.debug
        error_handler = None
        exception_handlers: dict[Any, ExceptionHandler] = {}

        for key, value in self.exception_handlers.items():
            if key in (500, Exception):
                error_handler = value
            else:
                exception_handlers[key] = value

        middleware = (
            [Middleware(ServerErrorMiddleware, handler=error_handler, debug=debug)]
            + self.user_middleware
            + [Middleware(ExceptionMiddleware, handlers=exception_handlers, debug=debug)]
        )

        app = self.router
        for cls, args, kwargs in reversed(middleware):
            app = cls(app, *args, **kwargs)
        return app

WebSocket: リアルタイムな通信をする。

特徴：一度つないだら、やり取りが双方向（Receive/Send）でループし続ける

starlette>applications.py

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()

@app.websocket("/ws") と書いたとき、内部では APIWebSocketRoute が生成されます。

fastapi/routing.py

        def decorator(func: DecoratedCallable) -> DecoratedCallable:
            self.add_api_websocket_route(
                path,
                func,
                name=name,
                dependencies=dependencies,
            )
            return func

        return decorator

APIWebSocketRoute オブジェクトが作られ、Starlette のルーターに登録されます。

fastapi/routing.py

    def add_api_websocket_route(
        self,
        path: str,
        endpoint: Callable[..., Any],
        name: str | None = None,
        *,
        dependencies: Sequence[Depends] | None = None,
    ) -> None:
        self.router.add_api_websocket_route(
            path,
            endpoint,
            name=name,
            dependencies=dependencies,
        )

実際に WebSocket の接続が来た時、Uvicorn から呼ばれる。

fastapi/applications.py

class FastAPI(Starlette):
    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if self.root_path:
            scope["root_path"] = self.root_path
        await super().__call__(scope, receive, send)　← 親（Starlette）へ

starlette/applications.py

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

starlette/applications.py

class Starlette:
   def build_middleware_stack(self) -> ASGIApp:
        debug = self.debug
        error_handler = None
        exception_handlers: dict[Any, ExceptionHandler] = {}

        middleware = (
            [Middleware(ServerErrorMiddleware, handler=error_handler, debug=debug)]
            + self.user_middleware
            + [Middleware(ExceptionMiddleware, handlers=exception_handlers, debug=debug)]
        )

       app = self.router

starlette/applications.py

class Starlette:
    def __init__(        
        self.debug = debug
        self.state = State()
        self.router = Router(routes, on_startup=on_startup, on_shutdown=on_shutdown, lifespan=lifespan)

starlette/routing.py

class Router:
    def __init__(
        self,
        routes: Sequence[BaseRoute] | None = None,
        redirect_slashes: bool = True,
        default: ASGIApp | None = None,
        on_startup: Sequence[Callable[[], Any]] | None = None,
        on_shutdown: Sequence[Callable[[], Any]] | None = None,

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> 
        await self.middleware_stack(scope, receive, send)

starlette/routing.py

class Router:
    async def app(self, scope: Scope, receive: Receive, send: Send) -> None:
        assert scope["type"] in ("http", "websocket", "lifespan")

        if "router" not in scope:
            scope["router"] = self

        if scope["type"] == "lifespan":
            await self.lifespan(scope, receive, send)
            return

    for route in self.routes:
            match, child_scope = route.matches(scope)
            if match == Match.FULL: # URL は /ws/battle か？タイプは websocket か？
                scope.update(child_scope)
                await route.handle(scope, receive, send) # WebSocketRouteへ遷移
                return

starlette/routing.py

async def handle(self, scope: Scope, receive: Receive, send: Send) -> 
            ~~~~
            await self.app(scope, receive, send)

starlette/routing.py

class Router:
   def __init__():
            ~~~
         self.middleware_stack = self.app
         if middleware:
            for cls, args, kwargs in reversed(middleware):
                self.middleware_stack = cls(self.middleware_stack, *args, **kwargs)