cadrumo.application.user_profile.projections module

Canonical projections from UserProfileRecord to consumer shapes.

The schema-driven user-profile backend is the canonical authority. Every existing consumer (deadlines, filing runtime, overview calendar, wizard, workflow adapters) reads through one of these projection helpers rather than keeping its own profile fact decoding branch.

The projections compose against a fact map (path -> str(value)) derived from the record. Deadline-engine projection uses the domain coercer so field-level coercion logic stays in one place.

profile_schema_for_record(record, *, schema=None)[source]

Return the schema explicitly supplied by the authority operation.

Core types: UserProfileRecord.

Return type:

ProfileSchemaDefinition

Parameters:
facts_to_values(facts, *, schema=None)[source]

Project a tuple of profile facts into the flat selector -> str(value) map.

The flat shape is the selector-keyed mapping consumed by taxpayer_profile_from_mapping() and similar coercers. Each schema field’s model_selectors are honored: a fact at identity.tax_id whose schema declares model_selectors = ["tax.id"] is emitted under the key tax.id. Facts whose path is not in the schema fall through untranslated.

Return type:

dict[str, str]

Parameters:
record_to_values(record, *, schema=None)[source]

Project a live profile record into the selector-keyed flat values mapping.

Parameters:
Return type:

dict[str, str]

snapshot_to_values(snapshot, *, schema=None)[source]

Project an immutable filing snapshot into the selector-keyed flat values mapping.

Return type:

dict[str, str]

Parameters:
in_window_order(facts)

Order facts so the chronologically last effective window wins.

The record model permits several live facts at one path with different effective-dating windows, and the schema declares effective_dated on most of its sections and fields, so which of them is effective is a real question rather than a hypothetical one. Both projections resolve it by taking the LAST fact per path, so ordering by window here is what makes that answer the in-force one.

The rule is the one the record resolvers already use - sort on valid_from, absent windows first, and let the last win.

valid_to is deliberately NOT consulted, and the reason is that honouring it needs an as_of date the caller supplies, not the clock. Expiry is only meaningful relative to some instant, and the only instant available here is “now”; a projection that dropped expired facts against date.today() would return different values for the same record on different days. These projections feed filing inputs and completeness checks, where the effective instant is the PERIOD being filed rather than the day the command runs - so an implicit-today rule would resolve a 2024 filing against 2026’s calendar, and would do it silently. Dropping a path also reads downstream as unset rather than expired, so a required field whose window closed would surface as missing rather than as ended.

Honouring expiry therefore means threading an explicit as_of through every caller, each deciding the instant its own read is effective at. That is a different change; until it is made, supersession is expressed by recording a later valid_from at the same path, which this ordering already resolves. A valid_to that nothing supersedes is reported to the operator by the profile validation surface rather than silently ignored.

The sort is stable, so a record whose facts carry no window at all keeps its declaration order exactly. Nothing in production sets a window today, which is what makes this behaviour-neutral now and correct when something starts.

Parameters:

facts (Sequence[UserProfileFact]) – The record’s facts, in declaration order.

Return type:

tuple[UserProfileFact, ...]

Returns:

The same facts, ordered so a later window resolves last.

record_to_path_values(record)[source]

Project a UserProfileRecord (or snapshot) into a schema-path-keyed string mapping.

Unlike record_to_values() (which projects via the schema’s model_selectors aliases), this keeps the canonical schema path as the key. The mapping is what the wizard catalogue, validate_profile_values(), and CLI status surfaces consume.

Return type:

dict[str, str]

Parameters:

record (UserProfileRecord | UserProfileSnapshot | None)

fact_value(record, path)[source]

Read one effective current fact through the canonical projection.

Core types: UserProfileRecord.

Return type:

str | None

Parameters:
class EffectiveFact(**data)[source]

Bases: BaseModel

The fact at one schema path whose effective window starts latest.

value is the rendered canonical token, or None when the path was explicitly CLEARED. A cleared path is not the same as an unset one: it carries an operator decision — I do not want this on my profile — and a caller that cannot see the difference will treat the deletion as absence and quietly undo it.

Several facts may be live at one path with different effective-dating windows, and this resolves them the way the record resolvers do: the latest valid_from wins, with an absent window ordering first. That is deliberately the same rule rather than a better one, so no two readers can disagree about which fact is effective.

valid_to is not consulted, so a fact whose window has closed still resolves. That is the shared rule rather than a limit of this projection, and _in_window_order() carries the reason: expiry needs an as_of the caller supplies, and resolving it against the clock would make one record project differently on different days.

Parameters:

data (Any)

value: str | None
source: str
record_to_effective_facts(record)[source]

Project a UserProfileRecord into the effective fact at each schema path.

Unlike record_to_path_values(), which resolves the last fact whose value is not None (correct for callers that want an effective VALUE, since a cleared path has none), this resolves the last fact per path REGARDLESS of value, so an explicit clear stays visible as value=None.

Returning one record per path rather than parallel value and source mappings is deliberate: two mappings can disagree about which fact is effective, and a value adjudicated against a different fact’s provenance is a subtler failure than the one such a split is usually introduced to fix. Here they cannot disagree, because there is only one answer.

Return type:

dict[str, EffectiveFact]

Parameters:

record (UserProfileRecord | UserProfileSnapshot | None)

projection_for_taxpayer(facts, *, tax_id_default='00000000T', iva_regime_default=None, schema=None)[source]

Return the deadline-engine TaxpayerProfile for the supplied profile facts.

Parameters:
Return type:

TaxpayerProfile

The single coercion path goes through taxpayer_profile_from_mapping() so canonical-token semantics stay in lockstep with the wizard descriptor.

profile_path_values_for_bucket(bucket_id)[source]

Wizard-free canonical projection of a bucket’s profile, or None when absent.

Reads the operator profile through record_to_path_values() WITHOUT building the full taxpayer profile, which requires the wizard setup-flow catalogue. That decouples callers from the catalogue so the projection resolves identically in a non-CLI calculation context, where the catalogue may be unregistered, instead of silently failing closed.

Returns None only when there is genuinely no profile for the bucket.

It lives here, beside the projection it delegates to, because two callers each carried a copy that described itself as reading through “the SINGLE projection” while being one of two.

Return type:

dict[str, str] | None

Parameters:

bucket_id (str)

profile_fact_index(record, schema)[source]

Build a selector -> typed-value index covering both selector forms.

A profile binding’s selector resolves either as the canonical section.field fact path (profile_key form) or as a schema model_selector alias (profile_model + field form). The index exposes each non-null fact under its canonical path AND under every model_selector the schema declares for it, so both selector forms find the value.

Values are preserved as their original UserProfileFactValue type (bool, Decimal, date, str, …) so that downstream channel routing can branch on the concrete Python type rather than re-parsing a str(value) rendering.

Return type:

dict[str, TypeAliasType]

Parameters: