Implementation of SessionScope Needed

[tomyou666]

I am using this library in my personal development project with FastAPI + SQLAlchemy. For instance, when creating an application that handles a database, I often find the need for a SessionScope to manage instances on a per-session basis.

Here are the requirements I have for SessionScope:

1. It should be able to hold providers managed by a unique ID such as session_id.
2. It should allow specifying session_id during dependency resolution.

I attempted to implement SessionScope by inheriting from the Scope class, following the example in docs/scopes.rst. However, I found that inheriting from the Scope class alone doesn't allow managing with session_id effectively.

I believe this is structurally challenging in Injector, Provider, and Scope classes to propagate a unique ID like session_id (or scope_id). If I'm wrong, I'd like your opinion.

So, I propose the following solutions:

example)

1. Allow Injector to accept session_id (or another unique ID) as an argument.

class Injector:
    @synchronized(lock)
    def get(
        self,
        interface: Type[T],
        scope: Union[ScopeDecorator, Type[Scope], None] = None,
+       session_id: Optional[str] = None,
    ) -> T:
        """_summary_

        Args:
            interface (Type[T]): interface
            scope (Union[ScopeDecorator, Type[Scope], None], optional): scope. Defaults to None.
+           session_id (Optional[str], optional): session_id. Defaults to None.

        Returns:
            T: _description_
        """
        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)
        # For request scope, include scopeID as an argument
+       if isinstance(scope_instance, SessionScope):
+           provider_instance = scope_instance.get(interface, binding.provider, session_id)
        else:
            provider_instance = scope_instance.get(interface, binding.provider)

+       result = provider_instance.get(self, session_id=session_id)
        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

2. Allow Provider classes and their subclasses to accept session_id (or another unique ID) as an argument.

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

3. Implementation of SessionScope

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
    """

    _context: Dict[str, Dict[type, Provider]]

    def configure(self) -> None:
        self._context = {}

    @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 self._context[id][key]
        except KeyError:
            instance = self._get_instance(key, provider, self.injector)
            provider = InstanceProvider(instance)
            if id not in self._context:
                self._context[id] = {}
            self._context[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)

Regarding 3., I don't know how to destroy the _context session at an arbitrary time. Is there a better way to do this as well?