in_n_out

pluggable dependency injection and result processing.

Generally speaking, providers and processors are defined as follows:

• Provider: TypeAlias = Callable[[], Any]: a callable that can accept no arguments and returns an instance of some type. When we refer to a
• Processor: TypeAlias = Callable[[Any], Any]: a callable that accepts a single positional argument (an instance of some type) and returns anything (the return value is ignored).

Store

Store(name: str)

A Store is a collection of providers and processors.

Source code in in_n_out/_store.py

def __init__(self, name: str) -> None:
    self._name = name
    self._providers: list[_RegisteredCallback] = []
    self._processors: list[_RegisteredCallback] = []
    self._namespace: Namespace | Callable[[], Namespace] | None = None
    self.on_unresolved_required_args: RaiseWarnReturnIgnore = "warn"
    self.on_unannotated_required_args: RaiseWarnReturnIgnore = "warn"
    self.guess_self: bool = True

name property

name: str

Return the name of this Store.

namespace property writable

namespace: dict[str, object]

Return namespace for type resolution, if this store has one.

If no namespace is set, this will return an empty dict.

clear

clear() -> None

Clear all providers and processors.

Source code in in_n_out/_store.py

def clear(self) -> None:
    """Clear all providers and processors."""
    self._providers.clear()
    self._processors.clear()
    with contextlib.suppress(AttributeError):
        del self._cached_processor_map
    with contextlib.suppress(AttributeError):
        del self._cached_provider_map

create classmethod

create(name: str) -> Store

Create a new Store instance with the given name.

This name can be used to refer to the Store in other functions.

Parameters:

• name (str) –

  A name for the Store.

Returns:

• Store –

  A Store instance with the given name.

Raises:

• KeyError –

  If the name is already in use, or the name is 'global'.

Source code in in_n_out/_store.py

@classmethod
def create(cls, name: str) -> Store:
    """Create a new Store instance with the given `name`.

    This name can be used to refer to the Store in other functions.

    Parameters
    ----------
    name : str
        A name for the Store.

    Returns
    -------
    Store
        A Store instance with the given `name`.

    Raises
    ------
    KeyError
        If the name is already in use, or the name is 'global'.
    """
    name = name.lower()
    if name == _GLOBAL:
        raise KeyError("'global' is a reserved store name")
    elif name in cls._instances:
        raise KeyError(f"Store {name!r} already exists")
    cls._instances[name] = cls(name)
    return cls._instances[name]

destroy classmethod

destroy(name: str) -> None

Destroy Store instance with the given name.

Parameters:

• name (str) –

  The name of the Store.

Raises:

• ValueError –

  If the name matches the global store name.

• KeyError –

  If the name is not in use.

Source code in in_n_out/_store.py

@classmethod
def destroy(cls, name: str) -> None:
    """Destroy Store instance with the given `name`.

    Parameters
    ----------
    name : str
        The name of the Store.

    Raises
    ------
    ValueError
        If the name matches the global store name.
    KeyError
        If the name is not in use.
    """
    name = name.lower()
    if name == _GLOBAL:
        raise ValueError("The global store cannot be destroyed")
    elif name not in cls._instances:
        raise KeyError(f"Store {name!r} does not exist")
    del cls._instances[name]

get_store classmethod

get_store(name: str | None = None) -> Store

Get a Store instance with the given name.

Parameters:

• name (str, default: None ) –

  The name of the Store.

Returns:

• Store –

  A Store instance with the given name.

Raises:

• KeyError –

  If the name is not in use.

Source code in in_n_out/_store.py

@classmethod
def get_store(cls, name: str | None = None) -> Store:
    """Get a Store instance with the given `name`.

    Parameters
    ----------
    name : str
        The name of the Store.

    Returns
    -------
    Store
        A Store instance with the given `name`.

    Raises
    ------
    KeyError
        If the name is not in use.
    """
    name = (name or _GLOBAL).lower()
    if name not in cls._instances:
        raise KeyError(f"Store {name!r} does not exist")
    return cls._instances[name]

inject

inject(func: Callable[P, R] | None = None, *, providers: bool = True, processors: bool = False, localns: dict | None = None, on_unresolved_required_args: RaiseWarnReturnIgnore | None = None, on_unannotated_required_args: RaiseWarnReturnIgnore | None = None, guess_self: bool | None = None) -> Callable[..., R] | Callable[[Callable[P, R]], Callable[..., R]]

Decorate func to inject dependencies at calltime.

Assuming providers is True (the default), this will attempt retrieve instances of the types required by func and inject them into func at calltime. The primary consequence of this is that func may be called without parameters (assuming the required providers have been registered). See usages examples below.

Note that an injected function may still be explicitly invoked with parameters.

See register for more information on how to register providers and processors.

Parameters:

• func (Callable, default: None ) –

  A function to decorate. Type hints are used to determine what to inject.

• providers (bool, default: True ) –

  Whether to inject dependency providers. If True (default), then when this function is called, arguments will be injected into the function call according to providers that have been registered in the store.

• processors (bool, default: False ) –

  Whether to invoke all processors for this function's return type the when this function is called. Important: this causes side effects. By default, False. Output processing can also be enabled (with additionl fine tuning) by using the @store.process_result decorator.

• localns (dict | None, default: None ) –

  Optional local namespace for name resolution, by default None

• on_unresolved_required_args (RaiseWarnReturnIgnore, default: None ) –

  What to do when a required parameter (one without a default) is encountered with an unresolvable type annotation. Must be one of the following (by default 'warn'):

  • 'raise': immediately raise an exception
  • 'warn': warn and return the original function
  • 'return': return the original function without warning
  • 'ignore': continue decorating without warning (at call time, this function will fail without additional arguments).
• on_unannotated_required_args (RaiseWarnReturnIgnore, default: None ) –

  What to do when a required parameter (one without a default) is encountered with an no type annotation. These functions are likely to fail when called later if the required parameter is not provided. Must be one of the following (by default 'warn'):

  • 'raise': immediately raise an exception
  • 'warn': warn, but continue decorating
  • 'return': immediately return the original function without warning
  • 'ignore': continue decorating without warning.
