FastAPI 구조 살펴보기 2

Dev Smile·2024년 11월 17일

FastAPI Starlette

FastAPI

목록 보기

5/10

FastAPI 구조 살펴보기 두 번째 글입니다.

https://ceb10n.medium.com/understanding-fastapi-how-starlette-works-d518a3d22222

오늘은 이 글을 참고하였습니다.

이전 글에서 FastAPI가 경량 ASGI 프레임워크/툴킷인 Starlette을 기반으로 만들어졌다는 것을 설명드렸습니다.

이제 FastAPI의 작동 원리를 이해하기 위해, Starlette가 어떤 기능을 제공하는지와 HTTP 요청을 처리하는 방식 등을 자세히 알아보겠습니다.

⚠️ 참고: starlette를 사용하기 위해서는 Python 3.8 이상의 버전이 필요합니다.

1. Startlette 이해하기

1-1. Hello World

Starlette를 사용하여 이전 글에서 구현했던 간단한 Hello World를 다시 만들어 보겠습니다.

from starlette.applications import Starlette
from starlette.responses import PlainTextResponse
from starlette.routing import Route

async def hello(request):
    return PlainTextResponse("Hello, World!")

app = Starlette(routes=[
    Route('/', hello),
])

이전 글에서 작성했던 코드와 비교해서 확인해보겠습니다.

class SimplestFrameworkEver:
    async def __call__(self, scope, receive, send):
        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!",
        })

app = SimplestFrameworkEver()

두 경우 모두 클래스(SimplestFrameworkEver, Starlette)가 있으며 이 클래스의 인스턴스를 생성하고 이를 처리하기 위해 ASGI 서버에 전달합니다.

ASGI 사양(https://asgi.readthedocs.io/en/latest/index.html)에 따르면, ASGI 애플리케이션은 하나의 비동기 호출 가능 객체로 정의되며, 이 객체는 "scope"라는 딕셔너리와 함께 "receive" 및 "send"라는 두 가지 비동기 함수 매개변수를 받아야 합니다.
따라서 Starlette 개체를 ASGI 서버에 전달하는 경우 __call__ 메서드가 구현된 클래스여야 합니다.

핵심 요점
- ASGI는 비동기 함수이어야 함.
- 이 함수는 scope, receive, send 세 가지 매개변수를 받아야 함.
  - scope: scope는 현재 요청의 컨텍스트 정보를 담고 있는 딕셔너리입니다. 예를 들어, HTTP 요청일 경우 요청 메서드(GET, POST 등), 경로, 헤더 등의 정보가 포함됩니다. 이 정보는 요청 처리에 필요한 모든 배경 정보를 제공해줍니다.
  - receive: receive는 서버가 클라이언트로부터 메시지를 비동기적으로 받을 때 사용하는 함수입니다. 웹소켓과 같은 경우, 클라이언트로부터 데이터를 받을 때 이 함수를 사용합니다.
  - send: send는 서버가 클라이언트에게 응답을 보낼 때 사용하는 비동기 함수입니다. 서버가 클라이언트에게 메시지나 데이터를 전송할 때 이 함수를 사용합니다.
- Starlette 객체를 ASGI 서버에 전달하려면, 그 객체는 __call__ 메서드를 구현한 클래스여야 함. __call__ 메서드는 파이썬에서 특별한 메서드로, 이 메서드를 구현하면 객체를 함수처럼 호출할 수 있게 됩니다. Starlette와 같은 웹 프레임워크는 ASGI 서버와 통신할 때, 이 메서드를 통해 요청을 처리합니다. Starlette 객체가 ASGI 서버에 전달되었을 때, ASGI 서버는 이 객체가 비동기적으로 호출될 수 있는지(즉, __call__ 메서드를 가지고 있는지)를 확인해야 합니다. __call__ 메서드는 서버가 이 객체를 요청 처리 함수처럼 사용할 수 있게 해줍니다.

1-2. Starlette 코드 확인

Starlette가 ASGI 사양을 지키는 지 확인하기 위해 Starlette의 소스 코드를 확인해보겠습니다.

Starlette 깃허브(https://github.com/encode/starlette/)의 starlette/applications.py 에서 Starlette클래스를 찾을 수 있습니다.

Starlette 클래스 : https://github.com/encode/starlette/blob/18bbb5c948a3591f02ac2c83c08db1f5cae6c444/starlette/applications.py#L27

# https://github.com/encode/starlette/blob/18bbb5c948a3591f02ac2c83c08db1f5cae6c444/starlette/applications.py#L109
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)

“call” 메서드는, 함수를 호출 하는 것처럼 class의 객체도 호출할 수 있게 만들어 줍니다.

위 코드에서 Starlette는 "callable" 클래스이며 각 요청을 실행할 미들웨어 목록이 있는 것으로 보입니다.

코드는 몇 줄이 되지 않지만, 보이는 것만큼 단순하지는 않습니다. 그리고 middleware_stack이 무엇인지도 알아야 합니다.

1-2-1. ASGIApp

지금까지 본 바에 따르면, Starlette는 scope, receive, send 매개변수를 받는 호출 가능한 객체입니다. 이것이 바로 ASGI 애플리케이션이 되어야 하는 모습입니다. 그렇다면 middleware_stack은 무엇일까요?

self.middleware_stack: ASGIApp | None = None

middleware_stack은 ASGIApp입니다. 그러나 Starlette의 타입을 보면 ASGIApp은 ASGI callable일 뿐입니다.

ASGIApp = typing.Callable[[Scope, Receive, Send], typing.Awaitable[None]]

Starlette.build_middleware_stack을 살펴보면 이상한 코드 조각을 볼 수 있습니다.

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=app, *args, **kwargs)
return app

