Skip to content

Extraction

extract_from_event_type(annotation: Any) -> None | tuple[Any, AnnotationDetails]

Handle plain Event types - user wants the full event passed through.

Returns:

Type Description
None | tuple[Any, AnnotationDetails]

Tuple of (Event type, identity function) if annotation is Event subclass.
None | tuple[Any, AnnotationDetails]

None otherwise.
Source code in src/hassette/bus/extraction.py 
33
34
35
36
37
38
39
40
41
42
43
def extract_from_event_type(annotation: Any) -> None | tuple[Any, AnnotationDetails]:
    """Handle plain Event types - user wants the full event passed through.

    Returns:
        Tuple of (Event type, identity function) if annotation is Event subclass.
        None otherwise.
    """
    if not is_event_type(annotation):
        return None

    return (annotation, AnnotationDetails(extractor=identity))

has_dependency_injection(signature: Signature) -> bool

Check if a signature uses any dependency injection.

Source code in src/hassette/bus/extraction.py 
46
47
48
49
50
51
52
53
54
55
def has_dependency_injection(signature: Signature) -> bool:
    """Check if a signature uses any dependency injection."""
    for param in signature.parameters.values():
        if param.annotation is Parameter.empty:
            continue

        if is_annotated_type(param.annotation) or is_event_type(param.annotation):
            return True

    return False

validate_di_signature(signature: Signature) -> None

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

Raises:

Type Description
ValueError

If signature has VAR_POSITIONAL (*args) or POSITIONAL_ONLY (/) parameters.
Source code in src/hassette/bus/extraction.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def validate_di_signature(signature: Signature) -> None:
    """Validate that a signature with DI doesn't have incompatible parameter types.

    Raises:
        ValueError: 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}"
            )

extract_from_signature(signature: Signature) -> dict[str, tuple[Any, AnnotationDetails[Any]]]

Extract parameter types and extractors from a function signature.

Returns a dict mapping parameter name to (type, extractor_callable). Validates that DI signatures don't have incompatible parameter kinds.

Raises:

Type Description
ValueError

If signature has incompatible parameters with DI.
Source code in src/hassette/bus/extraction.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def extract_from_signature(signature: Signature) -> dict[str, tuple[Any, AnnotationDetails[Any]]]:
    """Extract parameter types and extractors from a function signature.

    Returns a dict mapping parameter name to (type, extractor_callable).
    Validates that DI signatures don't have incompatible parameter kinds.

    Raises:
        ValueError: If signature has incompatible parameters with DI.
    """
    # Validate signature first
    validate_di_signature(signature)

    param_details: dict[str, tuple[Any, AnnotationDetails[Any]]] = {}

    for param in signature.parameters.values():
        annotation = param.annotation

        # Skip parameters without annotations
        if annotation is Parameter.empty:
            continue

        result = extract_from_annotated(annotation) or extract_from_event_type(annotation)

        if result:
            param_details[param.name] = result

    return param_details