• guess_self (bool, default: None ) –

  Whether to infer the type of the first argument if the function is an unbound class method (by default, True) This is done as follows:

  • if '.' (but not '<locals>') is in the function's __qualname__
  • and if the first parameter is named 'self' or starts with "_"
  • and if the first parameter annotation is inspect.empty
  • then the name preceding func.__name__ in the function's __qualname__ (which is usually the class name), is looked up in the function's __globals__ namespace. If found, it is used as the first parameter's type annotation.

  This allows class methods to be injected with instances of the class.

Returns:

• Callable –

  A function with dependencies injected

Examples:

>>> import in_n_out as ino
>>> class Thing:
...     def __init__(self, name: str):
...         self.name = name
>>> @ino.inject
... def func(thing: Thing):
...     return thing.name

>>> # no providers available yet
>>> func()
TypeError: ... missing 1 required positional argument: 'thing'

>>> # register a provider
>>> ino.register(providers={Thing: Thing("Thing1")})
>>> print(func())
'Thing1'

>>> # can still override with parameters
>>> func(Thing("OtherThing"))
'OtherThing'

Source code in in_n_out/_store.py

def inject(
    self,
    func: Callable[P, R] | None = None,
    *,
    providers: bool = True,
    processors: bool = False,
    localns: dict | None = None,
    on_unresolved_required_args: RaiseWarnReturnIgnore | None = None,
    on_unannotated_required_args: RaiseWarnReturnIgnore | None = None,
    guess_self: bool | None = None,
) -> Callable[..., R] | Callable[[Callable[P, R]], Callable[..., R]]:
    """Decorate `func` to inject dependencies at calltime.

    Assuming `providers` is True (the default), this will attempt retrieve
    instances of the types required by `func` and inject them into `func` at
    calltime.  The primary consequence of this is that `func` may be called
    without parameters (assuming the required providers have been registered).
    See usages examples below.

    Note that an injected function may *still* be explicitly invoked with
    parameters.

    See `register` for more information on how to register providers and processors.

    Parameters
    ----------
    func : Callable
        A function to decorate. Type hints are used to determine what to inject.
    providers : bool
        Whether to inject dependency providers. If `True` (default), then when this
        function is called, arguments will be injected into the function call
        according to providers that have been registered in the store.
    processors : bool
        Whether to invoke all processors for this function's return type the when
        this function is called. Important: this causes *side effects*. By default,
        `False`. Output processing can also be enabled (with additionl fine tuning)
        by using the `@store.process_result` decorator.
    localns : dict | None
        Optional local namespace for name resolution, by default None
    on_unresolved_required_args : RaiseWarnReturnIgnore
        What to do when a required parameter (one without a default) is encountered
        with an unresolvable type annotation.
        Must be one of the following (by default 'warn'):

        - `'raise'`: immediately raise an exception
        - `'warn'`: warn and return the original function
        - `'return'`: return the original function without warning
        - `'ignore'`: continue decorating without warning (at call time, this
            function will fail without additional arguments).
    on_unannotated_required_args : RaiseWarnReturnIgnore
        What to do when a required parameter (one without a default) is encountered
        with an *no* type annotation. These functions are likely to fail when called
        later if the required parameter is not provided.
        Must be one of the following (by default 'warn'):

        - `'raise'`: immediately raise an exception
        - `'warn'`: warn, but continue decorating
        - `'return'`: immediately return the original function without warning
        - `'ignore'`: continue decorating without warning.
    guess_self : bool
        Whether to infer the type of the first argument if the function is an
        unbound class method (by default, `True`) This is done as follows:

        - if `'.'` (but not `'<locals>'`) is in the function's `__qualname__`
        - and if the first parameter is named 'self' or starts with `"_"`
        - and if the first parameter annotation is `inspect.empty`
        - then the name preceding `func.__name__` in the function's `__qualname__`
        (which is usually the class name), is looked up in the function's
        `__globals__` namespace. If found, it is used as the first parameter's
        type annotation.

        This allows class methods to be injected with instances of the class.

    Returns
    -------
    Callable
        A function with dependencies injected

    Examples
    --------
    >>> import in_n_out as ino
    >>> class Thing:
    ...     def __init__(self, name: str):
    ...         self.name = name
    >>> @ino.inject
    ... def func(thing: Thing):
    ...     return thing.name

    >>> # no providers available yet
    >>> func()
    TypeError: ... missing 1 required positional argument: 'thing'

    >>> # register a provider
    >>> ino.register(providers={Thing: Thing("Thing1")})
    >>> print(func())
    'Thing1'

    >>> # can still override with parameters
    >>> func(Thing("OtherThing"))
    'OtherThing'
    """
    on_unres = on_unresolved_required_args or self.on_unresolved_required_args
    on_unann = on_unannotated_required_args or self.on_unannotated_required_args
    _guess_self = guess_self or self.guess_self

    # inner decorator, allows for optional decorator arguments
    def _inner(func: Callable[P, R]) -> Callable[P, R]:
        # if the function takes no arguments and has no return annotation
        # there's nothing to be done
        if not providers:
            return self.inject_processors(func) if processors else func

        # bail if there aren't any annotations at all
        code: CodeType | None = getattr(unwrap(func), "__code__", None)
        if (code and not code.co_argcount) and "return" not in getattr(
            func, "__annotations__", {}
        ):
            return func

        # get a signature object with all type annotations resolved
        # this may result in a NameError if a required argument is unresolvable.
        # There may also be unannotated required arguments, which will likely fail
        # when the function is called later. We break this out into a separate
        # function to handle notifying the user on these cases.
        sig = _resolve_sig_or_inform(
            func,
            localns={**self.namespace, **(localns or {})},
            on_unresolved_required_args=on_unres,
            on_unannotated_required_args=on_unann,
            guess_self=_guess_self,
        )
        if sig is None:  # something went wrong, and the user was notified.
            return func

        _fname = getattr(func, "__qualname__", func)

        # get provider functions for each required parameter
        @wraps(func)
        def _exec(*args: P.args, **kwargs: P.kwargs) -> R:
            # we're actually calling the "injected function" now
            logger.debug(
                "Executing @injected %s%s with args: %r, kwargs: %r",
                _fname,
                sig,
                args,
                kwargs,
            )

            # use bind_partial to allow the caller to still provide their own args
            # if desired. (i.e. the injected deps are only used if not provided)
            bound = sig.bind_partial(*args, **kwargs)
            bound.apply_defaults()

            # first, get and call the provider functions for each parameter type:
            _injected_names: set[str] = set()
            for param in sig.parameters.values():
                if (
                    param.name not in bound.arguments
                    or bound.arguments[param.name] is None
                ):
                    provided = self.provide(param.annotation)
                    if provided is not None or is_optional(param.annotation):
                        logger.debug(
                            "  injecting %s: %s = %r",
                            param.name,
                            param.annotation,
                            provided,
                        )
                        _injected_names.add(param.name)
                        bound.arguments[param.name] = provided

            # call the function with injected values
            logger.debug(
                "  Calling %s with %r (injected %r)",
                _fname,
                bound.arguments,
                _injected_names,
            )
            try:
                result = func(**bound.arguments)  # type: ignore[call-arg]
            except TypeError as e:
                if "missing" not in e.args[0]:
                    raise  # pragma: no cover
                # likely a required argument is still missing.
                # show what was injected and raise
                logger.exception(e)
                for param in sig.parameters.values():
                    if (
                        param.name not in bound.arguments
                        and param.default is param.empty
                    ):
                        logger.error(
                            f"Do not have argument for {param.name}: using "
                            "providers "
                            f"{list(self.iter_providers(param.annotation))}"
                        )

                mod_name = getattr(func, "__module__", "")
                func_name = getattr(func, "__qualname__", str(func))
                full_name = f"{mod_name}.{func_name}"
                raise TypeError(
                    f"Error calling in-n-out injected function {full_name!r} with "
                    f"kwargs {bound.arguments!r}.\n\n"
                    f"See {type(e).__name__} above for more details."
                ) from e

            return result

        out = _exec

        # if it came in as a generatorfunction, it needs to go out as one.
        if isgeneratorfunction(func):

            @wraps(func)
            def _gexec(*args: P.args, **kwargs: P.kwargs) -> R:  # type: ignore
                yield from _exec(*args, **kwargs)  # type: ignore

            out = _gexec  # type: ignore

        # update some metadata on the decorated function.
        out.__signature__ = sig  # type: ignore [attr-defined]
        out.__annotations__ = {
            **{p.name: p.annotation for p in sig.parameters.values()},
            "return": sig.return_annotation,
        }
        out.__doc__ = (
            out.__doc__ or ""
        ) + "\n\n*This function will inject dependencies when called.*"

        if processors:
            return self.inject_processors(out, type_hint=sig.return_annotation)
        return out

    return _inner(func) if func is not None else _inner