이 코드에서는 Starlette가 일종의 책임 연쇄(https://refactoring.guru/design-patterns/chain-of-responsibility) 구조로 미들웨어들을 생성하고, 마지막으로 라우터나 엔드포인트를 처리하는 것입니다.

다시 말해서 위 코드는 Starlette에서 미들웨어를 쌓아 올리는 방식, 즉 “미들웨어 스택”을 구성하는 코드입니다. 코드를 쪼개어 설명해보겠습니다.

middleware ⇒ middleware 변수는 미들웨어 리스트를 만들어내는 과정입니다.
1. [Middleware(ServerErrorMiddleware, handler=error_handler, debug=debug)] ⇒ 서버 오류를 처리하기 위한 ServerErrorMiddleware가 리스트에 첫 번째 요소로 추가됩니다.
2. self.user_middleware ⇒ 유저가 정의한 미들웨어가 중간에 추가됩니다.
3. [Middleware(ExceptionMiddleware, handlers=exception_handlers, debug=debug)] ⇒ 예외 처리를 담당하는 ExceptionMiddleware가 추가됩니다.
4. 위 과정을 거치면 middleware에는, [ServerErrorMiddleware, self.user_middleware, ExceptionMiddleware] 순서로 미들웨어가 들어갑니다.
app ⇒ 현재 애플리케이션의 라우터를 가리킵니다.
1. 이 app 객체에 미들웨어를 하나씩 추가하여 스택을 쌓아갈 예정입니다.
역순으로 미들웨어 쌓기
1. reversed(middleware) ⇒ 미들웨어 리스트를 뒤집어, [ExceptionMiddleware, self.user_middleware, ServerErrorMiddleware] 순서로 미들웨어를 불러옵니다.
  - 왜냐하면, 미들웨어는 겹겹이 쌓이면서 가장 바깥쪽 미들웨어가 맨 처음 실행되기 때문입니다.
2. for cls, args, kwargs in reversed(middleware) ⇒ 각 미들웨어의 클래스(cls)와 인자(args, kwargs)를 꺼내어 사용합니다.
3. app = cls(app=app, *args, **kwargs) ⇒ 현재의 app을 인자로 하여 새로운 미들웨어 인스턴스를 생ㄷ성하고, 이 인스턴스를 app으로 갱신합니다.
  - 결과적으로는 app에 모든 미들웨어가 포함된 형태가 됩니다.
return app ⇒ 모든 미들웨어가 적용된 app 객체를 반환합니다.
- 이렇게 반환된 app은 요청이 들어오면 각 미들웨어를 거쳐서 최종적으로 라우터에 도달합니다.
요약
1. 이 코드의 전체적인 흐름은 “요청이 미들웨어 스택을 통과해 애플리케이션에 도달하도록 구성” 하는 것입니다.
2. 미들웨어가 역순으로 쌓이기 때문에, ServerErrorMiddleware는 가장 바깥 쪽에, ExceptionMiddleware는 가장 안쪽에 위치하여, 애플리케이션이 요청을 처리하기 전후에 오류나 예외를 처리하게 됩니다.
3. 결론적으로 작동 방식은 다음과 같습니다.
```
-> ServerErrorMiddleware
    -> Other Middlewares
        -> ExceptionMiddleware
            -> Router
```

각 ASGIApp은 다른 ASGIApp을 의존성으로 받아들이고, 호출될 때마다 다음 앱을 호출합니다. 이 과정은 ExceptionMiddleware에 도달할 때까지 이어지며, ExceptionMiddleware는 예외 처리를 위해 라우터를 감싸게 됩니다.

그 후 라우터는 더 이상 다른 ASGIApp을 의존성으로 받지 않고, 자체적으로 경로를 매칭하고 경로 작업 함수를 실행하는 자체 애플리케이션 함수(ASGI 앱)를 구현합니다.

async def app(self, scope: Scope, receive: Receive, send: Send) -> None:
    # ... previous code

    for route in self.routes:
        # Determine if any route matches the incoming scope,
        # and hand over to the matching route if found.
        match, child_scope = route.matches(scope)
        if match == Match.FULL:
            scope.update(child_scope)
            await route.handle(scope, receive, send)
            return

    # ... code continues

위 코드의 app 함수는 비동기적으로 HTTP 요청을 받아서 routes 중 하나와 일치하는지 확인한 후, 일치하는 라우트가 있으면 해당 라우트에 요청을 전달하여 처리하는 역할을 합니다. 이번에도 코드를 쪼개어 설명해보겠ㅅ브니다.

async def app(self, scope: Scope, receive: Receive, send: Send) -> None ⇒ 함수 정의
1. app 함수는 비동기 함수로, scope, receive, send라는 세가지 매개변수를 받습니다.
2. scope ⇒ HTTP요청의 메타데이터(요청 타입, 경로 등)를 포함
3. receive ⇒ 클라이언트가 보낸 데이터를 비동기적으로 읽는 함수
4. send ⇒ 클라이언트에게 응답을 비동기적으로 보내는 함수
for route in self.routes ⇒ routes를 순회합니다.
1. self.routes ⇒ 애플리케이션이 가지고 있는 라우트 목록
2. 각 route를 순회하면서 요청이 이 route와 일치하는지 확인합니다.
match, child_scope = route.matches(scope) ⇒ 라우트와 일치 여부 확인
1. route.matches(scope) ⇒ 현재 scope와 route가 일치하는지 검사하고, 결과로 match와 child_scope를 반환합니다.
2. match ⇒ 라우트가 일치하는 정도를 나타내며, Match.FULL이 되면 완전히 일치함을 나타냅니다.
3. child_scope ⇒ 일치하는 라우트가 추가로 제공할 수 있는 추가 정보(필요한 경우)를 담고 있습니다.
그 밑의 코드 ⇒ 일치하는 라우트가 있는 경우 요청 처리
1. if match == Match.FULL ⇒ 위에서 말씀드렸듯, match가 Match.FULL 인 경우 해당 route가 완전히 일치한다는 것을 의미합니다.
2. scope.update(child_scope) ⇒ scope에 child_scope의 정보를 업데이트하여 추가적인 데이터를 포함 시킵니다.
3. await route.handle(scope, receive, send) ⇒ route.handle()을 호출해 해당 라우트에서 요청을 처리합니다.
  - handle() 함수는 클라이언트로부터 receive로 데이털르 받고, send로 응답을 보냅니다.
4. return ⇒ 요처이 처리되었으므로 함수를 종료하고, 이후의 라우트 확인을 생략합니다.
요약
1. 이 함수의 목적은 요청을 받아 routes 중 하나와 일치하는 지 확인하고, 일치하는 라우트가 있다면 해당 라우트에 요청을 넘겨 처리하는 것 입니다.
2. 즉, 요청이 들어오면 routes를 순차적으로 검사하여 일치하는 라우트를 찾고, 찾으면 route.handle() 을 통해 실제 처리 로직을 실행하여 클라이언트에게 응답합니다.

이제 Starlette 요청의 흐름을 알았으니, 요청의 경로를 기록하는 간단한 미들웨어 하나와, 응답이 전송된 후 모든 것이 잘 처리되었음을 기록하는 또 다른 미들웨어를 만들 수 있습니다.

class LogRequestMiddleware:
    def __init__(self, app: ASGIApp):
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send):
        logging.info(f"-> received a request @ {scope['path']}")
        await self.app(scope, receive, send)

