FastAPI 핵심 탐구: Starlette 깊이 파헤쳐보기
FastAPI (파스타피아이)는 사실상 Starlette (스타레테)를 감싼 API (응용 프로그래밍 인터페이스) 래퍼(wrapper)입니다.
FastAPI (파스타피아이)를 제대로 이해하려면 먼저 Starlette (스타레테)부터 알아야 하는데요.
ASGI (에이지에스아이) 프로토콜
Uvicorn (유비콘)은 공통 인터페이스인 ASGI (에이지에스아이) 프로토콜을 통해 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")
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 (라이프스팬) 함수
"""
# 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 (앱) 안에 다른 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 (앱)이 되도록 설계되었다는 것을 알 수 있습니다.
미들웨어
위에서 언급했듯이 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에는 이 요구 사항을 충족하는 많은 미들웨어 구현이 있습니다.
그러나 이 장에서는 모든 미들웨어를 다루지는 않고, 경로에서 가장 가까운 것부터 가장 먼 것까지 몇 가지 대표적인 미들웨어만 분석할 것입니다.
예외 처리 미들웨어 - 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 (에이지에스아이) call 메서드
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)
사용자 미들웨어
다음은 사용자 미들웨어입니다.
우리가 가장 많이 접하게 되는 미들웨어인데요.
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:
# Before (전처리)
response: Response = await call_next(request)
# After (후처리)
return response
요청 전처리를 수행하고 싶다면 before (전처리) 블록에 관련 코드를 작성하고, 요청 후처리를 하고 싶다면 after (후처리) 블록에 코드를 작성하면 됩니다.
사용법이 매우 간단하고, 동일한 scope (스코프) 내에 있기 때문에 이 메서드의 변수를 context (컨텍스트) 또는 동적 변수를 통해 전파할 필요가 없습니다.
(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 (에이지에스아이) 표준 함수 서명 (signature)이 있는 함수로, 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
# uvicorn (유비콘)의 receive (리시브) 객체를 request.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()를 호출하여 코루틴에 예외가 있으면 코루틴의 오류가 throw (스로우)됩니다.
task.result()
# 예외가 throw (스로우)되지 않으면 사용자 오류로 인해 응답이 비어 있을 수 있습니다.
# 이때 클라이언트에 응답을 반환할 수 없으므로 후속 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 (리스폰스) 메서드에 넣습니다.
# Response (리스폰스) 자체도 ASGI (에이지에스아이) APP (앱)과 유사한 클래스입니다. According (어코딩)
ServerErrorMiddleware (서버에러미들웨어)
ServerErrorMiddleware (서버에러미들웨어)는 ExceptionMiddleware (익셉션미들웨어)와 매우 유사합니다.
(그래서 이 부분은 자세히 설명하지 않겠습니다.) 전반적인 로직은 거의 동일합니다.
그러나 ExceptionMiddleware (익셉션미들웨어)는 라우팅 예외를 캡처하고 처리하는 역할을 하는 반면, ServerErrorMiddleware (서버에러미들웨어)는 항상 합법적인 HTTP (에이치티티피) 응답이 반환되도록 하는 폴백 (fallback) (대체) 조치 역할을 주로 합니다.
ServerErrorMiddleware (서버에러미들웨어)의 간접 호출 함수는 ExceptionMiddleware (익셉션미들웨어)와 동일합니다.
하지만 등록된 HTTP (에이치티티피) 상태 코드가 500인 경우에만 콜백이 ServerErrorMiddleware (서버에러미들웨어)에 등록됩니다.
@app.exception_handlers(500)
def not_found_route() -> None: passServerErrorMiddleware (서버에러미들웨어)는 ASGI (에이지에스아이) APP (앱)의 최상위 레벨에 있습니다.
폴백 (fallback) (대체) 예외를 처리하는 역할을 담당합니다.
해야 할 일은 간단합니다.
다음 레벨 ASGI (에이지에스아이) APP (앱) 처리 중에 예외가 발생하면 폴백 (fallback) (대체) 로직으로 진입합니다.
- 디버그가 활성화된 경우 디버그 페이지를 반환합니다.
- 등록된 콜백이 있는 경우 등록된 콜백을 실행합니다.
- 위의 둘 다 해당되지 않으면 500 응답을 반환합니다.
경로 (Route)
Starlette (스타레테)에서 경로는 두 부분으로 나뉩니다.
하나는 제가 Real App (리얼 앱)의 Router (라우터)라고 부르는 부분으로, 미들웨어 바로 아래 레벨에 있습니다.
주로 경로 조회 및 매칭, 애플리케이션 시작 및 종료 처리 등 미들웨어를 제외한 Starlette (스타레테)의 거의 모든 것을 담당합니다.
다른 하나는 Router (라우터)에 등록된 경로로 구성됩니다.
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 (스타레테) 애플리케이션 내부에서 실행 중인 경우 예외를 throw (스로우)하여 구성 가능한 예외 처리기가 응답 반환을 처리할 수 있도록 합니다. 일반 ASGI (에이지에스아이) 앱의 경우 응답만 반환합니다.
if "app" in scope:
# starlette.applications의 __call__ 메서드에서 Starlette (스타레테)가 자체적으로 scope (스코프)에 저장하는 것을 볼 수 있습니다.
# 여기서 예외를 throw (스로우)한 후 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 (에이지에스아이) lifespan (라이프스팬) 메시지를 처리합니다.
"""
# Lifespan (라이프스팬) 실행 로직. 실행 시 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 (컨텍스트)가 여러 번 yield (yield)되었습니다."
first = False
await send({"type": "lifespan.startup.complete"})
await receive()
else:
for item in self.lifespan_context(app): # type: ignore
assert first, "Lifespan (라이프스팬) context (컨텍스트)가 여러 번 yield (yield)되었습니다."
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":
# 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 tree (경로 트리)를 구현하여 대체했습니다.
(자세한 내용은 route_trie.py 참조) 하지만 성능 테스트를 해보니 경로 수가 50개를 넘지 않으면 루프 매칭 성능이 경로 트리보다 더 나은 것을 확인했습니다.
수가 100개를 넘지 않으면 둘은 비슷합니다.
그리고 일반적인 상황에서 우리가 지정하는 경로 수는 100개를 넘지 않을 것입니다.
따라서 이 경로 부분의 매칭 성능에 대해 걱정할 필요는 없습니다.
여전히 걱정된다면 Mount (마운트)를 사용하여 경로를 그룹화하여 매칭 수를 줄일 수 있습니다.
기타 경로
Mount (마운트)는 BaseRoute를 상속하고, HostRoute, WebSocketRoute와 같은 다른 경로도 마찬가지입니다.
이들은 유사한 메서드를 제공하며, 구현에 약간의 차이만 있습니다.
(주로 초기화, 경로 매칭, 역방향 조회에서 차이가 있습니다.) 먼저 BaseRoute를 살펴볼까요?
class BaseRoute:
def matches(self, scope: Scope) -> typing.Tuple[Match, Scope]:
# 표준 매칭 함수 서명 (signature). 각 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는 많은 기능을 제공하지 않으며, 다른 경로는 BaseRoute를 기반으로 확장된다는 것을 알 수 있습니다.
- Route (경로): 표준 HTTP (에이치티티피) 경로입니다. HTTP (에이치티티피) URL (유알엘) 및 HTTP (에이치티티피) 메서드를 통해 경로 매칭을 담당한 다음 HTTP (에이치티티피) 경로를 호출하는 메서드를 제공합니다.
- WebSocketRoute (웹소켓경로): 표준 WebSocket (웹소켓) 경로입니다. HTTP (에이치티티피) URL (유알엘)에 따라 경로를 매칭하고,
starlette.websocket의 WebSocket (웹소켓)을 통해 세션을 생성하여 해당 함수에 전달합니다. - Mount (마운트): 경로의 중첩 캡슐화입니다.
'Python' 카테고리의 다른 글
| Python tile-tools 시작하기: Mapbox 타일 작업, 이제 어렵지 않아요! (1) | 2025.03.22 |
|---|---|
| 파이썬 Switch 문, 이제 이렇게 쓰세요! (2025년 최신 가이드) - Switch Case 완벽 예제 (0) | 2025.03.22 |
| 파이썬 비동기 프로그래밍, 코루틴의 모든 것 (0) | 2025.03.22 |
| 파이썬 Garbage Collection 완벽 분석: 개발자가 반드시 알아야 할 모든 것 (0) | 2025.03.19 |
| FastAPI + Uvicorn: 엄청난 속도의 기술, 그 뒷이야기를 알아볼까요? (0) | 2025.03.19 |