inject_processors

inject_processors(func: Callable[P, R] | None = None, *, type_hint: type[T] | object | None = None, first_processor_only: bool = False, raise_exception: bool = False) -> Callable[[Callable[P, R]], Callable[P, R]] | Callable[P, R]

Decorate a function to process its output.

Variant of inject, but only injects processors (for the sake of more explicit syntax).

When the decorated function is called, the return value will be processed with store.process(return_value) before returning the result.

Important! This means that calling func will likely have side effects.

Parameters:

• func (Callable, default: None ) –

  A function to decorate. Return hints are used to determine what to process.

• type_hint (type[T] | object | None, default: None ) –

  Type hint for the return value. If not provided, the type will be inferred first from the return annotation of the function, and if that is not provided, from the type(return_value).

• first_processor_only (bool, default: False ) –

  If True, only the first processor will be invoked, otherwise all processors will be invoked, in descending weight order.

• raise_exception (bool, default: False ) –

  If True, and a processor raises an exception, it will be raised and the remaining processors will not be invoked.

Returns:

• Callable –

  A function that, when called, will have its return value processed by store.process(return_value)

Source code in in_n_out/_store.py

def inject_processors(
    self,
    func: Callable[P, R] | None = None,
    *,
    type_hint: type[T] | object | None = None,
    first_processor_only: bool = False,
    raise_exception: bool = False,
) -> Callable[[Callable[P, R]], Callable[P, R]] | Callable[P, R]:
    """Decorate a function to process its output.

    Variant of [`inject`][in_n_out.Store.inject], but only injects processors
    (for the sake of more explicit syntax).

    When the decorated function is called, the return value will be processed
    with `store.process(return_value)` before returning the result.

    Important! This means that calling `func` will likely have *side effects*.

    Parameters
    ----------
    func : Callable
        A function to decorate. Return hints are used to determine what to process.
    type_hint : type[T] | object | None
        Type hint for the return value.  If not provided, the type will be inferred
        first from the return annotation of the function, and if that is not
        provided, from the `type(return_value)`.
    first_processor_only : bool, optional
        If `True`, only the first processor will be invoked, otherwise all
        processors will be invoked, in descending weight order.
    raise_exception : bool, optional
        If `True`, and a processor raises an exception, it will be raised
        and the remaining processors will not be invoked.

    Returns
    -------
    Callable
        A function that, when called, will have its return value processed by
        `store.process(return_value)`
    """

    def _deco(func: Callable[P, R]) -> Callable[P, R]:
        if isgeneratorfunction(func):
            raise TypeError(
                "Cannot decorate a generator function with inject_processors"
            )

        nonlocal type_hint
        if type_hint is None:
            annotations = getattr(func, "__annotations__", {})
            if "return" in annotations:
                type_hint = annotations["return"]

        @wraps(func)
        def _exec(*args: P.args, **kwargs: P.kwargs) -> R:
            result = func(*args, **kwargs)
            if result is not None:
                self.process(
                    result,
                    type_hint=type_hint,
                    first_processor_only=first_processor_only,
                    raise_exception=raise_exception,
                    _funcname=getattr(func, "__qualname__", str(func)),
                )
            return result

        return _exec

    return _deco(func) if func is not None else _deco

iter_processors

iter_processors(type_hint: type[T] | object) -> Iterator[Callable[[T], Any]]

Iterate over all processors of type_hint.

Parameters:

• type_hint (type[T] | object) –

  A type or type hint for which to return processors.

Yields:

• Iterable[Callable[[], T | None]] –

  Iterable of processor callbacks.

Source code in in_n_out/_store.py

def iter_processors(
    self, type_hint: type[T] | object
) -> Iterator[Callable[[T], Any]]:
    """Iterate over all processors of `type_hint`.

    Parameters
    ----------
    type_hint : type[T] | object
        A type or type hint for which to return processors.

    Yields
    ------
    Iterable[Callable[[], T | None]]
        Iterable of processor callbacks.
    """
    return self._iter_type_map(type_hint, self._cached_processor_map)