class LogResponseMiddleware:
    def __init__(self, app: ASGIApp) -> None:
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send):
        await self.app(scope, receive, send)
        logging.info("-> wow, we did it")

async def hello(request):
    logging.info("Great news, we got a request!")
    return PlainTextResponse("Hello, World!")

app = Starlette(
    routes=[
        Route('/', hello),
    ],
    middleware=[
        Middleware(LogRequestMiddleware),
        Middleware(LogResponseMiddleware)
    ]
)

이 코드에서는 LogRequestMiddleware와 LogResponseMiddleware라는 두 가지 미들웨어를 정의하고, 각각 요청이 들어올 때와 응답을 보낼 때 로그를 남기는 역할을 합니다. 코드를 확인하면서, 각각의 미들웨어와 hello 함수, app 객체의 구조를 단계별로 설명하겠습니다.

LogRequestMiddleware 클래스
```
class LogRequestMiddleware:
    def __init__(self, app: ASGIApp):
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send):
        logging.info(f"-> received a request @ {scope['path']}")
        await self.app(scope, receive, send)
```
1. __init__: LogRequestMiddleware는 생성될 때 app을 받아서 self.app에 저장합니다. 이 app은 Starlette 애플리케이션으로, 미들웨어가 요청을 처리한 뒤에 이어서 호출할 대상입니다.
2. __call__: 이 메서드는 요청이 들어올 때 호출됩니다.
  - logging.info(f"-> received a request @ {scope['path']}"): 요청의 경로 정보를 scope['path']에서 추출하여 로그에 기록합니다. 예를 들어, 경로가 /hello라면 로그에 > received a request @ /hello와 같은 메시지가 남습니다.
  - await self.app(scope, receive, send): 다음 미들웨어(혹은 애플리케이션)로 요청을 전달합니다.
