aeat.application.aggregation._grouping module

Shared group-by + name-cache helper for per-modelo aggregators.

Used by: _retenciones, _counterpart to bucket observations and cache canonical names.

Both implement the same shape of aggregation: bucket observations by a composite key, then roll up each bucket. They additionally need to resolve a stable human-readable name per (source_kind, identity_nif) pair across multiple observations.

This module extracts that shared mechanism. The per-domain aggregators retain their domain-specific rollup composition (e.g. counterpart adds country + readiness fields) — only the group-and-name-cache step is shared.

group_and_collect_names(observations, *, group_key_fn, identity_key_fn, name_fn)[source]

Bucket observations by group key and cache a canonical name per identity.

Behaviour invariants (shared by both per-modelo aggregators):

  • Iteration order of observations is preserved within each bucket.

  • The first non-empty name_fn(obs) per identity_key_fn(obs) wins; later non-empty names for the same identity are discarded.

  • An empty / falsy name is skipped (does not overwrite a prior win).

Parameters:
Return type:

tuple[dict[TypeVar(GroupKey, bound= tuple[object, ...]), list[TypeVar(T)]], dict[TypeVar(IdentityKey, bound= tuple[object, ...]), str]]

Returns:

A two-tuple (grouped, names) where grouped maps each group_key_fn(obs) to the list of observations sharing that key (insertion order), and names maps each identity_key_fn(obs) to the first non-empty name observed.

filter_observations_for_modelo(observations, *, modelo, catalogue, attribute_fn, aggregator_label)[source]

Keep observations whose classifying attribute is in-scope for modelo.

Shared by both per-modelo aggregators: _counterpart filters on operation_kind against an OperationKind347 / 349 catalogue; _retenciones filters on scheme against a RetencionScheme catalogue. The only per-domain inputs are the catalogue, the attribute getter, and the label used in the unsupported-modelo error.

Parameters:

  • observations (tuple[TypeVar(T), ...]) – Typed observation records to filter.

  • modelo (str) – The requested modelo code; must key into catalogue.

  • catalogue (Mapping[str, Container[TypeVar(AttrValue)]]) – Maps each supported modelo code to the container of eligible attribute values.

  • attribute_fn (Callable[[TypeVar(T)], TypeVar(AttrValue)]) – Extracts the classifying attribute from each observation.

  • aggregator_label (str) – Human-readable aggregator name for the AggregationUnsupportedModeloError message.

Raises:

AggregationUnsupportedModeloError – When modelo is not a key in catalogue.

Return type:

tuple[TypeVar(T), ...]

Returns:

The observations whose classifying attribute is eligible for modelo, in input order.