iter_providers

iter_providers(type_hint: type[T] | object) -> Iterator[Callable[[], T | None]]

Iterate over all providers of type_hint.

Parameters:

• type_hint (type[T] | object) –

  A type or type hint for which to return providers.

Yields:

• Iterable[Callable[[], T | None]] –

  Iterable of provider callbacks.

Source code in in_n_out/_store.py

def iter_providers(
    self, type_hint: type[T] | object
) -> Iterator[Callable[[], T | None]]:
    """Iterate over all providers of `type_hint`.

    Parameters
    ----------
    type_hint : type[T] | object
        A type or type hint for which to return providers.

    Yields
    ------
    Iterable[Callable[[], T | None]]
        Iterable of provider callbacks.
    """
    return self._iter_type_map(type_hint, self._cached_provider_map)

mark_processor

mark_processor(func: ProcessorVar | None = None, *, type_hint: object | None = None, weight: float = 0) -> Callable[[ProcessorVar], ProcessorVar] | ProcessorVar

Decorate func as a processor of its first parameter type.

Parameters:

• func (Processor | None, default: None ) –

  A function to decorate. If not provided, a decorator is returned.

• type_hint (object | None, default: None ) –

  Optional type or type hint that this processor can handle. If not provided, the type hint of the first parameter of func will be used.

• weight (float, default: 0 ) –

  A weight with which to sort this processor. Higher weights are given priority, by default 0. When invoking processors, all processors will be invoked in descending weight order, unless first_processor_only is set to True.

Returns:

• Callable[[Processor], Processor] | Processor –

  If func is not provided, a decorator is returned, if func is provided then the function is returned.

Examples:

>>> @store.processor
>>> def process_int(x: int) -> None:
...     print("Processing int:", x)

Source code in in_n_out/_store.py

def mark_processor(
    self,
    func: ProcessorVar | None = None,
    *,
    type_hint: object | None = None,
    weight: float = 0,
) -> Callable[[ProcessorVar], ProcessorVar] | ProcessorVar:
    """Decorate `func` as a processor of its first parameter type.

    Parameters
    ----------
    func : Processor | None
        A function to decorate. If not provided, a decorator is returned.
    type_hint : object | None
        Optional type or type hint that this processor can handle. If not
        provided, the type hint of the first parameter of `func` will be used.
    weight : float, optional
        A weight with which to sort this processor. Higher weights are given
        priority, by default 0.  When invoking processors, all processors
        will be invoked in descending weight order, unless `first_processor_only`
        is set to `True`.

    Returns
    -------
    Callable[[Processor], Processor] | Processor
        If `func` is not provided, a decorator is returned, if `func` is provided
        then the function is returned.

    Examples
    --------
    >>> @store.processor
    >>> def process_int(x: int) -> None:
    ...     print("Processing int:", x)
    """

    def _deco(func: ProcessorVar) -> ProcessorVar:
        try:
            self.register_processor(func, type_hint=type_hint, weight=weight)
        except ValueError as e:
            warnings.warn(str(e), stacklevel=2)
        return func

    return _deco(func) if func is not None else _deco

mark_provider

mark_provider(func: ProviderVar | None = None, *, type_hint: object | None = None, weight: float = 0) -> Callable[[ProviderVar], ProviderVar] | ProviderVar

Decorate func as a provider of its first parameter type.

Note, If func returns Optional[Type], it will be registered as a provider for Type.

Parameters:

• func (Provider | None, default: None ) –

  A function to decorate. If not provided, a decorator is returned.

• type_hint (object | None, default: None ) –

  Optional type or type hint for which to register this provider. If not provided, the return annotation of func will be used.

• weight (float, default: 0 ) –

  A weight with which to sort this provider. Higher weights are given priority, by default 0

Returns:

• Callable[[Provider], Provider] | Provider –

  If func is not provided, a decorator is returned, if func is provided then the function is returned..

Examples:

>>> @store.provider
>>> def provide_int() -> int:
...     return 42

Source code in in_n_out/_store.py

def mark_provider(
    self,
    func: ProviderVar | None = None,
    *,
    type_hint: object | None = None,
    weight: float = 0,
) -> Callable[[ProviderVar], ProviderVar] | ProviderVar:
    """Decorate `func` as a provider of its first parameter type.

    Note, If func returns `Optional[Type]`, it will be registered as a provider
    for Type.

    Parameters
    ----------
    func : Provider | None
        A function to decorate. If not provided, a decorator is returned.
    type_hint : object | None
        Optional type or type hint for which to register this provider. If not
        provided, the return annotation of `func` will be used.
    weight : float
        A weight with which to sort this provider. Higher weights are given
        priority, by default 0

    Returns
    -------
    Callable[[Provider], Provider] | Provider
        If `func` is not provided, a decorator is returned, if `func` is provided
        then the function is returned..

    Examples
    --------
    >>> @store.provider
    >>> def provide_int() -> int:
    ...     return 42
    """

    def _deco(func: ProviderVar) -> ProviderVar:
        try:
            self.register_provider(func, type_hint=type_hint, weight=weight)
        except ValueError as e:
            warnings.warn(str(e), stacklevel=2)
        return func

    return _deco(func) if func is not None else _deco

process

process(result: Any, *, type_hint: type[T] | object | None = None, first_processor_only: bool = False, raise_exception: bool = False, _funcname: str = '') -> None

Process an instance of type_.

This will iterate over all processors of type_ and invoke the first one that accepts result, unless first_processor_only is set to False, in which case all processors will be invoked.

Parameters:

• result (Any) –

  The result to process

• type_hint (type[T] | object | None, default: None ) –

  An optional type hint to provide to the processor. If not provided, the type of result will be used.

• first_processor_only (bool, default: False ) –

  If True, only the first processor will be invoked, otherwise all processors will be invoked, in descending weight order.

• raise_exception (bool, default: False ) –

  If True, and a processor raises an exception, it will be raised and the remaining processors will not be invoked.

• _funcname (str, default: '' ) –

  The name of the function that called this method. This is used internally for debugging

Source code in in_n_out/_store.py

