FastAPI에서 APP Mount + CORSMiddleware

째하·2024년 8월 27일
FastAPIcorsmountpytohn
0
post-thumbnail

FastAPI App Mount CORS

개요

최근 QB 개발 서버에 Prometheus Cilent 기능을 붙여 HTTP 트래픽 횟수,

API 호출 횟수 등 데이터 추적을 할려고 한다. Prometheus 서버는

QB 개발 서버의 특정 API 주소(/metrics)를 호출해 API 데이터를 파싱 후
Prometheus 서버에 저장한다. 해당 metrics 주소를 Origin 을 적용해 private하게

접근할려고 한다.

문제 사항

# starlette/aplications.py
class Starlette:
    def add_middleware(
        self,
        middleware_class: type[_MiddlewareClass[P]],
        *args: P.args,
        **kwargs: P.kwargs,
    ) -> None:
        if self.middleware_stack is not None:  # pragma: no cover
            raise RuntimeError("Cannot add middleware after an application has started")
        self.user_middleware.insert(0, Middleware(middleware_class, *args, **kwargs))
        
# ./test_main.py

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://main.co.kr"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
app.add_middleware(LoggiMiddleware)
app.add_middleware(MetricsMiddleware)

class CORSMiddlewareTest(CORSMiddleware): ...

metrics_app = FastAPI()

