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을 Uvicorn 호출을 위한 lifespan으로 변환하기 때문입니다
        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이며, 예외가 발견될 때 오류 스택을 출력하거나 디버깅을 쉽게 위해 디버그 모드에서 오류 페이지를 표시할 수 있습니다.
        # 두 번째 레이어는 사용자 미들웨어 레이어로, 등록된 모든 사용자 미들웨어가 저장됩니다.
        # 세 번째 레이어는 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은 예외 처리를 미들웨어에 위임하며, 이는 다른 웹 애플리케이션 프레임워크에서는 드물게 볼 수 있습니다. 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 상태 코드와 Exception 유형을 모두 지원합니다
        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를 통해 예외에 대한 해당 콜백 함수를 찾습니다.
        # 
        # 사용자가 기본 클래스를 마운트할 수 있으며, 나중에 이 기본 클래스의 하위 클래스도 기본 클래스에 등록된 콜백을 호출합니다.
        # 예를 들어, 사용자가 기본 클래스를 등록하고, 그런 다음 두 개의 예외, 사용자 예외와 시스템 예외가 모두 이 기본 클래스에서 상속됩니다.
        # 나중에 함수가 사용자 예외 또는 시스템 예외를 발생시키면 기본 클래스에 등록된 해당 콜백이 실행됩니다.
        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

            # 응답당 하나의 예외만 처리
            if response_started:
                msg = "처리된 예외를 잡았지만 응답이 이미 시작되었습니다."
                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()
            # 예외가 발생하지 않으면 사용자 오류일 수 있습니다 (예: 빈 응답 반환).
            # 이때 클라이언트에 응답을 반환할 수 없으므로, 500 응답 생성을 용이하게 하기 위해 예외를 발생시켜야 합니다.
            raise RuntimeError("응답이 반환되지 않았습니다.")

        # 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와 매우 유사합니다 (따라서 이 부분은 더 이상 자세히 설명하지 않습니다). 전체 로직은 대부분 동일합니다. 그러나 등록된 HTTP 상태 코드가 500일 때만 콜백이 ServerErrorMiddleware에 등록됩니다.

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

ServerErrorMiddleware는 ASGI APP의 최상위 레벨에 있습니다. 후속 예외 처리 작업을 맡습니다. 해야 할 일은 간단합니다. 다음 레벨 ASGI APP 처리 중에 예외가 발생하면 후속 로직으로 들어갑니다.

• 1. 디버그가 활성화된 경우 디버그 페이지를 반환합니다.
• 2. 등록된 콜백이 있으면 등록된 콜백을 실행합니다.
• 3. 위의 둘 다 아닌 경우 500 응답을 반환합니다.

3. 라우트

Starlette에서 라우트는 두 부분으로 나뉩니다. 하나는 미들웨어 레벨 아래에 있는 제가 Real App의 Router라고 부르는 부분입니다. 주요 로드 조회 및 일치, 애플리케이션 시작 및 종료 처리 등 미들웨어를 제외한 Starlette의 거의 모든 것을 담당합니다. 다른 하나는 Router에 등록된 라우트로 구성됩니다.

3.1. Router

Router는 간단합니다. 주요 책임은 라우트를 로드하고 일치시키는 것입니다. 라우트 로드 부분을 제외한 소스 코드와 주석은 다음과 같습니다.

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 찾을 수 없음입니다.
    await self.default(scope, receive, send)

Router 코드는 상당히 간단합니다. 대부분의 코드는 `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 (라우트, 마운트, 미들웨어 등)로 요청을 분배합니다.

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                     # 응답, 헤더 및 쿠키 초기화, 다양한 응답 클래스에 따라 응답 데이터를 생성하고, 그 다음 클래스 ASGI 호출 인터페이스를 가집니다. 이 인터페이스는 Uvicorn 서비스에 ASGI 프로토콜을 보냅니다. 보낸 후 백그라운드 작업이 있으면 완료될 때까지 실행됩니다.
├── 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 컨테이너의 receive 루프에서 지속적으로 데이터 가져오기
        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):
                # 모든 body 데이터가 얻어졌음을 의미
                break
        elif message["type"] == "http.disconnect":
            # 클라이언트와의 연결이 닫혔음을 의미
            self._is_disconnected = true
            # 예외 발생. `await request.body()` 또는 `await request.json()`을 호출하는 사용자는 예외를 발생시킵니다.
            raise ClientDisconnect()
    # 끝을 표시하기 위해 빈 바이트 반환
    yield b""

이 구현은 간단하지만 작은 버그가 있습니다. Nginx 또는 다른 웹 서비스에 익숙한 사람들은 일반적인 중간 서버는 본문 데이터를 처리하지 않고 전달하기만 한다는 것을 알고 있습니다. ASGI도 마찬가지입니다. URL과 헤더를 처리한 후 Uvicorn은 ASGI APP를 호출하고 send 및 receive 객체를 아래로 전달합니다. 이 두 객체는 여러 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: 웹 호스팅, 비동기 태스크 및 Redis를 위한 차세대 서버리스 플랫폼

마지막으로 FastAPI를 배포하는 데 가장 적합한 플랫폼인 Leapcell을 공유하고 싶습니다.

1. 다국어 지원

• JavaScript, Python, Go 또는 Rust로 개발하세요.

2. 무료로 무제한 프로젝트 배포

• 사용한 만큼만 지불하세요. 요청 수, 요금 없음.

3. 비교할 수 없는 비용 효율성

• 사용한 만큼 지불하며 유휴 요금 없음.
• 예: $25로 6.94M개의 요청을 60ms 평균 응답 시간 지원.

4. 간소화된 개발자 경험

• 직관적인 UI로 쉬운 설정.
• 완전 자동화된 CI/CD 파이프라인 및 GitOps 통합.
• 실행 가능한 인사이트를 위한 실시간 메트릭 및 로깅.

5. 쉬운 확장성 및 고성능

• 자동 확장을 통해 높은 동시성을 쉽게 처리.
• 제로 운영 오버헤드 - 빌드에만 집중하세요.

문서에서 더 자세히 알아보세요!

Leapcell 트위터: https://x.com/LeapcellHQ