Function aggregation_specs_to_agg_ops

sdk/python/feast/aggregation/__init__.py:126–138

(
    agg_specs: Iterable[Any],
    *,
    time_window_unsupported_error_message: str,
)

def aggregation_specs_to_agg_ops(
    agg_specs: Iterable[Any],
    *,
    time_window_unsupported_error_message: str,
) -> Dict[str, Tuple[str, str]]:
    agg_ops: Dict[str, Tuple[str, str]] = {}
    for agg in agg_specs:
        if getattr(agg, "time_window", None) is not None:
            raise ValueError(time_window_unsupported_error_message)
        alias = getattr(agg, "name", None) or f"{agg.function}_{agg.column}"
        func_name = _FUNCTION_ALIASES.get(agg.function, agg.function)
        agg_ops[alias] = (func_name, agg.column)
    return agg_ops


__all__ = ["Aggregation", "aggregation_specs_to_agg_ops"]