Di

Shared dependency injection primitives.

Consumers (the event bus, and future consumers like the scheduler) use this package to inspect handler/predicate signatures, build an injection plan, and resolve kwargs at dispatch time - without needing to know about each other's source types.

See design/specs/004-extract-event-handling-di/design.md for the full architecture.

CallableInvoker dataclass

Resolves a dependency injection plan into kwargs at call time.

Does not store or call the target callable - the caller is responsible for invoking the target with the resolved kwargs. This keeps the shared layer agnostic to how (or whether) the target is sync/async, a function, or a bound method.

Source code in src/hassette/di/invoker.py

@dataclass(frozen=True, slots=True)
class CallableInvoker:
    """Resolves a dependency injection plan into kwargs at call time.

    Does not store or call the target callable - the caller is responsible for invoking
    the target with the resolved kwargs. This keeps the shared layer agnostic to how (or
    whether) the target is sync/async, a function, or a bound method.
    """

    params: tuple[InjectionParam, ...]

    def invoke(self, available: dict[type, Any]) -> dict[str, Any]:
        """Build kwargs for the target callable from `available` source objects.

        Args:
            available: Maps a source type to the live object of that type, e.g.
                `{Event: event}`. Lookup is exact-match - no `isinstance` fallback.

        Raises:
            KeyError: If a param's `source_type` has no entry in `available`.
        """
        return {param.name: param.extractor(available[param.source_type]) for param in self.params}

invoke(available: dict[type, Any]) -> dict[str, Any]

Build kwargs for the target callable from available source objects.

Parameters:

Name	Type	Description	Default
available	dict[type, Any]	Maps a source type to the live object of that type, e.g. {Event: event}. Lookup is exact-match - no isinstance fallback.	required

Raises:

Type	Description
KeyError	If a param's source_type has no entry in available.

Source code in src/hassette/di/invoker.py

def invoke(self, available: dict[type, Any]) -> dict[str, Any]:
    """Build kwargs for the target callable from `available` source objects.

    Args:
        available: Maps a source type to the live object of that type, e.g.
            `{Event: event}`. Lookup is exact-match - no `isinstance` fallback.

    Raises:
        KeyError: If a param's `source_type` has no entry in `available`.
    """
    return {param.name: param.extractor(available[param.source_type]) for param in self.params}

AnnotatedMatcher dataclass

Matches Annotated[T, metadata] parameters.

metadata may be an AnnotationDetails instance or a bare callable extractor - bare callables are auto-wrapped into AnnotationDetails(extractor=metadata). Any other metadata shape emits a warning and is treated as a non-match.

Source code in src/hassette/di/matchers.py

@dataclass(frozen=True, slots=True)
class AnnotatedMatcher:
    """Matches `Annotated[T, metadata]` parameters.

    `metadata` may be an `AnnotationDetails` instance or a bare callable extractor - bare
    callables are auto-wrapped into `AnnotationDetails(extractor=metadata)`. Any other
    metadata shape emits a warning and is treated as a non-match.
    """

    source_type: type

    def match(self, param: inspect.Parameter) -> InjectionParam | None:
        annotation = param.annotation
        if annotation is inspect.Parameter.empty:
            return None

        if not is_annotated_type(annotation):
            return None

        type_details = get_type_and_details(annotation)
        if type_details is None:
            return None

        inner_type, metadata = type_details

        if isinstance(metadata, AnnotationDetails):
            details = metadata
        elif callable(metadata):
            details = AnnotationDetails(extractor=metadata)
        else:
            warn(
                f"Invalid Annotated metadata: {metadata} is not AnnotationDetails or callable extractor",
                stacklevel=2,
            )
            return None

        target_type = normalize_annotation(inner_type)

        return InjectionParam(
            name=param.name,
            source_type=details.source_type or self.source_type,
            target_type=target_type,
            extractor=details.extractor,
            converter=details.converter,
        )