def process(
    self,
    result: Any,
    *,
    type_hint: type[T] | object | None = None,
    first_processor_only: bool = False,
    raise_exception: bool = False,
    _funcname: str = "",
) -> None:
    """Process an instance of `type_`.

    This will iterate over all processors of `type_` and invoke the first
    one that accepts `result`, unless `first_processor_only` is set to `False`,
    in which case all processors will be invoked.

    Parameters
    ----------
    result : Any
        The result to process
    type_hint : type[T] | object | None
        An optional type hint to provide to the processor.  If not provided,
        the type of `result` will be used.
    first_processor_only : bool, optional
        If `True`, only the first processor will be invoked, otherwise all
        processors will be invoked, in descending weight order.
    raise_exception : bool, optional
        If `True`, and a processor raises an exception, it will be raised
        and the remaining processors will not be invoked.
    _funcname: str, optional
        The name of the function that called this method.  This is used internally
        for debugging
    """
    if type_hint is None:
        type_hint = type(result)
    _processors: Iterable[Callable] = self.iter_processors(type_hint)
    logger.debug(
        "Invoking processors on result %r from function %r", result, _funcname
    )
    for processor in _processors:
        try:
            logger.debug("  P: %s", processor)
            processor(result)
        except Exception as e:  # pragma: no cover
            if raise_exception:
                raise e
            warnings.warn(
                f"Processor {processor!r} failed to process result {result!r}: {e}",
                stacklevel=2,
            )
        if first_processor_only:
            break

provide

provide(type_hint: type[T] | object) -> T | None

Provide an instance of type_hint.

This will iterate over all providers of type_hint and return the first one that returns a non-None value.

Parameters:

• type_hint (type[T] | object) –

  A type or type hint for which to return a value

Returns:

• T | None –

  The first non-None value returned by a provider, or None if no providers return a value.

Source code in in_n_out/_store.py

def provide(self, type_hint: type[T] | object) -> T | None:
    """Provide an instance of `type_hint`.

    This will iterate over all providers of `type_hint` and return the first
    one that returns a non-`None` value.

    Parameters
    ----------
    type_hint : type[T] | object
        A type or type hint for which to return a value

    Returns
    -------
    T | None
        The first non-`None` value returned by a provider, or `None` if no
        providers return a value.
    """
    for provider in self.iter_providers(type_hint):
        result = provider()
        if result is not None:
            return result
    return None

register

register(*, providers: ProviderIterable | None = None, processors: ProcessorIterable | None = None) -> InjectionContext

Register multiple providers and/or processors at once.

This may be used as a context manager to temporarily register providers and/or processors.

The format for providers/processors is one of: - a mapping of {type_hint: provider} pairs - an iterable of 1, 2, or 3-tuples, where each tuple in the iterable is: - (callback,) - (callback, type_hint,) - (callback, type_hint, weight)

Parameters:

• providers (CallbackIterable | None, default: None ) –

  mapping or iterable of providers to register. See format in notes above.

• processors (CallbackIterable | None, default: None ) –

  mapping or iterable of processors to register. See format in notes above.

Returns:

• InjectionContext –

  Context manager for unregistering providers and processors. If the context is entered with with store.register(): ..., then callbacks will be unregistered when the context is exited. Callbacks may also be unregistered manually using the .cleanup() method of the returned context manager.

Examples:

>>> with store.register(
        providers={int: lambda: 42},  # provided as hint->callback map
        processors=[
            (my_callback),  # hint inferred from signature
            (my_other_callback, str),  # hint explicitly provided
            (my_third_callback, int, 10)  # weight explicitly provided
        ],
    ):
        ...

Source code in in_n_out/_store.py

def register(
    self,
    *,
    providers: ProviderIterable | None = None,
    processors: ProcessorIterable | None = None,
) -> InjectionContext:
    """Register multiple providers and/or processors at once.

    This may be used as a context manager to temporarily register providers and/or
    processors.

    The format for providers/processors is one of:
        - a mapping of {type_hint: provider} pairs
        - an iterable of 1, 2, or 3-tuples, where each tuple in the iterable is:
            -  (callback,)
            -  (callback, type_hint,)
            -  (callback, type_hint, weight)

    Parameters
    ----------
    providers : CallbackIterable | None
        mapping or iterable of providers to register. See format in notes above.
    processors : CallbackIterable | None
        mapping or iterable of processors to register. See format in notes above.

    Returns
    -------
    InjectionContext
        Context manager for unregistering providers and processors. If the context
        is entered with `with store.register(): ...`, then callbacks will be
        unregistered when the context is exited.  Callbacks may also be unregistered
        manually using the `.cleanup()` method of the returned context manager.

    Examples
    --------
    >>> with store.register(
            providers={int: lambda: 42},  # provided as hint->callback map
            processors=[
                (my_callback),  # hint inferred from signature
                (my_other_callback, str),  # hint explicitly provided
                (my_third_callback, int, 10)  # weight explicitly provided
            ],
        ):
            ...
    """
    return InjectionContext(self, providers=providers, processors=processors)

register_processor

register_processor(processor: Processor, type_hint: object | None = None, weight: float = 0) -> InjectionContext

Register processor as a processor of type_hint.

Processors are callbacks that are invoked when an injected function returns an instance of type_hint.

Parameters:

• type_hint (object, default: None ) –

  A type or type hint that processor can handle.

• processor (Callable) –

  A processor callback. Must accept at least one argument.

• weight (float, default: 0 ) –

  A weight with which to sort this processor. Higher weights are given priority, by default 0. When invoking processors, all processors will be invoked in descending weight order, unless first_processor_only is set to True.

Returns:

• Callable –

  A function that unregisters the processor.

Source code in in_n_out/_store.py

def register_processor(
    self,
    processor: Processor,
    type_hint: object | None = None,
    weight: float = 0,
) -> InjectionContext:
    """Register `processor` as a processor of `type_hint`.

    Processors are callbacks that are invoked when an injected function returns
    an instance of `type_hint`.

    Parameters
    ----------
    type_hint : object
        A type or type hint that `processor` can handle.
    processor : Callable
        A processor callback. Must accept at least one argument.
    weight : float, optional
        A weight with which to sort this processor. Higher weights are given
        priority, by default 0.  When invoking processors, all processors
        will be invoked in descending weight order, unless `first_processor_only`
        is set to `True`.

    Returns
    -------
    Callable
        A function that unregisters the processor.
    """
    return self.register(processors=[(processor, type_hint, weight)])

