227 lines
7.5 KiB
Python
227 lines
7.5 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
"""Decorator-driven argument aspects and their invocation lifetime."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass
|
|
from functools import wraps
|
|
import inspect
|
|
from types import FrameType
|
|
from typing import (
|
|
Any,
|
|
Callable,
|
|
Dict,
|
|
Generic,
|
|
List,
|
|
Set,
|
|
Tuple,
|
|
TypeVar,
|
|
cast,
|
|
get_args,
|
|
get_origin,
|
|
get_type_hints,
|
|
)
|
|
|
|
from .aspect_value import AspectValue
|
|
|
|
|
|
T = TypeVar("T")
|
|
F = TypeVar("F", bound=Callable[..., Any])
|
|
|
|
|
|
class AspectError(RuntimeError):
|
|
"""AspectError reports invalid aspect declarations or resolution failures."""
|
|
|
|
|
|
class AspectAnnotationError(AspectError):
|
|
"""AspectAnnotationError reports an unusable aspect-bearing annotation."""
|
|
|
|
|
|
class AspectResolutionError(AspectError):
|
|
"""AspectResolutionError reports a missing caller-context value."""
|
|
|
|
|
|
class AspectAmbiguityError(AspectError):
|
|
"""AspectAmbiguityError reports multiple compatible caller-context values."""
|
|
|
|
|
|
class Aspect(AspectValue[T], ABC, Generic[T]):
|
|
"""Aspect resolves one typed value from one decorated invocation."""
|
|
|
|
def __init__(
|
|
self,
|
|
frame: FrameType,
|
|
args: tuple[Any, ...],
|
|
kwargs: dict[str, Any],
|
|
target_type: type[T],
|
|
) -> None:
|
|
"""Capture invocation context before resolving this aspect's value."""
|
|
|
|
self.frame: FrameType | None = frame
|
|
self.args = args
|
|
self.kwargs = kwargs
|
|
self.target_type = target_type
|
|
super().__init__(self.create())
|
|
|
|
@abstractmethod
|
|
def create(self) -> T:
|
|
"""Create the typed value represented by this aspect."""
|
|
|
|
def release(self) -> None:
|
|
"""Discard live invocation state after the decorated call completes."""
|
|
|
|
self.frame = None
|
|
self.args = ()
|
|
self.kwargs = {}
|
|
self._release_value()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _AspectParameter:
|
|
"""_AspectParameter records static metadata for one injectable parameter."""
|
|
|
|
name: str
|
|
aspect_type: type[Aspect[Any]]
|
|
target_type: type[Any]
|
|
|
|
|
|
def _definition_types() -> dict[str, type[Any]]:
|
|
"""Return nearby type declarations without retaining their live frames."""
|
|
|
|
current = inspect.currentframe()
|
|
decorator = current.f_back if current is not None else None
|
|
definition = decorator.f_back if decorator is not None else None
|
|
enclosing = definition.f_back if definition is not None else None
|
|
types: dict[str, type[Any]] = {}
|
|
# Preserve local classes used by postponed annotations at the declaration site.
|
|
for frame in (definition, enclosing):
|
|
if frame is None:
|
|
continue
|
|
types.update(
|
|
{
|
|
name: value
|
|
for name, value in frame.f_locals.items()
|
|
if isinstance(value, type)
|
|
}
|
|
)
|
|
del current, decorator, definition, enclosing
|
|
return types
|
|
|
|
|
|
def _aspect_parameters(
|
|
function: Callable[..., Any],
|
|
signature: inspect.Signature,
|
|
definition_types: dict[str, type[Any]],
|
|
) -> tuple[_AspectParameter, ...]:
|
|
"""Return the aspect-bearing parameters declared by one callable."""
|
|
|
|
try:
|
|
compatibility_globals = dict(function.__globals__)
|
|
compatibility_globals.update(
|
|
{"dict": Dict, "list": List, "set": Set, "tuple": Tuple}
|
|
)
|
|
hints = get_type_hints(
|
|
function,
|
|
globalns=compatibility_globals,
|
|
localns=definition_types,
|
|
)
|
|
except (NameError, TypeError) as error:
|
|
raise AspectAnnotationError(
|
|
f"Cannot resolve aspect annotations for {function.__qualname__}: {error}"
|
|
) from error
|
|
parameters: list[_AspectParameter] = []
|
|
# Select declared Aspect subtypes rather than coupling interception to Caller.
|
|
for name, parameter in signature.parameters.items():
|
|
annotation = hints.get(name, parameter.annotation)
|
|
origin = get_origin(annotation) or annotation
|
|
if not isinstance(origin, type) or not issubclass(origin, Aspect):
|
|
continue
|
|
arguments = get_args(annotation)
|
|
# Require one concrete runtime search type for every declared aspect.
|
|
if len(arguments) != 1 or not isinstance(arguments[0], type):
|
|
raise AspectAnnotationError(
|
|
f"{function.__qualname__}.{name} must declare one runtime target type."
|
|
)
|
|
# Preserve Python's variadic binding semantics rather than manufacturing a value list.
|
|
if parameter.kind in (
|
|
inspect.Parameter.VAR_POSITIONAL,
|
|
inspect.Parameter.VAR_KEYWORD,
|
|
):
|
|
raise AspectAnnotationError(
|
|
f"{function.__qualname__}.{name} cannot be a variadic aspect parameter."
|
|
)
|
|
parameters.append(
|
|
_AspectParameter(
|
|
name=name,
|
|
aspect_type=cast("type[Aspect[Any]]", origin),
|
|
target_type=arguments[0],
|
|
)
|
|
)
|
|
return tuple(parameters)
|
|
|
|
|
|
def _resolution_error(
|
|
error: AspectError,
|
|
function: Callable[..., Any],
|
|
parameter: _AspectParameter,
|
|
) -> AspectError:
|
|
"""Attach decorated-call context to one aspect resolution failure."""
|
|
|
|
return type(error)(
|
|
f"{function.__qualname__}.{parameter.name} "
|
|
f"({parameter.aspect_type.__name__}[{parameter.target_type.__name__}]): {error}"
|
|
)
|
|
|
|
|
|
def aspect(function: F) -> F:
|
|
"""Inject omitted Aspect-annotated parameters for one invocation."""
|
|
|
|
signature = inspect.signature(function)
|
|
definition_types = _definition_types()
|
|
|
|
@wraps(function)
|
|
def wrapper(*args: Any, **kwargs: Any) -> Any:
|
|
"""Bind ordinary arguments, inject aspects, and release them afterward."""
|
|
|
|
parameters = _aspect_parameters(function, signature, definition_types)
|
|
# Preserve ordinary wrapped functions without creating caller-resolution state.
|
|
if not parameters:
|
|
return function(*args, **kwargs)
|
|
bound = signature.bind_partial(*args, **kwargs)
|
|
current = inspect.currentframe()
|
|
caller_frame = current.f_back if current is not None else None
|
|
# Aspect resolution requires the direct caller rather than an internal wrapper frame.
|
|
if caller_frame is None:
|
|
raise AspectResolutionError(
|
|
f"Cannot resolve caller context for {function.__qualname__}."
|
|
)
|
|
injected: list[Aspect[Any]] = []
|
|
try:
|
|
# Fill only omitted aspect parameters, preserving all explicit values unchanged.
|
|
for parameter in parameters:
|
|
if parameter.name in bound.arguments:
|
|
continue
|
|
try:
|
|
resolved = parameter.aspect_type(
|
|
caller_frame,
|
|
tuple(args),
|
|
dict(kwargs),
|
|
parameter.target_type,
|
|
)
|
|
except AspectError as error:
|
|
raise _resolution_error(error, function, parameter) from error
|
|
bound.arguments[parameter.name] = resolved
|
|
injected.append(resolved)
|
|
complete = signature.bind(*bound.args, **bound.kwargs)
|
|
return function(*complete.args, **complete.kwargs)
|
|
finally:
|
|
# Release every injected value before this wrapper frame returns to its caller.
|
|
for resolved in injected:
|
|
resolved.release()
|
|
del caller_frame, current
|
|
|
|
return cast(F, wrapper)
|