일러두기
- api function(controller)을 python 웹개발 쪽에서 view라는 명칭으로 부르는 경우가 많아서 view = “path operation function” 로 상정하고 글을 작성 했습니다.
- FastAPI 0.95.0 부터 Type Annotation에 Annotated 가 도입되어 공식문서에 도입을 장려하고 있습니다. 이 글도 Annotated 기반으로 작성합니다. 자세한 내용은 링크 를 참조하세요.
Django 와 Flask에는 없는 것
django 에서 view param으로 넘기는 종류는 크게 4개 입니다.
•
request
•
path param
•
kwargs from path() or re_path()
•
default view param
flask에서 view param으로 전달 하는 인자의 종류는 2가지 입니다.
•
path parameter
•
default view parameter
fastapi는 7 종의 인자를 view로 전달 할 수 있습니다.
•
request
•
path param
•
query param
•
header param
•
cookie param
•
request body param
•
default view param
•
Dependency Injector
위의 7가지 param 중에서 Dependency Injector를 제외한 나머지는 request 관련 데이터를 함수인자로 전달 하는지, request 객체로 전달하는지에서 비롯된 param 입니다. django 나 flask에서는 함수인자로 전달 하지 않고 다른 방법을 통해 사용합니다.
Dependency Injector 는 django와 flask에서는 보지 못했던 종류입니다. 공식문서에는 아래와 같이 설명 합니다.
FastAPI has a very powerful but intuitive Dependency Injection system.
It is designed to be very simple to use, and to make it very easy for any developer to integrate other components with FastAPI.
You don't call it directly (don't add the parenthesis at the end), you just pass it as a parameter to Depends().
And that function takes parameters in the same way that path operation functions do.
TLDR; 강력하고 사용편한 직관의 콤포넌트 병합 시스템입니다.
Depends() 는 크게 2가지 역할을 합니다.
1.
view에 전달되는 인자들이 동일하게 Depends() 에 전달한 Callable 객체에 전달 되게 합니다.
a.
view param 처리 로직을 공통 로직으로 추출해 낼수 있게 됩니다.
from typing import Annotated
from fastapi import FastAPI, Header, Cookie, Query, Depends
from pydantic import BaseModel
class Item(BaseModel):
title: str
content: str
app = FastAPI()
# Before Depends
@app.post("/hello-world/{path}")
async def sample_view(
path: int,
body: Item,
query: Annotated[int, Query()],
header: Annotated[str, Header()],
cookie: Annotated[str, Cookie()]
):
...
return {"message": "welcome"}
@app.post("/hello-world2/{path}")
async def sample_view2(
path: int,
body: Item,
query: Annotated[int, Query()],
header: Annotated[str, Header()],
cookie: Annotated[str, Cookie()]
):
...
return {"message": "welcome"}
Python
복사
# After Depends
async def common_parameters(
query: Annotated[int, Query()],
header: Annotated[str, Header()],
cookie: Annotated[str, Cookie()]
):
return {"query": query, "header": header, "header": header}
async def common_parameters2(
path: int,
body: Item
query: Annotated[int, Query()],
header: Annotated[str, Header()],
cookie: Annotated[str, Cookie()]
):
return {"query": query, "header": header, "header": header}
# Notice: path param and body param can be used in depends callable,
# but both of them must declared as view args.
# If path param is missed, fastapi can not find view.
# If body param is missed, fastapi can not get body and raise 422 error.
@app.post("/hello-world3/{path}")
async def sample_view3(
path: int,
body: Item,
params: Annotated[dict, Depends(common_parameters)]
):
...
return {"message": "welcome"}
@app.post("/hello-world4/{path}")
async def sample_view4(
path: int,
body: Item,
params: Annotated[dict, Depends(common_parameters2)]
):
...
return {"message": "welcome"}
Python
복사
b. view가 동작하기전에 request 요소를 검증 할 수 있습니다. ( view 스코프가 아닌, router 스코프(view 집합), Global App 스코프에도 스코프별로 각각 dependencies 를 설정할 수 있습니다.)
Dependencies in path operation decorators
In some cases you don't really need the return value of a dependency inside your path operation function. Or the dependency doesn't return a value.
But you still need it to be executed/solved.
For those cases, instead of declaring a path operation function parameter with Depends, you can add a list of dependencies to the path operation decorator.
async def verify_token(x_token: Annotated[str, Header()]):
if x_token != "fake-super-secret-token":
raise HTTPException(status_code=400, detail="X-Token header invalid")
async def verify_key(x_key: Annotated[str, Header()]):
if x_key != "fake-super-secret-key":
raise HTTPException(status_code=400, detail="X-Key header invalid")
return x_key
# App Scope Dependencies: 모든 요청에 대해 Depends를 실행합니다.
app = FastAPI(dependencies=[Depends(verify_key)])
@app.post(
"/hello-world5/{path}",
# View Scope Dependencies: View로 접수되는 Request에 대해서만 Depends를 실행합니다.
dependencies=[Depends(verify_token)]
)
async def sample_view5(
path: int,
body: Item,
params: Annotated[dict, Depends(common_parameters2)]
):
...
return {"message": "welcome"}
Python
복사
2.
view에서 필요한 콤포넌트들을 호출 합니다.
•
view에서 필요한 콤포넌트 호출
•
view에서 필요한 자원들 호출 및 클로징 작업( with yield )
# call component
async def get_some_client():
return SomeClient()
# call resource requiring close()
async def get_db():
db = DBSession()
try:
yield db
# code below this is done after response is sent to client
# so adding any try, catch statement here will not change response.
finally:
db.close()
# call instance()
class Resource:
def __init__(self, a: str, b: str):
self.a = a
self.b = b
def __call__(self) -> str:
return self.a + "-" + self.b
resource = Resource("another", "client")
@app.post(
"/hello-world6/{path}",
dependencies=[Depends(verify_token)]
)
async def sample_view6(
path: int,
body: Item,
some_client: Annotated[SomeClient, Depends(get_db)],
db: Annotated[DBSession, Depends(get_db)],
resource: Annotated[str, Depends(resource)],
bg_task: BackgroundTasks
):
# do SomeClient and Resource jobs
...
# do DB jobs
...
# register backgrounds Job with DB
bg_task.add_task(log_body_after_response, db, body=body)
return {"message": "welcome"}
Python
복사
view에 @데코레이터를 추가하는 것으로 동일 기능을 할 수 있다고 생각 할 수 있습니다. 하지만 view 데코레이터의 유효 스코프는 view 실행 전, 후 뿐입니다. response가 끝나고, BackgroundTasks 까지 context 유지가 되지 않습니다. 이에 반해 Depends() 는 request-BackgroundTasks 까지가 유효스코프입니다. 리퀘스트가 처리된 뒤의 Background에서도 해당 resource를 사용 할 수 있습니다.
This is what allows anything set in the dependency (e.g. a DB session) to, for example, be used by background tasks.
Origin
•
view의 args가 아닌 곳에서 Depends 사용은 불가능합니다.
# python console
def get_three():
return 3
def get_val(id: int = Depends(get_three)):
return id
>>> get_val()
Depends(get_three)
Python
복사
# view
@app.post(
"/hello-world7/{path}"
)
async def sample_view7(
path: int,
body: Item,
):
three = get_val() # three = Depends(get_three)
...
return {"message": "welcome"}
Python
복사
공식문서에 제가 설명한 기본 내용 외에도 Depends 에 관한 내용을 잘 설명해주고 있으니 한번 읽어보시면 좋겠습니다.
•
Classes as Dependencies
•
Sub-dependencies
•
Global Dependencies
•
Testing Dependencies with Overrides
공식문서에서 설명하지 않는 부분을 다뤄보도록 하겠습니다.
Depends 는 어디서 어떻게 동작할까요?
# fastapi/routing.py
...
def get_request_handler(...)
...
async def app(request: Request) -> Response:
...
# function name이 solve_dependencies 입니다. 설명이 필요 없는 직관 이름 입니다.
🟠 solved_result = await solve_dependencies(
# <starlette.requests.Request object at 0x107f0af10>
request=request,
# dependant는 FastAPI App내 Depends를 가지고 있는 객체 입니다.
# <fastapi.dependencies.models.Dependant object at 0x107e7b710>
dependant=dependant,
# content of body
body=body,
# <fastapi.applications.FastAPI object at 0x106f6d850>
# dependency 오버라이드 데이터를 가져올 객체입니다.
dependency_overrides_provider=dependency_overrides_provider,
)
# solved_result 마지막 결과값은 dependency_cache입니다.
# `solve_dependencies` 내포함수에서 사용하는 값이므로 무시합니다.
values, errors, background_tasks, sub_response, _ = solved_result
Python
복사
* dependant 와 dependencies의 차이
dependant는 view 관련 정보를 들고 있는 객체입니다.
e.g.) path_params, query_params, body_params, dependencies
dependencies는 dependant의 attr로, view와 연관된 Depends 들을 들고 있는 list 입니다.
solve_dependencies 를 살펴봅니다.
# .../fastapi/dependencies/utils.py
async def solve_dependencies(
*,
request: Union[Request, WebSocket],
dependant: Dependant,
body: Optional[Union[Dict[str, Any], FormData]] = None,
background_tasks: Optional[BackgroundTasks] = None,
response: Optional[Response] = None,
dependency_overrides_provider: Optional[Any] = None,
dependency_cache: Optional[Dict[Tuple[Callable[..., Any], Tuple[str]], Any]] = None,
) -> Tuple[
Dict[str, Any],
List[ErrorWrapper],
Optional[BackgroundTasks],
Response,
Dict[Tuple[Callable[..., Any], Tuple[str]], Any],
]:
...
# dependant의 dependencies 에 등록된 Depends()에 대해 for loop를 돌립니다.
for sub_dependant in dependant.dependencies:
...
# `solve_dependencies` recursive가 적용되어 있습니다.
# Depends chaining을 여기서 푸는걸 알수 있습니다.
solved_result = await solve_dependencies(
request=request,
dependant=use_sub_dependant,
body=body,
background_tasks=background_tasks,
response=response,
dependency_overrides_provider=dependency_overrides_provider,
dependency_cache=dependency_cache,
)
(
sub_values,
sub_errors,
background_tasks,
_, # the subdependency returns the same response we have
sub_dependency_cache,
) = solved_result
...
if sub_errors:
errors.extend(sub_errors)
continue
if sub_dependant.use_cache and sub_dependant.cache_key in dependency_cache:
# 캐쉬된 Depends가 있으면 여기서 값을 꺼냅니다.
solved = dependency_cache[sub_dependant.cache_key]
elif is_gen_callable(call) or is_async_gen_callable(call):
stack = request.scope.get("fastapi_astack")
assert isinstance(stack, AsyncExitStack)
# generator를 처리합니다.
solved = await solve_generator(
call=call, stack=stack, sub_values=sub_values
)
elif is_coroutine_callable(call):
# async callable를 처리합니다.
solved = await call(**sub_values)
else:
# sync callable를 처리합니다.
solved = await run_in_threadpool(call, **sub_values)
if sub_dependant.name is not None:
values[sub_dependant.name] = solved
if sub_dependant.cache_key not in dependency_cache:
dependency_cache[sub_dependant.cache_key] = solved
...
return values, errors, background_tasks, response, dependency_cache
Python
복사
실행 하는 코드는 찾았는데, 내가 작성한 Depends()는 언제, 어떻게 찾게 될지 찾아봅니다.
Depends 예시 3개 를 만들어 놓고, pycharm 디버깅을 통해 흐름을 따라가 보겠습니다.
# make sample dependable func
# main.py
app = FastAPI()
async def get_3():
🔴 a = 1
b = 2
return a + b
def get_5():
🔴 a = 2
b = 3
return a + b
async def get_7():
try:
🔴 a = 3
b = 4
yield a + b
finally:
print("get_7 close")
@app.get("/api")
async def api_test(
var3: Annotated[int, Depends(get_3)],
var5: Annotated[int, Depends(get_5)],
var7: Annotated[int, Depends(get_7)]
):
return {"message": sum([var3, var5, var7])}
Python
복사
/api 실행 시점에 Depends 를 처리하는 로직은 보이지 않고, 함수 내부 로직이 실행되는 것만 걸립니다. Depends 읽고 FastAPI Dependant 인스턴스에 등록하는 것은 View 실행 시점이 아닌, FastAPI App loading 시점입니다. 디버깅 포인트를 잡고, 디버깅을 시작하면 이미 코드를 읽었기 때문에 걸리지 않습니다.
FastAPI Dependant에 등록을 어디서 하는지 @app.get(…) 을 따라가 보겠습니다.
* Router 와 Route의 차이
Route는 개별 endpoint func를 지칭합니다.
Router는 Route의 집합 입니다.
# main.py
🟠@app.get("/api")
...
# .../fastapi/application.py
class FastAPI(Starlette):
...
def __init__(self,...):
self.router: routing.APIRouter = routing.APIRouter(...)
def get(self, ...):
🟠 return self.router.get(...)
# .../fastapi/rounting.py
class APIRouter(routing.Router):
...
def get(self, ...):
🟠 return self.api_route(...)
def api_route(self, ...):
def decorator(func: DecoratedCallable) -> DecoratedCallable:
🟠 self.add_api_route(..., func, ...)
# view를 @decorate 하는데, *args, **kwargs는 func에 전달하지 않습니다.
# func args, kwargs 들은 add_api_route로 뽑아내서
# request 들어올때 func에 설정한 type에 맞춰 처리한 다음 func에 전달합니다.
return func
return decorator
def add_api_route(self, ...,func, ...):
# route_class = fastapi.routing.APIRoute
# 새로운 route 인스턴스를 만들고, routes에 추가합니다.
🟠 route = route_class(...)
self.routes.append(route)
Python
복사
# .../fastapi/rounting.py
class APIRoute(routing.Route):
🟠 def __init__(...):
...
# endpoint == view
# endpoint가 callable 인지 확인 합니다.
assert callable(endpoint), "An endpoint must be a callable"
# view 정보를 바탕으로 route의 dependant를 생성합니다.
🟠 self.dependant = get_dependant(path=self.path_format, call=self.endpoint)
# view path decorator 에 정의한 dependencies를 dependant.dependencies에 등록합니다.
for depends in self.dependencies[::-1]:
self.dependant.dependencies.insert(
0,
get_parameterless_sub_dependant(depends=depends, path=self.path_format),
)
Python
복사
# .../fastapi/dependencies/utils.py
def get_dependant(
*,
path: str,
call: Callable[..., Any],
name: Optional[str] = None,
security_scopes: Optional[List[str]] = None,
use_cache: bool = True,
) -> Dependant:
path_param_names = get_path_param_names(path)
# view(endpoint)의 param(func signature) 를 Type 적용된 값으로 불러옵니다.
endpoint_signature = get_typed_signature(call)
# endpoint_signature
# <Signature (var3: typing.Annotated[int, Depends(get_3)], var5: typing.Annotated[int, Depends(get_5)], var7: typing.Annotated[int, Depends(get_7)])>
signature_params = endpoint_signature.parameters
# signature_params
# mappingproxy(OrderedDict([('var3', <Parameter "var3: typing.Annotated[int, Depends(get_3)]">), ('var5', <Parameter "var5: typing.Annotated[int, Depends(get_5)]">), ('var7', <Parameter "var7: typing.Annotated[int, Depends(get_7)]">)]))
dependant = Dependant(
call=call,
name=name,
path=path,
security_scopes=security_scopes,
use_cache=use_cache,
)
for param_name, param in signature_params.items():
is_path_param = param_name in path_param_names
# view param의 type, depends,
# 그리고 Depends가 적용된 FastAPI Field(Path, Query, Heaer...)를 리턴합니다.
🟠 type_annotation, depends, param_field = analyze_param(
param_name=param_name,
annotation=param.annotation,
value=param.default,
is_path_param=is_path_param,
)
# depends 가 있다면 sub-dependency를 추적합니다.
if depends is not None:
sub_dependant = get_param_sub_dependant(
param_name=param_name,
depends=depends,
path=path,
security_scopes=security_scopes,
)
dependant.dependencies.append(sub_dependant)
continue
# param_field가 있다면, dependant에 추가합니다.
....
Python
복사
# .../fastapi/dependencies/utils.py
def analyze_param(
*,
param_name: str,
annotation: Any,
value: Any,
is_path_param: bool,
) -> Tuple[Any, Optional[params.Depends], Optional[ModelField]]:
field_info = None
used_default_field_info = False
depends = None
type_annotation: Any = Any
"""
param annotation을 분석해서
type_annotation, Depends, FastAPI ModelField 를 리턴합니다.
param_name: view args 이름
annotation: typing.Annotated[int, Depends(get_3)]
value: view args default value
"""
# annotation이 Annotated 일 경우 처리합니다.
if (
annotation is not inspect.Signature.empty
and get_origin(annotation) is Annotated # type: ignore[comparison-overlap]
):
...
# fastapi_annotation: annotation에서 뽑아낸 view args annotation 입니다.
# fastapi_annotation이 FastAPI ModelField 일 경우 처리합니다.
if isinstance(fastapi_annotation, FieldInfo):
...
# fastapi_annotation이 Depends일 경우 처리합니다.
elif isinstance(fastapi_annotation, params.Depends):
depends = fastapi_annotation
elif annotation is not inspect.Signature.empty:
type_annotation = annotation
# default가 Depends일 경우 처리 합니다.
if isinstance(value, params.Depends):
...
depends = value
# default가 FastAPI ModelField 일 경우 처리 합니다.
elif isinstance(value, FieldInfo):
...
# 아래에서 좀더 설명합니다.
if depends is not None and depends.dependency is None:
depends.dependency = type_annotation
...
return type_annotation, depends, field
Python
복사
type hinting 만으로 어떻게 Depends 동작할까요?
commons: Annotated[CommonQueryParams, Depends(CommonQueryParams)]
↓↓↓↓↓
commons: Annotated[CommonQueryParams, Depends()]
Python
복사
# empty Depends() 등록 살피기
# depends가 있지만, dependency(Depends()에 등록하는 Callable) 가 없을 경우
# type_annotation을 dependency에 할당합니다.
if depends is not None and depends.dependency is None:
depends.dependency = type_annotation
Python
복사
어떻게 Backgrounds까지 Context가 유지 되는 걸까요?
제너레이터로 작성된 Depends() 의 경우, 별도의 스택(AsyncExitStack)에서 동작하게 됩니다. 동기코드는 스레드풀에서, 비동기코드는 비동기 매니저 안에서 동작합니다. 스택은 request 처리 사이클이 종료 될때 종료 합니다. 그래서 view가 끝나도 전체 response 사이클이 종료되기까지 context가 유지 됩니다.
# .../fastapi/dependencies/utils.py
async def solve_generator(
*, call: Callable[..., Any], stack: AsyncExitStack, sub_values: Dict[str, Any]
) -> Any:
if is_gen_callable(call):
cm = contextmanager_in_threadpool(contextmanager(call)(**sub_values))
elif is_async_gen_callable(call):
cm = asynccontextmanager(call)(**sub_values)
return await stack.enter_async_context(cm)
Python
복사
Depends() 파싱 순서, 동작 순서
dependencies 를 등록하는 순서는 app → router → route를 따라 내려가면서 등록합니다.
동작 순서는 solve_dependencies에서 dependencies에 등록된 순서대로 진행이되나 Depends 체이닝의 제일 끝까지 타고 내려가 올라오면서 실행합니다. use_cache=False 를 하지 않는다면, 동일 Depends() 의 결과가 있을 경우 재활용합니다.
Depends() 를 사용한 쿼리 파람 생성 방법 성능비교 (with wrk) 및 장단점 비교
Depends()를 사용해서 FastAPI Param을 클래스로 공통화 하는 3가지 방법이 있습니다.
1.
Python class
2.
pydantic BaseModel
3.
python dataclass
벤치 환경 설정은 다음과 같이하고, 로컬환경에서 테스트 하였습니다.
•
query 파람 100개씩 설정
•
wrk 사용: 스레드 10, 커넥션 100, 유지시간 30초 비교
아래와 같이 준비를 하고 테스트를 진행해봅니다.
int_query = Annotated[int, Query(..., gt=0)]
class Test1:
def __init__(
self,
q1: int_query,
...,
q100: int_query
):
q1 = q1
...
q100 = q100
class Test2(BaseModel):
q1: int_query
...
q100: int_query
@dataclass
class Test3:
q1: int_query
...
q100: int_query
# TypedDict로는 진행이 되지 않습니다.
# ValueError: no signature found for builtin type <class 'dict'>
class Test4(TypedDict):
q1: int_query
...
q100: int_query
@app.get("/test1")
async def test1(query: Annotated[Test1, Depends()]):
return {"message": query.q1}
@app.get("/test2")
async def test2(query: Annotated[Test2, Depends()]):
return {"message": query.q1}
@app.get("/test3")
async def test3(query: Annotated[Test3, Depends()]):
return {"message": query.q1}
Python
복사
Latency(AVG) | Req/Sec(AVG) | |
BasicClass | 127.99ms | 78.26 |
Pydantic BaseModel | 134.35ms | 74.66 |
DataClass | 149.29ms | 67.25 |
DataClass 대비, BasicClass가 약 10% 정도 빠른 latency를 보여주고 있습니다.
BasicClass 는 __init__ args와 attr 할당 2가지를 적어야 합니다. type annotation으로 단순하게 처리되는 다른 두 class에 비해 살짝 아쉽습니다. BaseModel 의 경우, 원하는 attr validator 생성이 method 생성으로 간편한 반면 BasicClass는 로직을 작성해줘야 합니다.
편의성을 생각하신다면 BaseModel, 성능이 우선이라면 BasicClass를 사용하시는게 좋아 보입니다.
마치며
•
Depends 사용성 좋고, 강력하고 쉬운데, 풀어내는 과정은 단순하지 않습니다.
•
타입을 지원하면서 동시에 타입의 지원으로 돌아가는 FastAPI 순환 고리가 계속 잘 굴러가면 좋겠습니다.
•
FastAPI 흥해라!
블로그 메인