register_provider

register_provider(provider: Provider, type_hint: object | None = None, weight: float = 0) -> InjectionContext

Register provider as a provider of type_hint.

Parameters:

• provider (Callable) –

  A provider callback. Must be able to accept no arguments.

• type_hint (object | None, default: None ) –

  A type or type hint that provider provides. If not provided, it will be inferred from the return annotation of provider.

• weight (float, default: 0 ) –

  A weight with which to sort this provider. Higher weights are given priority, by default 0

Returns:

• Callable –

  A function that unregisters the provider.

Source code in in_n_out/_store.py

def register_provider(
    self,
    provider: Provider,
    type_hint: object | None = None,
    weight: float = 0,
) -> InjectionContext:
    """Register `provider` as a provider of `type_hint`.

    Parameters
    ----------
    provider : Callable
        A provider callback. Must be able to accept no arguments.
    type_hint : object | None
        A type or type hint that `provider` provides.  If not provided, it will
        be inferred from the return annotation of `provider`.
    weight : float, optional
        A weight with which to sort this provider. Higher weights are given
        priority, by default 0

    Returns
    -------
    Callable
        A function that unregisters the provider.
    """
    return self.register(providers=[(provider, type_hint, weight)])

inject

inject(func: Callable[P, R] | None = None, *, providers: bool = True, processors: bool = False, localns: dict | None = None, on_unresolved_required_args: RaiseWarnReturnIgnore | None = None, on_unannotated_required_args: RaiseWarnReturnIgnore | None = None, guess_self: bool | None = None, store: str | Store | None = None) -> Callable[..., R] | Callable[[Callable[P, R]], Callable[..., R]]

Decorate func to inject dependencies at calltime from store or the global store.

See Store.inject for details.

Source code in in_n_out/_global.py

@_add_store_to_doc
def inject(
    func: Callable[P, R] | None = None,
    *,
    providers: bool = True,
    processors: bool = False,
    localns: dict | None = None,
    on_unresolved_required_args: RaiseWarnReturnIgnore | None = None,
    on_unannotated_required_args: RaiseWarnReturnIgnore | None = None,
    guess_self: bool | None = None,
    store: str | Store | None = None,
) -> Callable[..., R] | Callable[[Callable[P, R]], Callable[..., R]]:
    """Decorate `func` to inject dependencies at calltime from `store` or the global store.

    See [`Store.inject`][in_n_out.Store.inject] for details.
    """  # noqa: E501
    return _store_or_global(store).inject(
        func=func,
        providers=providers,
        processors=processors,
        localns=localns,
        on_unresolved_required_args=on_unresolved_required_args,
        on_unannotated_required_args=on_unannotated_required_args,
        guess_self=guess_self,
    )

inject_processors

inject_processors(func: Callable[P, R] | None = None, *, hint: object | type[T] | None = None, first_processor_only: bool = False, raise_exception: bool = False, store: str | Store | None = None) -> Callable[[Callable[P, R]], Callable[P, R]] | Callable[P, R]

Decorate a function to process its output from store or the global store.

See Store.inject_processors for details.

Source code in in_n_out/_global.py

@_add_store_to_doc
def inject_processors(
    func: Callable[P, R] | None = None,
    *,
    hint: object | type[T] | None = None,
    first_processor_only: bool = False,
    raise_exception: bool = False,
    store: str | Store | None = None,
) -> Callable[[Callable[P, R]], Callable[P, R]] | Callable[P, R]:
    """Decorate a function to process its output from `store` or the global store.

    See [`Store.inject_processors`][in_n_out.Store.inject_processors] for details.
    """
    return _store_or_global(store).inject_processors(
        func=func,
        type_hint=hint,
        first_processor_only=first_processor_only,
        raise_exception=raise_exception,
    )

iter_processors

iter_processors(type_hint: object | type[T], store: str | Store | None = None) -> Iterable[Callable[[T], Any]]

Iterate over all processors of type_hint in store or the global store.

See Store.iter_processors for details.

Source code in in_n_out/_global.py

@_add_store_to_doc
def iter_processors(
    type_hint: object | type[T], store: str | Store | None = None
) -> Iterable[Callable[[T], Any]]:
    """Iterate over all processors of `type_hint` in `store` or the global store.

    See [`Store.iter_processors`][in_n_out.Store.iter_processors] for details.
    """
    return _store_or_global(store).iter_processors(type_hint)

iter_providers

iter_providers(type_hint: type[T], store: str | Store | None = None) -> Iterable[Callable[[], T | None]]

Iterate over all providers of type_hint in store or the global store.

See Store.iter_providers for details.

Source code in in_n_out/_global.py

@_add_store_to_doc
def iter_providers(
    type_hint: type[T], store: str | Store | None = None
) -> Iterable[Callable[[], T | None]]:
    """Iterate over all providers of `type_hint` in `store` or the global store.

    See [`Store.iter_providers`][in_n_out.Store.iter_providers] for details.
    """
    return _store_or_global(store).iter_providers(type_hint)

mark_processor

mark_processor(func: ProcessorVar | None = None, *, weight: float = 0, type_hint: object | None = None, store: str | Store | None = None) -> Callable[[ProcessorVar], ProcessorVar] | ProcessorVar

Decorate func as a processor in store or the global store.

See Store.mark_processor for details.

Source code in in_n_out/_global.py

@_add_store_to_doc
def mark_processor(
    func: ProcessorVar | None = None,
    *,
    weight: float = 0,
    type_hint: object | None = None,
    store: str | Store | None = None,
) -> Callable[[ProcessorVar], ProcessorVar] | ProcessorVar:
    """Decorate `func` as a processor in `store` or the global store.

    See [`Store.mark_processor`][in_n_out.Store.mark_processor] for details.
    """
    return _store_or_global(store).mark_processor(
        func, weight=weight, type_hint=type_hint
    )

mark_provider

mark_provider(func: ProviderVar | None = None, *, weight: float = 0, type_hint: object | None = None, store: str | Store | None = None) -> Callable[[ProviderVar], ProviderVar] | ProviderVar

Decorate func as a provider in store or the global store.

See Store.mark_provider for details.

Source code in in_n_out/_global.py

