from collections.abc import Callable, Mapping
from typing import Any
import aspis.internal as Ai
from .curry import curry
def process_mapping(obj: Mapping, fn: Callable) -> Mapping:
"""
Recursively process a mapping structure by applying a function to all leaf values.
Args:
obj: A mapping (dict) to process. May contain nested mappings.
fn: Function to apply to each leaf (non-Mapping) value.
Returns:
Mapping: A new mapping with the same structure but transformed values.
"""
return {k: process_mapping(v, fn) if isinstance(v, Mapping) else fn(v) for k, v in obj.items()}
def has_completed(obj: Mapping) -> bool:
"""
Check if all functions in a spec mapping have been fully applied.
Recursively checks if the mapping contains any Callable values or
nested mappings that haven't been fully applied.
Args:
obj: A mapping to check for completion.
Returns:
bool: True if all functions are fully applied, False otherwise.
"""
return not any(
map(lambda v: (isinstance(v, Mapping) and not has_completed(v)) or isinstance(v, Callable), obj.values())
)
def nested_apply(obj: Mapping, *args: Any, **kwargs: Any) -> Mapping:
"""
Apply arguments to all callable values in a nested mapping structure.
Args:
obj: A mapping containing callable values.
*args: Positional arguments to apply to each callable.
**kwargs: Keyword arguments to apply to each callable.
Returns:
Mapping: A new mapping with arguments applied to all callables.
"""
return process_mapping(obj, lambda v: Ai.eager_partial(v, *args, **kwargs) if isinstance(v, Callable) else v)
def convert_vars(obj: Mapping) -> Mapping:
"""
Convert non-callable, non-Mapping values to empty dictionaries.
This normalizes the spec structure by replacing leaf values that aren't
functions or nested mappings with empty dicts.
Args:
obj: A mapping to normalize.
Returns:
Mapping: A normalized mapping structure.
"""
return process_mapping(
obj,
lambda v: {} if not (isinstance(v, Mapping) or isinstance(v, Callable)) else v,
)
[docs]
@curry
def apply_spec(spec: Mapping, *args: Any, **kwargs: Any) -> Mapping | Callable:
"""
Given a spec object recursively mapping properties to functions,
creates a function producing an object of the same structure,
by mapping each property to the result of calling its associated function
with the supplied arguments.
Args:
spec (dict): A dictionary mapping properties to functions.
*args (Any): The arguments to be passed to the functions.
**kwargs (Any): The keyword arguments to be passed to the functions.
Returns:
Mapping: An object with the same structure as `spec` where each property is the result
of calling the associated function with the arguments.
Example:
>>> from aspis.common import add, multiply
>>> get_metrics = apply_spec({"sum": add, "product": multiply})
>>> get_metrics(2, 3)
{'sum': 5, 'product': 6}
"""
new_spec = convert_vars(spec)
new_spec = nested_apply(new_spec, *args, **kwargs)
if has_completed(new_spec):
return new_spec
return lambda *margs, **mkwargs: nested_apply(new_spec, *margs, **mkwargs)