スポンサーリンク
概要
最近FastAPIを触り始めたのですが、Dependsで関数ではなくクラスインスタンスを注入するための個人的なメモです。
スポンサーリンク
説明
Dependsとは
簡単に言ってしまうと、FastAPI専用のD/I(依存性注入)の仕組みです。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
from typing import Union from fastapi import Depends, FastAPI app = FastAPI() # 共通で呼ばれる関数 async def common_parameters( q: Union[str, None] = None, skip: int = 0, limit: int = 100 ): return {"q": q, "skip": skip, "limit": limit} @app.get("/items/") async def read_items(commons: dict = Depends(common_parameters)): # 関数実行前にcommon_parametersが呼ばれて、dictが作られる return commons @app.get("/users/") async def read_users(commons: dict = Depends(common_parameters)): return commons |
実際に各HTTPメソッドが呼び出される前に、Dependsに指定した関数を実行し、その戻り値を引数として渡すことができます。
よく挙げられる使用用途としては、下記になります。
- UnitTestでスタブやモックなどの処理に置き換えが可能
- 前処理の共通化 (ヘッダの特定のパラメータチェック、ユーザーの認証チェックなど)
- OpenAPIのドキュメントのスキーマなどに統合
2に関しては、他のWebフレームワークでもミドルウェア機能として提供されていて、リクエストの前後に処理を挟むために使われることが多いかと思います。 (ただしFastAPIにもミドルウェア機能はあり。)
より詳細な情報は公式ページの説明を参照してください。
個人的に感じた点
1つのコードで説明するような簡単なサンプルでは、先に挙げた関数を直接渡してしまえばいいですが、下記の点で少し不便(使いにくい)と感じました。
- ルーティングの定義が書かれたコード内に、特定のロジックを持つ関数を書くのは何か気持ち悪い
- 固定処理じゃ無い場合には、関数のスコープ外に設定などを持たせないといけない
そこで関数ではなく、クラス自体のインスタンスを渡せないのかと思いました。そうすることで、コンストラクタで設定を渡すことでクラス内に各種設定なども持たせることができます。
クラスによる依存性注入
公式の解説にも書いてありますが、クラスを使った依存性の注入も可能になっています。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 |
from typing import Union from fastapi import Depends, FastAPI app = FastAPI() fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}] class CommonQueryParams: def __init__(self, q: Union[str, None] = None, skip: int = 0, limit: int = 100): self.q = q self.skip = skip self.limit = limit @app.get("/items/") async def read_items(commons=Depends(CommonQueryParams)): response = {} if commons.q: response.update({"q": commons.q}) items = fake_items_db[commons.skip : commons.skip + commons.limit] response.update({"items": items}) return response |
ただ、これだと実際にはコンストラクタが呼ばれるだけなので、初期化処理で何か設定を持たせるとなると毎回処理が実行されてしまいます。
コンストラクタ自体はモジュール内の初期化処理などで、別途呼び出しを行い、設定などを保持させて、それらの設定を使って別途処理をしたいので目的に合いません。
もう少し探ってみたところ、セキュリティの項目で使用されている、OAuth2PasswordBearerがまさにやりたいことを実現しているので実際のコードを覗いてみたところ、下記のように実行していました。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 |
class OAuth2PasswordBearer(OAuth2): def __init__( self, tokenUrl: str, scheme_name: Optional[str] = None, scopes: Optional[Dict[str, str]] = None, description: Optional[str] = None, auto_error: bool = True, ): if not scopes: scopes = {} flows = OAuthFlowsModel(password={"tokenUrl": tokenUrl, "scopes": scopes}) super().__init__( flows=flows, scheme_name=scheme_name, description=description, auto_error=auto_error, ) async def __call__(self, request: Request) -> Optional[str]: authorization = request.headers.get("Authorization") scheme, param = get_authorization_scheme_param(authorization) if not authorization or scheme.lower() != "bearer": if self.auto_error: raise HTTPException( status_code=HTTP_401_UNAUTHORIZED, detail="Not authenticated", headers={"WWW-Authenticate": "Bearer"}, ) else: return None return param |
ということで、__call__関数を定義して、呼ぶことができるようです。
個人的にはなじみの無い関数なのですが、クラスのインスタンスをメソッドとして呼び出した際に呼ばれる機能のようですね。
1 2 3 4 5 6 |
class Sample: def __call__(self): print("Hello") s = Sample() s() # __call__が呼ばれる |
下記のように__call__関数を記載して、別のスコープで作成したクラスインスタンス自体をDependsに渡すと、この__call__がリクエストの度に実行されてやりたいことが実現できまました。
コンストラクタで固定の設定を持たせた上で、特定の処理を実行させたいということが実現できました。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
from typing import Union from fastapi import Depends, FastAPI app = FastAPI() class CommonParam: p1: str def __init__(self, p1: str): self.p1 = p1 async def __call__(self) -> str: return p1 # クラスをインスタンスとして保持 ("test1"を設定) common_param = CommonParam("test1") @app.get("/items/") async def read_items(param1: str = Depends(common_param)): return param1 # param1には"test1"が入っている |