@_add_store_to_doc
def mark_provider(
    func: ProviderVar | None = None,
    *,
    weight: float = 0,
    type_hint: object | None = None,
    store: str | Store | None = None,
) -> Callable[[ProviderVar], ProviderVar] | ProviderVar:
    """Decorate `func` as a provider in `store` or the global store.

    See [`Store.mark_provider`][in_n_out.Store.mark_provider] for details.
    """
    return _store_or_global(store).mark_provider(
        func, weight=weight, type_hint=type_hint
    )

process

process(result: Any, *, type_hint: object | type[T] | None = None, first_processor_only: bool = False, raise_exception: bool = False, store: str | Store | None = None) -> None

Process an instance of type_ with processors from store or the global store.

See Store.process for details.

Source code in in_n_out/_global.py

@_add_store_to_doc
def process(
    result: Any,
    *,
    type_hint: object | type[T] | None = None,
    first_processor_only: bool = False,
    raise_exception: bool = False,
    store: str | Store | None = None,
) -> None:
    """Process an instance of `type_` with processors from `store` or the global store.

    See [`Store.process`][in_n_out.Store.process] for details.
    """
    return _store_or_global(store).process(
        result=result,
        type_hint=type_hint,
        first_processor_only=first_processor_only,
        raise_exception=raise_exception,
    )

provide

provide(type_hint: type[T], store: str | Store | None = None) -> T | None

Provide an instance of type_hint with providers from store or the global store.

See Store.provide for details.

Source code in in_n_out/_global.py

@_add_store_to_doc
def provide(
    type_hint: type[T],
    store: str | Store | None = None,
) -> T | None:
    """Provide an instance of `type_hint` with providers from `store` or the global store.

    See [`Store.provide`][in_n_out.Store.provide] for details.
    """  # noqa: E501
    return _store_or_global(store).provide(type_hint=type_hint)

register

register(*, processors: ProcessorIterable | None = None, providers: ProviderIterable | None = None, store: str | Store | None = None) -> InjectionContext

Register multiple providers and/or processors in store or the global store.

See Store.register for details.

Source code in in_n_out/_global.py

@_add_store_to_doc
def register(
    *,
    processors: ProcessorIterable | None = None,
    providers: ProviderIterable | None = None,
    store: str | Store | None = None,
) -> InjectionContext:
    """Register multiple providers and/or processors in `store` or the global store.

    See [`Store.register`][in_n_out.Store.register] for details.
    """
    return _store_or_global(store).register(providers=providers, processors=processors)

register_processor

register_processor(processor: Processor, type_hint: object | None = None, weight: float = 0, store: str | Store | None = None) -> InjectionContext

Register a processor in store or the global store.

See Store.register_processor for details.

Source code in in_n_out/_global.py

@_add_store_to_doc
def register_processor(
    processor: Processor,
    type_hint: object | None = None,
    weight: float = 0,
    store: str | Store | None = None,
) -> InjectionContext:
    """Register a processor in `store` or the global store.

    See [`Store.register_processor`][in_n_out.Store.register_processor] for details.
    """
    return _store_or_global(store).register_processor(
        processor=processor, type_hint=type_hint, weight=weight
    )

register_provider

register_provider(provider: Provider, type_hint: object | None = None, weight: float = 0, store: str | Store | None = None) -> InjectionContext

Register a provider in store or the global store.

See Store.register_provider for details.

Source code in in_n_out/_global.py

@_add_store_to_doc
def register_provider(
    provider: Provider,
    type_hint: object | None = None,
    weight: float = 0,
    store: str | Store | None = None,
) -> InjectionContext:
    """Register a provider in `store` or the global store.

    See [`Store.register_provider`][in_n_out.Store.register_provider] for details.
    """
    return _store_or_global(store).register_provider(
        provider=provider, type_hint=type_hint, weight=weight
    )

resolve_single_type_hints

resolve_single_type_hints(*objs: Any, globalns: dict | None = None, localns: dict | None = None, include_extras: bool = False) -> tuple[Any, ...]

Get type hints for one or more isolated type annotations.

Wrapper around resolve_type_hints (see docstring for that function for parameter docs).

typing.get_type_hints only works for modules, classes, methods, or functions, but the typing module doesn't make the underlying type evaluation logic publicly available. This function creates a small mock object with an __annotations__ dict that will work as an argument to typing.get_type_hints(). It then extracts the resolved hints back into a tuple of hints corresponding to the input objects.

Returns:

• tuple[Any, ...] –

  Tuple

Examples:

>>> resolve_single_type_hints("hi", localns={"hi": typing.Any})
(typing.Any,)

Source code in in_n_out/_type_resolution.py

def resolve_single_type_hints(
    *objs: Any,
    globalns: dict | None = None,
    localns: dict | None = None,
    include_extras: bool = False,
) -> tuple[Any, ...]:
    """Get type hints for one or more isolated type annotations.

    Wrapper around [`resolve_type_hints`][in_n_out.resolve_type_hints]
    (see docstring for that function for parameter docs).

    [`typing.get_type_hints`][typing.get_type_hints] only works for modules, classes,
    methods, or functions, but the typing module doesn't make the underlying type
    evaluation logic publicly available. This function creates a small mock object with
    an `__annotations__` dict that will work as an argument to
    `typing.get_type_hints()`.  It then extracts the resolved hints back into a tuple of
    hints corresponding to the input objects.

    Returns
    -------
    tuple[Any, ...]
        Tuple

    Examples
    --------
    >>> resolve_single_type_hints("hi", localns={"hi": typing.Any})
    (typing.Any,)
    """
    annotations = {str(n): v for n, v in enumerate(objs)}
    mock_obj = type("_T", (), {"__annotations__": annotations})()
    hints = resolve_type_hints(
        mock_obj, globalns=globalns, localns=localns, include_extras=include_extras
    )
    return tuple(hints[k] for k in annotations)

resolve_type_hints

resolve_type_hints(obj: _get_type_hints_obj_allowed_types, globalns: dict | None = None, localns: dict | None = None, include_extras: bool = False) -> dict[str, Any]

Return type hints for an object.

This is a small wrapper around typing.get_type_hints() that adds namespaces to the global and local namespaces.

see docstring for typing.get_type_hints.

Parameters:

• obj (module, class, method, or function) –

  must be a module, class, method, or function.

• globalns (dict | None, default: None ) –

  optional global namespace, by default None.

• localns (dict | None, default: None ) –

  optional local namespace, by default None.

