はじめに
PythonのDI(Dependency Injection)を実装するライブラリinjector を利用してSessionScopeをカスタム実装したのでメモします。
背景
PythonでFastAPIを利用した個人開発を行っていた際に、FastAPIでDependsというDI機能を利用していたのですが、どうもうまく運用できません。
理由1. 別のモックを注入するのに適した構造になっていない
- Stack Overflowにあったのですが、テストで違うモックをインジェクションするのでも面倒な印象です。
理由2. インジェクションするたびに毎度Depends()と書かないといけないので冗長
- 1リクエスト1インジェクションコードぐらいにしたい。
理由3. スコープなし
そこで、外部ライブラリでよさそうなDIが欲しいと思い、ライブラリを検索したところinjector DIライブラリがありました。
このinjector は使い勝手が良い感じですが、スコープのバリエーションが少ないです。
| プロトタイプスコープ（デフォルト） | DIコンテナが要求されるたびに新しいインスタンスを生成 |
| シングルトンスコープ | 一度作成したインスタンスをアプリケーション全体で共有 |
| ThreadLocalScope | 公式ドキュメントにはないがソースにはある スレッドごとにインスタンスを生成するらしい |
例えばDBを扱うアプリケーションを作成した場合はセッションごとにインスタンスを管理するためのSessionScopeがほしくなります。
公式ドキュメントによるとカスタムスコープを実装するにはScopeインターフェースを実装したカスタムクラスを作成するとのことです。
しかしSessionScopeを実装するにあたり、インターフェースだけではなくて、ライブラリのソースコードも変える必要が出てきました。
というわけでどう変えたも含めてSessionScopeを実装します。
環境
- Python 3.11.1
- injector 0.21.0
injectorライブラリの使い方
injectorを使ったことのない人向けに簡単な使い方を紹介。
インストール
下記コマンドでインストール
$ pip install injector==0.21.0
基本的なDI
ディレクトリ構造
│ ex_service.py
│ di.py
│ main.py
実装
# ex_service.py
# 抽象クラス
class Car(metaclass=ABCMeta):
@abstractmethod
def disp_car_type(self) -> None:
raise NotImplementedError
# 抽象クラス
class Color(metaclass=ABCMeta):
@abstractmethod
def disp_color(self) -> None:
raise NotImplementedError
# 抽象クラスを実装した具象クラス
class SuperCar(Car):
@inject
def __init__(self, color: Color): # ColorのインスタンスをDI
self.color = color
print("car初期化")
def disp_car_type(self):
# 色も表示する
self.color.disp_color()
print("car type is super car")
# シングルトンスコープで実装
@singleton
class Yellow(Color):
def __init__(self):
print("yellow初期化")
def disp_color(self):
print("color is yellow")
@inject
@dataclass
class ExService:
car: Car
def action(self) -> None:
self.car.disp_car_type()
- インジェクトするクラスやメソッドに@injectをつけるだけでよい
- プロトタイプスコープ以外のスコープを使用する場合、例えばシングルトンスコープの場合はクラスに@singletonをつける。
# di.py
class DI:
"""Dependency Injectionを実現する"""
def __init__(self) -> None:
# 依存関係を設定する関数を読み込む
self.injector = Injector(self.__class__.config)
# 依存関係を設定するメソッド
@classmethod
def config(cls: type, binder: Binder):
# バインド
binder.bind(interface=Car, to=SuperCar)
binder.bind(interface=Color, to=Yellow)
# injector.get()に引数を渡すと依存関係を解決してインスタンスを生成する
def resolve(self, cls: type):
return self.injector.get(cls, scope_id=scope_id)
# main.py
def main():
# Dependency クラスをインスタンス化
injector = DI()
# インスタンスを生成
a1: ExService = injector.resolve(ExService)
a2: ExService = injector.resolve(ExService)
a1.action()
a2.action()
if __name__ == "__main__":
# ログを表示したい場合は下記2行をコメントアウト
# logging.basicConfig(level=logging.DEBUG)
# logging.getLogger("injector").setLevel(logging.DEBUG)
main()]
実行結果
yellow初期化
car初期化
car初期化
color is yellow
car type is super car
color is yellow
car type is super car
- 車→スーパーカー、色→黄色 の実装ができている。
本題
さて、本題のSessionScopeについてです。
先ほども述べましたが、すでにあるスコープ以外に実装したい場合は、公式ドキュメントによるとScopeインターフェースを実装したカスタムクラスを作成する必要があります。
from injector import Scope
class CustomScope(Scope):
def get(self, key, provider):
return provider
from injector import ScopeDecorator
customscope = ScopeDecorator(CustomScope)
keyはクラスの型、providerは依存情報のようです。
例えば、プロトタイプスコープだとだとそのままproviderを返せばよいです。
シングルトンスコープだとディクショナリでkeyごとにproviderを持っておけばよいです。
また、公式ドキュメントの末尾にはこう書かれています。
For scopes with a transient lifetime, such as those tied to HTTP requests, the usual solution is to use a thread or greenlet-local cache inside the scope. The scope is "entered" in some low-level code by calling a method on the scope instance that creates this cache. Once the request is complete, the scope is "left" and the cache cleared.
翻訳：”HTTP リクエストに関連付けられたスコープなど、一時的な有効期間を持つスコープでは、 スレッドやグリーンレットローカルキャッシュをスコープ内で使用するのが一般的です。このキャッシュを作成するスコープインスタンスのメソッドを呼び出すことで、低レベルのコードでスコープに "入る "ことができます。リクエストが完了すると、スコープは "出て "いき、キャッシュはクリアされます。”
要はリクエストスコープとか実装したい場合は、Webライブラリのリクエストってスレッドもしくは非同期タスクごとに分かれて処理されるよね？そのスレッド、非同期タスクごとにprovider持ってればいいじゃん、ということです。
ただ、参考通りに実装するとFastAPIでは同期処理だとリクエストごとにスレッドが作成されますが、非同期処理だとメインスレッドで実行されるので、同期処理の場合はスレッドごとの管理、非同期処理の場合は非同期タスクごとの管理と使い分けが大変です。
（同期処理だけという縛りなら既存のThreadLocalScopeがリクエストスコープということになりますね）
それに私が実装したいのはSessionScopeです。
スレッドとか非同期とか関係なくsession_idごとに管理したいのです。
CustomScopeにはsession_idを挟み込む余地がありません。
なので無理やりsession_idを挟み込めるようにソースコードを変更しました。
1. ライブラリのソースコードの改変
- まずはsession_idを挟み込めるようにライブラリのソースコードを改変します。
- ソースコードは
<インストールパス>/site-packages/injector/__init__.pyにあります。
※injector==0.21.0のソースコードを改変しています。
※改変は自己責任でお願いします。
-
class Injector:を下記のように書き換えます。
class Injector:
"""
:param modules: Optional - a configuration module or iterable of configuration modules.
Each module will be installed in current :class:`Binder` using :meth:`Binder.install`.
Consult :meth:`Binder.install` documentation for the details.
:param auto_bind: Whether to automatically bind missing types.
:param parent: Parent injector.
.. versionadded:: 0.7.5
``use_annotations`` parameter
.. versionchanged:: 0.13.0
``use_annotations`` parameter is removed
"""
_stack: Tuple[Tuple[object, Callable, Tuple[Tuple[str, type], ...]], ...]
binder: Binder
def __init__(
self,
modules: Union[_InstallableModuleType, Iterable[_InstallableModuleType], None] = None,
auto_bind: bool = True,
parent: Optional['Injector'] = None,
) -> None:
# Stack of keys currently being injected. Used to detect circular
# dependencies.
self._stack = ()
self.parent = parent
# Binder
self.binder = Binder(self, auto_bind=auto_bind, parent=parent.binder if parent is not None else None)
if not modules:
modules = []
elif not hasattr(modules, '__iter__'):
modules = [cast(_InstallableModuleType, modules)]
# This line is needed to pelase mypy. We know we have Iteable of modules here.
modules = cast(Iterable[_InstallableModuleType], modules)
# Bind some useful types
self.binder.bind(Injector, to=self)
self.binder.bind(Binder, to=self.binder)
# Initialise modules
for module in modules:
self.binder.install(module)
@property
def _log_prefix(self) -> str:
return '>' * (len(self._stack) + 1) + ' '
@synchronized(lock)
def get(self, interface: Type[T], scope: Union[ScopeDecorator, Type[Scope], None] = None) -> T:
"""Get an instance of the given interface.
.. note::
Although this method is part of :class:`Injector`'s public interface
it's meant to be used in limited set of circumstances.
For example, to create some kind of root object (application object)
of your application (note that only one `get` call is needed,
inside the `Application` class and any of its dependencies
:func:`inject` can and should be used):
.. code-block:: python
class Application:
@inject
def __init__(self, dep1: Dep1, dep2: Dep2):
self.dep1 = dep1
self.dep2 = dep2
def run(self):
self.dep1.something()
injector = Injector(configuration)
application = injector.get(Application)
application.run()
:param interface: Interface whose implementation we want.
:param scope: Class of the Scope in which to resolve.
:returns: An implementation of interface.
"""
binding, binder = self.binder.get_binding(interface)
scope = scope or binding.scope
if isinstance(scope, ScopeDecorator):
scope = scope.scope
# Fetch the corresponding Scope instance from the Binder.
scope_binding, _ = binder.get_binding(scope)
scope_instance = scope_binding.provider.get(self)
log.debug(
'%sInjector.get(%r, scope=%r) using %r', self._log_prefix, interface, scope, binding.provider
)
provider_instance = scope_instance.get(interface, binding.provider)
result = provider_instance.get(self)
log.debug('%s -> %r', self._log_prefix, result)
return result
def create_child_injector(self, *args: Any, **kwargs: Any) -> 'Injector':
kwargs['parent'] = self
return Injector(*args, **kwargs)
def create_object(self, cls: Type[T], additional_kwargs: Any = None, session_id: Optional[str] = None) -> T:
"""Create a new instance, satisfying any dependencies on cls."""
additional_kwargs = additional_kwargs or {}
log.debug("%sCreating %r object with %r", self._log_prefix, cls, additional_kwargs)
try:
instance = cls.__new__(cls)
except TypeError as e:
reraise(
e,
CallError(cls, getattr(cls.__new__, "__func__", cls.__new__), (), {}, e, self._stack),
maximum_frames=2,
)
init = cls.__init__
try:
self.call_with_injection(init, self_=instance, kwargs=additional_kwargs, session_id=session_id)
except TypeError as e:
# Mypy says "Cannot access "__init__" directly"
init_function = instance.__init__.__func__ # type: ignore
reraise(e, CallError(instance, init_function, (), additional_kwargs, e, self._stack))
return instance
def call_with_injection(
self,
callable: Callable[..., T],
self_: Any = None,
args: Any = (),
kwargs: Any = {},
+ session_id: Optional[str] = None,
) -> T:
"""Call a callable and provide it's dependencies if needed.
:param self_: Instance of a class callable belongs to if it's a method,
None otherwise.
:param args: Arguments to pass to callable.
:param kwargs: Keyword arguments to pass to callable.
:type callable: callable
:type args: tuple of objects
:type kwargs: dict of string -> object
+ :type session_id: session_id -> Optional[str]
:return: Value returned by callable.
"""
bindings = get_bindings(callable)
signature = inspect.signature(callable)
full_args = args
if self_ is not None:
full_args = (self_,) + full_args
bound_arguments = signature.bind_partial(*full_args)
needed = dict((k, v) for (k, v) in bindings.items() if k not in kwargs and k not in bound_arguments.arguments)
dependencies = self.args_to_inject(
function=callable,
bindings=needed,
owner_key=self_.__class__ if self_ is not None else callable.__module__,
+ session_id=session_id,
)
dependencies.update(kwargs)
try:
return callable(*full_args, **dependencies)
except TypeError as e:
reraise(e, CallError(self_, callable, args, dependencies, e, self._stack))
# Needed because of a mypy-related issue (https://github.com/python/mypy/issues/8129).
assert False, "unreachable" # pragma: no cover
@private
@synchronized(lock)
def args_to_inject(
+ self, function: Callable, bindings: Dict[str, type], owner_key: object, session_id: Optional[str] = None
) -> Dict[str, Any]:
"""Inject arguments into a function.
:param function: The function.
:param bindings: Map of argument name to binding key to inject.
:param owner_key: A key uniquely identifying the *scope* of this function.
For a method this will be the owning class.
+ :param session_id: session_id.
:returns: Dictionary of resolved arguments.
"""
dependencies = {}
key = (owner_key, function, tuple(sorted(bindings.items())))
def repr_key(k: Tuple[object, Callable, Tuple[Tuple[str, type], ...]]) -> str:
owner_key, function, bindings = k
return "%s.%s(injecting %s)" % (tuple(map(_describe, k[:2])) + (dict(k[2]),))
log.debug("%sProviding %r for %r", self._log_prefix, bindings, function)
if key in self._stack:
raise CircularDependency(
"circular dependency detected: %s -> %s" % (" -> ".join(map(repr_key, self._stack)), repr_key(key))
)
self._stack += (key,)
try:
for arg, interface in bindings.items():
try:
+ instance: Any = self.get(interface, session_id=session_id)
except UnsatisfiedRequirement as e:
if not e.owner:
e = UnsatisfiedRequirement(owner_key, e.interface)
raise e
dependencies[arg] = instance
finally:
self._stack = tuple(self._stack[:-1])
return dependencies
-
class Provider(Generic[T]):以降を下記のように変更します。
class Provider(Generic[T]):
"""Provides class instances."""
__metaclass__ = ABCMeta
@abstractmethod
+ def get(self, injector: 'Injector', session_id: Optional[str] = None) -> T:
raise NotImplementedError # pragma: no cover
class ClassProvider(Provider, Generic[T]):
"""Provides instances from a given class, created using an Injector."""
def __init__(self, cls: Type[T]) -> None:
self._cls = cls
+ def get(self, injector: 'Injector', session_id: Optional[str] = None) -> T:
+ return injector.create_object(self._cls, session_id = session_id)
class CallableProvider(Provider, Generic[T]):
"""Provides something using a callable.
The callable is called every time new value is requested from the provider.
There's no need to explicitly use :func:`inject` or :data:`Inject` with the callable as it's
assumed that, if the callable has annotated parameters, they're meant to be provided
automatically. It wouldn't make sense any other way, as there's no mechanism to provide
parameters to the callable at a later time, so either they'll be injected or there'll be
a `CallError`.
::
>>> class MyClass:
... def __init__(self, value: int) -> None:
... self.value = value
...
>>> def factory():
... print('providing')
... return MyClass(42)
...
>>> def configure(binder):
... binder.bind(MyClass, to=CallableProvider(factory))
...
>>> injector = Injector(configure)
>>> injector.get(MyClass) is injector.get(MyClass)
providing
providing
False
"""
def __init__(self, callable: Callable[..., T]):
self._callable = callable
+ def get(self, injector: 'Injector', session_id: Optional[str] = None) -> T:
+ return injector.call_with_injection(self._callable, session_id = session_id)
def __repr__(self) -> str:
return '%s(%r)' % (type(self).__name__, self._callable)
class InstanceProvider(Provider, Generic[T]):
"""Provide a specific instance.
::
>>> class MyType:
... def __init__(self):
... self.contents = []
>>> def configure(binder):
... binder.bind(MyType, to=InstanceProvider(MyType()))
...
>>> injector = Injector(configure)
>>> injector.get(MyType) is injector.get(MyType)
True
>>> injector.get(MyType).contents.append('x')
>>> injector.get(MyType).contents
['x']
"""
def __init__(self, instance: T) -> None:
self._instance = instance
+ def get(self, injector: 'Injector', session_id: Optional[str] = None) -> T:
return self._instance
def __repr__(self) -> str:
return '%s(%r)' % (type(self).__name__, self._instance)
@private
class ListOfProviders(Provider, Generic[T]):
"""Provide a list of instances via other Providers."""
_providers: List[Provider[T]]
def __init__(self) -> None:
self._providers = []
def append(self, provider: Provider[T]) -> None:
self._providers.append(provider)
def __repr__(self) -> str:
return '%s(%r)' % (type(self).__name__, self._providers)
class MultiBindProvider(ListOfProviders[List[T]]):
"""Used by :meth:`Binder.multibind` to flatten results of providers that
return sequences."""
+ def get(self, injector: 'Injector', session_id: Optional[str] = None) -> List[T]:
return [i for provider in self._providers for i in provider.get(injector)]
class MapBindProvider(ListOfProviders[Dict[str, T]]):
"""A provider for map bindings."""
+ def get(self, injector: 'Injector', session_id: Optional[str] = None) -> Dict[str, T]:
map: Dict[str, T] = {}
for provider in self._providers:
map.update(provider.get(injector))
return map
- 長いコードですが、session_idを使えるように引数に追加しただけです。
2. SessionScopeの実装
- ここからはローカルでSessionScope実装していきます。
ディレクトリ構造
| global_value.py
│ injector.py
│ ex_service.py
│ di.py
│ main.py
実装
# global_value.py
# instance = a[session_id][class type]
session_scope_instance: Dict[str, Dict[type, Provider]] = {}
- グローバル変数です。
# injector.py
from typing import Optional, Type
import global_value as g
from injector import (
CallError,
Injector,
InstanceProvider,
Provider,
Scope,
ScopeDecorator,
T,
UnsatisfiedRequirement,
lock,
synchronized,
)
class SessionScope(Scope):
"""A :class:`Scope` that returns a per-Injector instance for a session_id and a key.
:data:`session` can be used as a convenience class decorator.
>>> class A: pass
>>> injector = Injector()
>>> provider = ClassProvider(A)
>>> session = SessionScope(injector)
>>> a = session.get(A, provider, session_id)
>>> b = session.get(A, provider, session_id2)
>>> a is b
False
>>> c = session.get(A, provider, session_id2)
>>> b is c
True
"""
@synchronized(lock)
def get(self, key: Type[T], provider: Provider[T], session_id: Optional[str] = None) -> Provider[T]:
id: str = session_id or "common"
try:
return g.session_scope_instance[id][key]
except KeyError:
instance = self._get_instance(key, provider, self.injector)
provider = InstanceProvider(instance)
if id not in g.session_scope_instance:
g.session_scope_instance[id] = {}
g.session_scope_instance[id][key] = provider
return provider
def _get_instance(self, key: Type[T], provider: Provider[T], injector: Injector) -> T:
if injector.parent and not injector.binder.has_explicit_binding_for(key):
try:
return self._get_instance_from_parent(key, provider, injector.parent)
except (CallError, UnsatisfiedRequirement):
pass
return provider.get(injector)
def _get_instance_from_parent(self, key: Type[T], provider: Provider[T], parent: Injector) -> T:
singleton_scope_binding, _ = parent.binder.get_binding(type(self))
singleton_scope = singleton_scope_binding.provider.get(parent)
provider = singleton_scope.get(key, provider)
return provider.get(parent)
session_scope = ScopeDecorator(SessionScope)
-
injectorソース内のInjectorクラスを継承してsession_idをインジェクションする際に指定できるようにします。
# ex_service.py
# 抽象クラス
class Car(metaclass=ABCMeta):
@abstractmethod
def disp_car_type(self) -> None:
raise NotImplementedError
# 抽象クラス
class Color(metaclass=ABCMeta):
@abstractmethod
def disp_color(self) -> None:
raise NotImplementedError
# セッションスコープをつける
+ @session_scope
class SuperCar(Car):
@inject
def __init__(self, color: Color): # ColorのインスタンスをDI
self.color = color
print("car初期化")
def disp_car_type(self):
# 色も表示する
self.color.disp_color()
print("car type is super car")
@singleton
class Yellow(Color):
def __init__(self):
print("yellow初期化")
def disp_color(self):
print("color is yellow")
@inject
@dataclass
class ExService:
car: Car
def action(self) -> None:
self.car.disp_car_type()
-
@session_scopeをつけるだけでセッションごとに管理してくれます…便利！
# di.py
class DI:
"""Dependency Injection"""
def __init__(self) -> None:
# 依存関係を設定する関数を読み込む
self.injector = CustomeInjector(self.__class__.config)
# 依存関係を設定するメソッド
@classmethod
def config(cls: type, binder: Binder):
# バインド
binder.bind(interface=Car, to=SuperCar)
binder.bind(interface=Color, to=Yellow)
# injector.get()に引数を渡すと依存関係を解決してインスタンスを生成する
def resolve(self, cls: type, session_id: Optional[str] = None):
return self.injector.get(cls, session_id=session_id)
# main.py
def main():
# Dependency クラスをインスタンス化
injector = DI()
# インスタンスを生成
a1: ExService = injector.resolve(ExService, "session_id_1")
a2: ExService = injector.resolve(ExService, "session_id_2")
a3: ExService = injector.resolve(ExService, "session_id_1")
print(f"別セッションでのインスタンス: {a1.car is a2.car}")
print(f"同セッションでのインスタンス: {a1.car is a3.car}")
a1.action()
a2.action()
a3.action()
if __name__ == "__main__":
# ログを表示したい場合は下記2行をコメントアウト
# logging.basicConfig(level=logging.DEBUG)
# logging.getLogger("injector").setLevel(logging.DEBUG)
main()
実行結果
yellow初期化
car初期化
car初期化
別セッションでのインスタンス: False
同セッションでのインスタンス: True
color is yellow
car type is super car
color is yellow
car type is super car
color is yellow
car type is super car
- セッションごとに管理できています。
- セッションの破棄ですが、セッションごとのインスタンスはglobal_value.pyで{ session_id: instance }のdict型でグローバル変数として管理しているので、一定時間監視する別スレッド関数を作ってセッションの有効期限が切れた際に破棄するようにしてください。
感想
- PythonのInjectorライブラリを使用してSessionScopeを実現しました。
- まさかライブラリのソースの中までカスタムするとは思いませんでした。
- 正直今回の改変は無理やり実装した感があるのでベストな方法だとは思いません。
- Injectorのgithubで質問しようとも考えましたが、個人で使う分にはもういいのかなと、割り切って使うことにしました。
- 他のライブラリでもっと簡単にSessionScopeを実装できるよ or もっとスマートな実装方法があるよ等ありましたらコメントくださいましたら幸いです。