TypeMatcher dataclass

Matches parameters whose (unwrapped) annotation is match_type or a subclass of it.

Parameterized generics (e.g. Event[Any]) are unwrapped to their origin before the subclass check, matching the behavior of hassette.events.type_checks.is_event_type.

Source code in src/hassette/di/matchers.py

@dataclass(frozen=True, slots=True)
class TypeMatcher:
    """Matches parameters whose (unwrapped) annotation is `match_type` or a subclass of it.

    Parameterized generics (e.g. `Event[Any]`) are unwrapped to their origin before the
    subclass check, matching the behavior of `hassette.events.type_checks.is_event_type`.
    """

    match_type: type

    def match(self, param: inspect.Parameter) -> InjectionParam | None:
        annotation = param.annotation
        if annotation is inspect.Parameter.empty:
            return None

        origin = get_origin(annotation)
        base_type = origin or annotation

        matched = inspect.isclass(base_type) and issubclass(base_type, self.match_type)
        if not matched and (origin is types.UnionType or origin is typing.Union):
            args = get_args(annotation)
            matched = any(inspect.isclass(arg) and issubclass(arg, self.match_type) for arg in args)

        if not matched:
            return None

        return InjectionParam(
            name=param.name,
            source_type=self.match_type,
            target_type=annotation,
            extractor=identity,
        )

AnnotationDetails dataclass

Bases: Generic[T]

Details about an annotation used for dependency injection.

Source code in src/hassette/di/types.py

@dataclass(slots=True, frozen=True)
class AnnotationDetails(Generic[T]):
    """Details about an annotation used for dependency injection."""

    extractor: Callable[[T], Any]
    """Function to extract the dependency from the source object."""

    converter: Callable[[Any, Any], Any] | None = None
    """Optional converter function to convert the extracted value to the desired type."""

    source_type: type[T] | None = None
    """The source type this extractor operates on, e.g. `Event` or `ScheduledJob`.

    When `None`, the consuming matcher (e.g. `AnnotatedMatcher`) falls back to its own
    constructor-provided `source_type`. Set this to override the matcher's default on a
    per-annotation basis - for example, an extractor that reads from `StateManager`
    alongside other `Event`-sourced extractors on the same handler.
    """

extractor: Callable[[T], Any] instance-attribute

Function to extract the dependency from the source object.

converter: Callable[[Any, Any], Any] | None = None class-attribute instance-attribute

Optional converter function to convert the extracted value to the desired type.

source_type: type[T] | None = None class-attribute instance-attribute

The source type this extractor operates on, e.g. Event or ScheduledJob.

When None, the consuming matcher (e.g. AnnotatedMatcher) falls back to its own constructor-provided source_type. Set this to override the matcher's default on a per-annotation basis - for example, an extractor that reads from StateManager alongside other Event-sourced extractors on the same handler.

InjectionParam dataclass

One parameter's resolved injection plan.

Produced by a ParameterMatcher during plan building (build_injection_plan) and consumed by CallableInvoker.invoke at dispatch time.

Source code in src/hassette/di/types.py

@dataclass(frozen=True, slots=True)
class InjectionParam:
    """One parameter's resolved injection plan.

    Produced by a `ParameterMatcher` during plan building (`build_injection_plan`) and
    consumed by `CallableInvoker.invoke` at dispatch time.
    """

    name: str
    """The parameter name on the target callable."""

    source_type: type
    """Key to look up in the `available` dict passed to `CallableInvoker.invoke`."""

    target_type: Any
    """The annotation's type, e.g. `LightState` - used by consumers that convert the
    extracted value (the shared layer itself does not convert)."""

    extractor: Callable[[Any], Any]
    """Function that pulls the value out of the source object."""

    converter: Callable[[Any, Any], Any] | None = None
    """Optional converter carried through from `AnnotationDetails.converter`."""

name: str instance-attribute

The parameter name on the target callable.

