FastAPI 구조 살펴보기 세 번째 글입니다.
https://ceb10n.medium.com/understanding-fastapi-how-fastapi-works-6df7a793fefb
오늘은 이 글을 참고하였습니다.
지난 두 글에서 ASGI 서버와 애플리케이션이 서로 통신하는 방식과 FastAPI의 기초인 Starllete가 작동하는 방식을 살펴보았습니다.
이제 FastAPI가 Starllete를 어떻게 사용하는지 자세히 살펴보겠습니다.
FastAPI, a Starllete app
우선, FastAPI의 작동 방식을 이해하기 위한 두 가지 주요 정보 소스는 다음과 같습니다.
따라서 FastAPI 저장소를 clone하고 살펴봅시다.
FastAPI의 첫 번째 진입점은 fastapi/applications.py에 있는 FastAPI 클래스입니다.
class FastAPI(Starlette):FastAPI도 다른 Starlette 클래스를 상속받아 정의합니다. 그러므로 Starlette과 마찬가지로 scope, receive, send를 받는 호출 가능한 객체라는 것을 예상할 수 있습니다.
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)FastAPI에는 예상한 대로 call 함수가 있을 뿐만 아니라 요청을 Starlette에 위임한다는 것을 알 수 있습니다.
초기화 시 FastAPI와 Starlette의 차이점
FastAPI가 Starlette을 상속받아 사용함으로, 초기화 중에 일부 기능을 추가할 가능성이 높습니다.
FastAPI의 __init__ 함수를 보면 두 가지 주요 내용을 볼 수 있습니다.
setup함수에서 OpenAPI 문서에 대한 라우트를 추가합니다.Router를APIRouter로 설정합니다.
setup은 FastAPI의 가장 멋진 기능 중 하나를 추가합니다. Swagger 및 Redoc을 사용하여 프로젝트에 무료 OpenAPI 문서를 추가합니다.
APIRouter는 모든 경로 작업이 존재하는 곳입니다. APIRouter를 생성하거나 FastAPI 앱을 사용하여 경로를 직접 추가하면 모든 경로가 FastAPI의 라우터에 포함됩니다.
이번 포스팅에서는 FastAPI의 라우터와 경로를 더 자세히 살펴보겠습니다. OpenAPI에 대해서는 다음에 얘기해보도록 하겠습니다.
Request 생명주기
FastAPI는 추가 기능을 갖춘 Starlette 앱이므로 FastAPI를 사용하는 요청 수명 주기가 Starlette의 요청 수명 주기와 거의 동일하다고 가정할 수 있습니다.
이전 글에서 얘기 했듯이 미들웨어 체인은 다음과 같습니다.
-> ServerErrorMiddleware
-> Other Middlewares
-> ExceptionMiddleware
-> RouterFastAPI로 작업할 때 자체 APIRouter로 Starlette의 Router를 재정의하는 것을 볼 수 있습니다.
즉, FastAPI가 여전히 Starlette의 생명 주기에 의존하고 있지만, 요청 처리는 자체적인 방식으로 수행하는 것을 선호한다는 것을 알 수 있습니다.
FastAPI를 사용하면 다음과 같은 이점을 얻을 수 있습니다.
-> FastAPI App
-> Starlette's App
-> Starlette's ServerErrorMiddleware
-> Starlette's ExceptionMiddleware
-> FastAPI's APIRouter (and Router, since it don't override Router's __call__)FastAPI 라우터 및 경로
FastAPI 앱을 만들 때 경로를 추가하는 두 가지 주요 방법이 있습니다.
- FastAPI의 인스턴스를 사용하여 직접 경로 추가:
app = FastAPI()
@app.get("/{name}")
async def hi(name: str):
return {"hi": name}- 일반적으로 대규모 앱에서 사용되는
APIRouter를 사용합니다.
app = FastAPI()
router = APIRouter(prefix="/v1")
@router.get("/compliments/{name}")
async def hi1(name: str):
return {"hi": name}
app.include_router(router)FastAPI가 어떻게 작동하는지 이해하려고 노력 중이므로 @app.{verb}를 사용할 때 무슨 일이 일어나는지 살펴보겠습니다.
def get(
self,
path: Annotated[
str,
Doc("... # docs here"),
],
*,
... # other args here
) -> Callable[[DecoratedCallable], DecoratedCallable]:
return self.router.get(
path,
... # code continues
)여기서 볼 수 있는 것은 FastAPI.{get,put,post,etc}는 APIRouter에 대한 경로를 포함하는 단순한 데코레이터라는 것입니다.
FastAPI.include_router는 어떻습니까?
FastAPI의 include_router 함수는 단순히 자체 APIRouter의 include_router를 호출합니다. 이는 기본적으로 APIRouter에 포함된 모든 경로를 반복하고 경로를 추가합니다.
# FastAPI include_router
def include_router(
self,
router: Annotated[routing.APIRouter, Doc("The `APIRouter` to include.")],
*,
... # other args
) -> None:
self.router.include_router(
router,
... # other args
)
# APIRouter include_router
def include_router(
self,
router: Annotated["APIRouter", Doc("The `APIRouter` to include.")],
... # other args
) -> None:
for route in router.routes:
if isinstance(route, APIRoute):
... # some logic here
self.add_api_route(
prefix + route.path,
route.endpoint,
... # other args
)APIRouter.include_router를 보면 Starlette의 경로, APIWebSocketRoute 등과 같은 다른 유형의 경로를 처리하는 것을 볼 수 있습니다.
그리고 내 경로 기능이 호출되면?
요청을 받으면 APIRouter가 __call__을 재정의하지 않기 때문에 Starlette의 라우터가 호출됩니다. 일치하는 경로를 찾으면 해당 경로의 핸들 함수를 호출합니다.
handle도 덮어쓰지 않으므로 Route에 속합니다. APIRoute가 하는 일은 Route의 app을 Starlette의 함수 request_response로 설정하고 APIRoute의 get_route_handler를 매개변수로 받는 것입니다.
class APIRoute(routing.Route):
def __init__(
self,
path: str,
endpoint: Callable[..., Any],
*,
... # other args
) -> None:
... # some logic here
self.app = request_response(self.get_route_handler())get_route_handler는 get_request_handler 함수를 반환합니다. 여기에서 종속 항목, pydantic 모델 등이 포함된 FastAPI 경로에 대한 Starlette 요청의 "변환"을 보기 시작합니다.
run_endpoint_function 함수를 실행합니다. 그리고 여기가 모든 해결된 종속성, pydantic 모델 등과 함께 경로 함수가 호출되는 곳입니다.
def get_request_handler(
... # args
) -> Callable[[Request], Coroutine[Any, Any, Response]]:
# logic here
async def app(request: Request) -> Response:
response: Union[Response, None] = None
async with AsyncExitStack() as file_stack:
# logic here
errors: List[Any] = []
async with AsyncExitStack() as async_exit_stack:
# logic here
if not errors:
raw_response = await run_endpoint_function(
dependant=dependant, values=values, is_coroutine=is_coroutine
)사용 중인 프레임워크가 코드를 올바르게 처리하는 방법을 살펴 보았습니다.