3. 이 미들웨어는 요청을 받을 때 로그를 남기는 역할만 하고, 요청 자체는 수정하지 않습니다.
LogResponseMiddleware 클래스
```
class LogResponseMiddleware:
    def __init__(self, app: ASGIApp) -> None:
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send):
        await self.app(scope, receive, send)
        logging.info("-> wow, we did it")
```
1. __init__: LogResponseMiddleware도 마찬가지로 생성 시 app을 받아 self.app에 저장합니다.
2. __call__: 요청이 들어오면 호출되며, 여기서는 app(scope, receive, send)가 먼저 호출됩니다.
  - await self.app(scope, receive, send): LogRequestMiddleware와 달리 이 미들웨어는 응답이 생성된 후에 동작합니다.
  - logging.info("-> wow, we did it"): 응답 처리가 완료된 후에 로그를 남깁니다.
hello 함수
```
async def hello(request):
    logging.info("Great news, we got a request!")
    return PlainTextResponse("Hello, World!")
```
1. hello: / 경로로 요청이 들어오면 실행되는 비동기 함수입니다.
2. logging.info("Great news, we got a request!"): 요청이 성공적으로 처리되었음을 로그에 남깁니다.
3. return PlainTextResponse("Hello, World!"): 텍스트 응답으로 "Hello, World!" 메시지를 클라이언트에게 보냅니다.
app 객체
```
app = Starlette(
    routes=[
        Route('/', hello),
    ],
    middleware=[
        Middleware(LogRequestMiddleware),
        Middleware(LogResponseMiddleware)
    ]
)
```
1. routes: Route('/', hello)를 통해 / 경로로 요청이 들어왔을 때 hello 함수가 실행되도록 설정합니다.
2. middleware: LogRequestMiddleware와 LogResponseMiddleware가 각각 설정되어, 요청이 들어오면 LogRequestMiddleware가 먼저 실행되고, 요청이 처리된 후에 LogResponseMiddleware가 실행됩니다.