• include_extras (bool, default: False ) –

  If False (the default), recursively replaces all 'Annotated[T, ...]' with 'T'.

Returns:

• dict[str, Any] –

  mapping of object name to type hint for all annotated attributes of obj.

Source code in in_n_out/_type_resolution.py

def resolve_type_hints(
    obj: _get_type_hints_obj_allowed_types,
    globalns: dict | None = None,
    localns: dict | None = None,
    include_extras: bool = False,
) -> dict[str, Any]:
    """Return type hints for an object.

    This is a small wrapper around `typing.get_type_hints()` that adds
    namespaces to the global and local namespaces.

    see docstring for [`typing.get_type_hints`][typing.get_type_hints].

    Parameters
    ----------
    obj : module, class, method, or function
        must be a module, class, method, or function.
    globalns : dict | None
        optional global namespace, by default None.
    localns : dict | None
        optional local namespace, by default None.
    include_extras : bool
        If `False` (the default), recursively replaces all 'Annotated[T, ...]'
        with 'T'.

    Returns
    -------
    dict[str, Any]
        mapping of object name to type hint for all annotated attributes of `obj`.
    """
    _localns = dict(_typing_names())
    if localns:
        _localns.update(localns)  # explicitly provided locals take precedence
    return typing.get_type_hints(
        _unwrap_partial(obj),
        globalns=globalns,
        localns=_localns,
        include_extras=include_extras,
    )

type_resolved_signature

type_resolved_signature(func: Callable, *, localns: dict | None = None, raise_unresolved_optional_args: bool = True, raise_unresolved_required_args: bool = True, guess_self: bool = True) -> Signature

Return a Signature object for a function with resolved type annotations.

Parameters:

• func (Callable) –

  A callable object.

• localns (dict | None, default: None ) –

  Optional local namespace for name resolution, by default None

• raise_unresolved_optional_args (bool, default: True ) –

  Whether to raise an exception when an optional parameter (one with a default value) has an unresolvable type annotation, by default True

• raise_unresolved_required_args (bool, default: True ) –

  Whether to raise an exception when a required parameter has an unresolvable type annotation, by default True

• guess_self (bool, default: True ) –

  Whether to infer the type of the first argument if the function is an unbound class method. This is done as follows: - if '.' (but not '') is in the function's qualname - and if the first parameter is named 'self' or starts with "_" - and if the first parameter annotation is inspect.empty - then the name preceding func.__name__ in the function's qualname (which is usually the class name), is looked up in the function's __globals__ namespace. If found, it is used as the first parameter's type annotation. This allows class methods to be injected with instances of the class.

Returns:

• Signature –

  inspect.Signature object with fully resolved type annotations, (or at least partially resolved type annotations if raise_unresolved_optional_args is False).

Raises:

• NameError –

  If a required argument has an unresolvable type annotation, or if raise_unresolved_optional_args is True and an optional argument has an unresolvable type annotation.

Source code in in_n_out/_type_resolution.py

def type_resolved_signature(
    func: Callable,
    *,
    localns: dict | None = None,
    raise_unresolved_optional_args: bool = True,
    raise_unresolved_required_args: bool = True,
    guess_self: bool = True,
) -> Signature:
    """Return a Signature object for a function with resolved type annotations.

    Parameters
    ----------
    func : Callable
        A callable object.
    localns : dict | None
        Optional local namespace for name resolution, by default None
    raise_unresolved_optional_args : bool
        Whether to raise an exception when an optional parameter (one with a default
        value) has an unresolvable type annotation, by default True
    raise_unresolved_required_args : bool
        Whether to raise an exception when a required parameter has an unresolvable
        type annotation, by default True
    guess_self : bool
        Whether to infer the type of the first argument if the function is an unbound
        class method. This is done as follows:
            - if '.' (but not '<locals>') is in the function's __qualname__
            - and if the first parameter is named 'self' or starts with "_"
            - and if the first parameter annotation is `inspect.empty`
            - then the name preceding `func.__name__` in the function's __qualname__
              (which is usually the class name), is looked up in the function's
              `__globals__` namespace. If found, it is used as the first parameter's
              type annotation.
        This allows class methods to be injected with instances of the class.

    Returns
    -------
    Signature
        [`inspect.Signature`][inspect.Signature] object with fully resolved type
        annotations,
        (or at least partially resolved type annotations if
        `raise_unresolved_optional_args` is `False`).

    Raises
    ------
    NameError
        If a required argument has an unresolvable type annotation, or if
        `raise_unresolved_optional_args` is `True` and an optional argument has
        an unresolvable type annotation.
    """
    sig = Signature.from_callable(func)
    hints = {}
    if guess_self and sig.parameters:
        p0 = next(iter(sig.parameters.values()))
        # The best identifier i can figure for a class method is that:
        # 1. its qualname contains a period (e.g. "MyClass.my_method"),
        # 2. the first parameter tends to be named "self", or some private variable
        # 3. the first parameter tends to be unannotated
        qualname = getattr(func, "__qualname__", "")
        if (
            "." in qualname
            and "<locals>" not in qualname  # don't support locally defd types
            and (p0.name == "self" or p0.name.startswith("_"))
            and p0.annotation is p0.empty
        ):
            # look up the class name in the function's globals
            cls_name = qualname.replace(func.__name__, "").rstrip(".")
            func_globals = getattr(func, "__globals__", {})
            if cls_name in func_globals:
                # add it to the type hints
                hints = {p0.name: func_globals[cls_name]}

    try:
        hints.update(resolve_type_hints(func, localns=localns))
    except (NameError, TypeError) as err:
        if raise_unresolved_optional_args:
            raise NameError(
                f"Could not resolve all annotations in signature {sig} ({err}). "
                "To allow optional parameters and return types to remain unresolved, "
                "use `raise_unresolved_optional_args=False`"
            ) from err
        hints = _resolve_params_one_by_one(
            sig,
            globalns=getattr(func, "__globals__", None),
            localns=localns,
            exclude_unresolved_mandatory=not raise_unresolved_required_args,
        )

    resolved_parameters = [
        param.replace(annotation=hints.get(param.name, param.empty))
        for param in sig.parameters.values()
    ]
    return sig.replace(
        parameters=resolved_parameters,
        return_annotation=hints.get("return", sig.empty),
    )