metrics_app .add_middleware(
    CORSMiddleware,
    allow_origins=["http://metrics.co.kr"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.mount("/metrics", metrics_app)

@metrics_app.get("/")
def get_prometheus_metrics():
    return "Test"

해당 구조에서 http://localhost:3000 서버에서 ~/metrics API 를 요청했으면 예상하는 동작은

가 발생해야 한다. 하지만

이렇게 CORS 에러가 발생하지 않고 정상적으로 데이터를 가져온다.
왜 그럴까?? FastAPI에서 MiddleWare를 어떻게 등록하고 어떤 순서로 실행하는 지,
CORSMiddleWare는 어떻게 동작하는 지 알아야 한다.

FastAPI Middleware

FastAPI에서 MiddleWare를 등록하는 순서

# starlette/aplications.py
class Starlette:
    def add_middleware(
        self,
        middleware_class: type[_MiddlewareClass[P]],
        *args: P.args,
        **kwargs: P.kwargs,
    ) -> None:
        if self.middleware_stack is not None:  # pragma: no cover
            raise RuntimeError("Cannot add middleware after an application has started")
        self.user_middleware.insert(0, Middleware(middleware_class, *args, **kwargs))
        
# ./test_main.py

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://main.co.kr"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
app.add_middleware(LoggiMiddleware)
app.add_middleware(MetricsMiddleware)

class CORSMiddlewareTest(CORSMiddleware): ...

metrics_app = FastAPI()

metrics_app .add_middleware(
    CORSMiddleware,
    allow_origins=["http://metrics.co.kr"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

## Logging
<class 'starlette.middleware.cors.CORSMiddleware'>
<class 'app.test_main.LoggiMiddleware'>
<class 'app.test_main.MetricsMiddleware'>
<class 'app.test_main.CORSMiddlewareTest'>

선언한 순서대로 적용이 되고 있는 걸 확인할 수 있다.

FastAPI에서 Middleware를 실행하는 순서

등록한 순서는 선언한 순서대로 적용이 되는 걸 확인할 수 있다.

하지만 실행은 선언한 순서대로 가는 지, 별개인지 확인해보자

Test 실행
Metrics 실행
<starlette.middleware.exceptions.ExceptionMiddleware object at 0x00000277F6C1E510> + ['http://main.co.kr']
<starlette.middleware.exceptions.ExceptionMiddleware object at 0x000001A0EA132B90> + ['http://metrics.co.kr']

선언한 순서가 아닌 역순으로 호출이 되는 걸 확인할 수 있다.

하지만 starlette 프레임워크에서는 선언한 순서대로 호출이 된다고 한다.

링크 : https://github.com/encode/starlette/issues/479

CORSMiddleware

CORS에 대한 설명은 넘어가겠다.

FastAPI에서는 CORSMiddleware를 제공해준다.

from fastapi.middleware.cors import CORSMiddleware

CORSMiddleware는 어떻게 API Request를 어떻게 중간에 가로채서 작업을 하는 지 코드 분석을 하고자 한다. ( 현재 코드에 불필요한 부분은 삭제했습니다.)

  # starlette/middleware/cors
  class CORSMiddleware:
    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] != "http":  # pragma: no cover
            await self.app(scope, receive, send)
            return

        method = scope["method"]
        headers = Headers(scope=scope)
        origin = headers.get("origin")
        if origin is None:
            await self.app(scope, receive, send)
            return
        if method == "OPTIONS" and "access-control-request-method" in headers:
            response = self.preflight_response(request_headers=headers)
            await response(scope, receive, send)
            return
        **await self.simple_response(scope, receive, send, request_headers=headers)**

    async def simple_response(
        self, scope: Scope, receive: Receive, send: Send, request_headers: Headers
    ) -> None:
        send = functools.partial(self.send, send=send, request_headers=request_headers)
        # self.send 함수를 send, request_headers 파라미터에 인자 값을 넣어 새로운 self.send() 함수를 만듦
        **await self.app(scope, receive, send)**
        
		 async def send(
        self, message: Message, send: Send, request_headers: Headers
    ) -> None:
        if message["type"] != "http.response.start":
            await send(message)
            return

        message.setdefault("headers", [])
        headers = MutableHeaders(scope=message)
        headers.update(self.simple_headers)
        origin = request_headers["Origin"]
        has_cookie = "cookie" in request_headers

        # If request includes any cookie headers, then we must respond
        # with the specific origin instead of '*'.
        if self.allow_all_origins and has_cookie:
            self.allow_explicit_origin(headers, origin)

        # If we only allow specific origins, then we have to mirror back
        # the Origin header in the response.
        elif not self.allow_all_origins and self.is_allowed_origin(origin=origin):
            self.allow_explicit_origin(headers, origin)

        await send(message)        
       
graph TD
    A[HTTP 요청] --> B[기본 앱: Metrics 미들웨어]
    B --> C[기본 앱: Logging 미들웨어]
    C --> D[기본 앱 : CORS 미들웨어]
    D --> E[서브 앱 : CORS 미들웨어]
    E --> F{요청이 서브 앱으로 가는가?}
	  F -->|아니오| I
    F -->|예| H[서브 앱 :CORS 미들웨어]
    H --> I[기본 앱 : CORS 미들웨어 요청 처리]
    I --> O[라우트 핸들러]
    O --> P[기본 앱 : CORS 미들웨어 응답 처리]
    P --> Q{응답이 서브 앱에서 왔는가?}
    Q --> |예|R[서브 앱 : CORS 미들웨어 응답 처리]
    Q --> |아니오| T
    R --> T[기본 앱 : CORS 미들웨어 응답 처리]
    T --> W[기본 앱 : Logging 미들웨어 응답 처리]
    W --> X[기본 앱 : Metircs 미들웨어 응답 처리]

CORSMiddleware 플로우 차트를 그려봤다. 제가 생각한 Middleware 동작과 코드에서 동작하는 Middleware 동작은 많이 달랐다.

문제 개선

1. CORSMiddleware 단일로 처리

CORSMiddleware가 여러 개 붙어도 결국에는 기본 앱인 app의 CORSMiddleware정책을 따르게 된다.

metrics_app의 CORSMiddleware를 삭제한다.

# ./test_main.py

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://main.co.kr"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
app.add_middleware(LoggiMiddleware)
app.add_middleware(MetricsMiddleware)

class CORSMiddlewareTest(CORSMiddleware): ...

metrics_app = FastAPI()

# 삭제한 부분
~~metrics_app .add_middleware(
    CORSMiddleware,
    allow_origins=["http://metrics.co.kr"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)~~

app.mount("/metrics", metrics_app)

@metrics_app.get("/")
def get_prometheus_metrics():
    return "Test"

2. Mount → API Router로 개선

# ./test_main.py

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://main.co.kr"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
app.add_middleware(LoggiMiddleware)
app.add_middleware(MetricsMiddleware)

@metrics_app.get("/metrics")
def get_prometheus_metrics():
    return "Test"

결국에는 ~/metrics을 사용하는 거니 APIRouter.get(”/metrics”)로 수정한다. 하지만 저처럼
~/metrics 에 특별한 정책을 사용하고 싶다면 후에 나오는 APIRouter.middleware 기능을 사용하면된다.(아직은 논의 중이다. 링크 : https://github.com/fastapi/fastapi/discussions/7691)

profile
째하
한 걸음씩 나아가는 개발자
이전 포스트

FCM 서버 성능 개선 과정(+FastAPI, Python)

다음 포스트

AWS ECR 비용 최적화 해보기(Python 스크립트를 곁들인)

0개의 댓글