그러면 다음과 같은 결과를 얻을 수 있습니다.

INFO:root:-> received a request @ /
INFO:root:Great news, we got a request!
INFO:     127.0.0.1:51770 - "GET / HTTP/1.1" 200 OK
INFO:root:-> wow, we did it

Starlette의 미들웨어에 대해 더 알아보려면, 자체 문서인 'middleware’(https://www.starlette.io/middleware/)를 읽어보세요. 프로젝트 팀이 이를 잘 문서화해 놓았습니다.

1-2-2. Routes and Router

마지막으로, 모든 미들웨어 체인을 거친 후에 ExceptionMiddleware로 감싸진 라우터가 실행됩니다. 이로 인해 라우터는 예외를 처리할 수 있게 됩니다. 라우터는 처리해야 할 라우트(경로)의 목록을 가지고 있습니다.

라우터가 일치하는 라우트(경로)를 찾으면, 그 라우트의 처리 함수를 호출합니다. 이 처리 함수는 우리가 만든 엔드포인트를 호출하는데, 엔드포인트는 기본적으로 Starlette 앱을 만들 때 전달한 함수나 클래스입니다:

app = Starlette(
    routes=[
        Route('/', hello), # -> hello is the function that will be handled by Router's handle
    ],

실제로 라우터도 ASGIApp이므로 라우터만 생성하여 모든 Starlette의 미들웨어를 해제할 수 있습니다.

app = Router(routes=[
    Route('/', hello)
])

2. 그래서 FastAPI는?

'글 제목은 'FastAPI 구조 살펴보기'인데, 벌써 두 번째 글인데도 계속 ASGI 스펙이나 Starlette에 대해 쓰고 있네.'

여기까지 읽었을 때, 이런 물음이 생기실 수 있을 것 같습니다.

하지만 제가 말했던 것을 기억하세요: FastAPI는 Starlette입니다. FastAPI는 Starlette 위에 구축된 프레임워크이고, FastAPI의 소스 코드를 보면 이렇게 나와 있습니다:

class FastAPI(Starlette):

    def __init__(
    # ... code continues

제가 설명한 내용은 FastAPI의 내부, 즉 FastAPI를 지탱하는 기반에 대한 것이었습니다.

다음 글에서는 FastAPI가 어떻게 Starlette와 ASGI 스펙을 확장하여 Pydantic 모델, OpenAPI 스펙 등을 기본 제공하는지 살펴볼 예정입니다.

Dev Smile

이전 포스트

FastAPI 구조 살펴보기 1

다음 포스트