source_type: type instance-attribute

Key to look up in the available dict passed to CallableInvoker.invoke.

target_type: Any instance-attribute

The annotation's type, e.g. LightState - used by consumers that convert the extracted value (the shared layer itself does not convert).

extractor: Callable[[Any], Any] instance-attribute

Function that pulls the value out of the source object.

converter: Callable[[Any, Any], Any] | None = None class-attribute instance-attribute

Optional converter carried through from AnnotationDetails.converter.

ParameterMatcher

Bases: Protocol

Protocol for matching a single inspect.Parameter to an InjectionParam.

Implementations inspect a parameter's annotation and return an InjectionParam when they recognize it, or None when they don't. build_injection_plan tries a sequence of matchers in order and takes the first match.

Source code in src/hassette/di/types.py

class ParameterMatcher(Protocol):
    """Protocol for matching a single `inspect.Parameter` to an `InjectionParam`.

    Implementations inspect a parameter's annotation and return an `InjectionParam` when
    they recognize it, or `None` when they don't. `build_injection_plan` tries a sequence
    of matchers in order and takes the first match.
    """

    def match(self, param: inspect.Parameter) -> InjectionParam | None: ...

build_injection_plan(sig: Signature, matchers: Sequence[ParameterMatcher]) -> tuple[InjectionParam, ...]

Walk a signature's parameters and build a dependency injection plan.

Each parameter is tried against matchers in order; the first matcher that returns a non-None InjectionParam wins. Parameters without annotations, and annotated parameters that no matcher recognizes, are silently skipped.

Raises:

Type	Description
DependencyInjectionError	If the signature has VAR_POSITIONAL or POSITIONAL_ONLY parameters (see validate_di_signature).

Source code in src/hassette/di/plan.py

def build_injection_plan(sig: Signature, matchers: Sequence[ParameterMatcher]) -> tuple[InjectionParam, ...]:
    """Walk a signature's parameters and build a dependency injection plan.

    Each parameter is tried against `matchers` in order; the first matcher that returns a
    non-`None` `InjectionParam` wins. Parameters without annotations, and annotated
    parameters that no matcher recognizes, are silently skipped.

    Raises:
        DependencyInjectionError: If the signature has VAR_POSITIONAL or POSITIONAL_ONLY
            parameters (see `validate_di_signature`).
    """
    validate_di_signature(sig)

    params: list[InjectionParam] = []
    for param in sig.parameters.values():
        if param.annotation is Parameter.empty:
            continue

        for matcher in matchers:
            result = matcher.match(param)
            if result is not None:
                params.append(result)
                break

    return tuple(params)

validate_di_signature(signature: Signature) -> None

Validate that a signature with DI doesn't have incompatible parameter types.

Raises:

Type	Description
DependencyInjectionError	If signature has VAR_POSITIONAL (*args) or POSITIONAL_ONLY (/) parameters.

Source code in src/hassette/di/plan.py

def validate_di_signature(signature: Signature) -> None:
    """Validate that a signature with DI doesn't have incompatible parameter types.

    Raises:
        DependencyInjectionError: If signature has VAR_POSITIONAL (*args) or
            POSITIONAL_ONLY (/) parameters.
    """
    for param in signature.parameters.values():
        if param.kind == Parameter.VAR_POSITIONAL:
            raise DependencyInjectionError(
                f"Handler with dependency injection cannot have *args parameter: {param.name}"
            )

        if param.kind == Parameter.POSITIONAL_ONLY:
            raise DependencyInjectionError(
                f"Handler with dependency injection cannot have positional-only parameter: {param.name}"
            )

identity(x: Any) -> Any

Identity function - returns the input as-is.

Used when a parameter needs the full source object without transformation.

Source code in src/hassette/di/types.py

def identity(x: Any) -> Any:
    """Identity function - returns the input as-is.

    Used when a parameter needs the full source object without transformation.
    """
    return x