"""Safe immutable projection for the six-zone AEAT Sync workspace."""
from __future__ import annotations
import hashlib
import hmac
import re
import secrets
from collections.abc import Mapping
from dataclasses import dataclass
from datetime import date
from enum import StrEnum
from typing import Annotated, Any, Final, NotRequired, Self, TypedDict, Unpack, override
from pydantic import BaseModel, Field, NonNegativeInt, StringConstraints, TypeAdapter, model_validator
from ...core.errors.hierarchy import CadrumoError, pydantic_validation_boundary
from ...core.filing_year import FilingYear
from ...core.hex import HEX_PATTERN_64
from ...core.identifier_grammar import NamespacedId
from ...core.identity.bucket import BucketId
from ...core.models import STRICT_FROZEN_CONFIG
from ...core.period import Period
from ...core.time.utc import UtcInstant
from ...domain.modelos.codes import ModeloCode
from ..operations.models import OperationDefinitionId
from ..operations.registry import OperationPublicContractSetV1
from ..operator_actions.catalogue import ActionCatalogue
from ..operator_actions.models import ActionReference
AEAT_SYNC_WORKSPACE_CONTRACT_VERSION: Final[int] = 1
_NOTIFICATION_SELECTION_KEY: Final[bytes] = secrets.token_bytes(32)
_NOTIFICATION_SELECTION_PREFIX: Final[str] = "aeat_sync.notification."
_NOTIFICATION_SELECTION_DIGEST_LENGTH: Final[int] = 64
_NOTIFICATION_SELECTION_PATTERN: Final[str] = (
rf"^{re.escape(_NOTIFICATION_SELECTION_PREFIX)}{HEX_PATTERN_64.removeprefix('^').removesuffix('$')}$"
)
type AeatSyncNotificationSelectionKey = Annotated[
str,
StringConstraints(
min_length=len(_NOTIFICATION_SELECTION_PREFIX) + _NOTIFICATION_SELECTION_DIGEST_LENGTH,
max_length=len(_NOTIFICATION_SELECTION_PREFIX) + _NOTIFICATION_SELECTION_DIGEST_LENGTH,
pattern=_NOTIFICATION_SELECTION_PATTERN,
),
]
"""Opaque, bounded identity for one projected notification row.
The admission coordinate remains on :class:`AeatSyncWorkspaceFactV1` only.
Projection derives this key from that coordinate and reconstructs the public
row, so the public snapshot retains a stable semantic focus address without
retaining the private notification identity.
"""
[docs]
class AeatSyncWorkspaceProjectionError(CadrumoError):
"""Already-loaded authorities cannot form one safe snapshot."""
[docs]
class AeatSyncWorkspaceZone(StrEnum):
"""Closed workspace areas."""
OVERVIEW = "overview"
CENSUS = "census"
FILED_DECLARATIONS = "filed_declarations"
NOTIFICATIONS = "notifications"
EVIDENCE_COMPARISON = "evidence_comparison"
RECONCILIATION = "reconciliation"
[docs]
class AeatSyncWorkspaceSource(StrEnum):
"""Canonical local and AEAT source authorities."""
LOCAL_PROFILE = "local.profile"
LOCAL_FILINGS = "local.filings"
LOCAL_NOTIFICATION_CUSTODY = "local.notification_custody"
LOCAL_RECONCILIATION = "local.reconciliation"
AEAT_CENSUS = "aeat.census"
AEAT_FILED_DECLARATIONS = "aeat.filed_declarations"
AEAT_NOTIFICATIONS = "aeat.notifications"
[docs]
class AeatSyncWorkspaceAvailability(StrEnum):
"""Availability and freshness of one source observation."""
AVAILABLE = "available"
LOCKED = "locked"
STALE = "stale"
NEVER_CAPTURED = "never_captured"
UNAVAILABLE = "unavailable"
[docs]
class AeatSyncSourceState(StrEnum):
"""Safe observed state of one comparison side."""
NOT_OBSERVED = "not_observed"
ABSENT = "absent"
PRESENT = "present"
ACCEPTED = "accepted"
REJECTED = "rejected"
CONFLICT = "conflict"
[docs]
class AeatSyncDiscrepancyKind(StrEnum):
"""Closed comparison outcomes."""
NONE = "none"
LOCAL_ONLY = "local_only"
AEAT_ONLY = "aeat_only"
STATE_MISMATCH = "state_mismatch"
CONTRADICTORY_SOURCE = "contradictory_source"
UNOBSERVED = "unobserved"
[docs]
class AeatSyncOverviewArea(StrEnum):
"""Closed overview areas."""
CENSUS = "census"
FILED_DECLARATIONS = "filed_declarations"
NOTIFICATIONS = "notifications"
EVIDENCE_COMPARISON = "evidence_comparison"
RECONCILIATION = "reconciliation"
[docs]
class AeatSyncCensusCategory(StrEnum):
"""The censo field families a census row is grouped under."""
ADDRESS = "address"
ACTIVITY = "activity"
OBLIGATION = "obligation"
CONTACT = "contact"
OTHER = "other"
[docs]
class AeatSyncCensusStatus(StrEnum):
"""The census comparison verdicts, plus the state of having none yet.
The first four are VERDICTS: each claims someone compared the local censo
field against the AEAT one and says how they came out. ``NOT_COMPARED``
claims the opposite -- that nobody has looked at the AEAT side at all --
and exists because before a pull there is no honest verdict to give. Made
to reuse one of the four, a pre-pull row would have to assert a comparison
that never happened; ``UNSET`` in particular says the field HAS no value,
which is a statement about the value rather than about whether anyone
checked.
"""
ADOPTED = "adopted"
CONFLICT = "conflict"
UNCHANGED = "unchanged"
UNSET = "unset"
NOT_COMPARED = "not_compared"
COMPARED_CENSUS_STATUSES: Final = frozenset(
{
AeatSyncCensusStatus.ADOPTED,
AeatSyncCensusStatus.CONFLICT,
AeatSyncCensusStatus.UNCHANGED,
AeatSyncCensusStatus.UNSET,
}
)
"""The census statuses that assert a comparison against the AEAT actually ran."""
[docs]
class AeatSyncLocalFilingState(StrEnum):
"""Local filing states."""
NOT_OBSERVED = "not_observed"
DRAFT = "draft"
READY = "ready"
FILED = "filed"
[docs]
class AeatSyncAeatObservationState(StrEnum):
"""AEAT declaration observation states."""
NOT_OBSERVED = "not_observed"
SUBMITTED = "submitted"
ACCEPTED = "accepted"
REJECTED = "rejected"
[docs]
class AeatSyncJustificanteState(StrEnum):
"""Justificante availability states."""
NOT_OBSERVED = "not_observed"
UNAVAILABLE = "unavailable"
AVAILABLE = "available"
VERIFIED = "verified"
[docs]
class AeatSyncNotificationCategory(StrEnum):
"""The notification families the AEAT distinguishes."""
FORMAL = "formal"
COMMUNICATION = "communication"
PENDING = "pending"
OTHER = "other"
[docs]
class AeatSyncNotificationReadState(StrEnum):
"""Notification read states."""
UNKNOWN = "unknown"
UNREAD = "unread"
READ = "read"
[docs]
class AeatSyncDocumentCustodyState(StrEnum):
"""Encrypted document custody states."""
NOT_CAPTURED = "not_captured"
HELD = "held"
REFUSED = "refused"
UNAVAILABLE = "unavailable"
[docs]
class AeatSyncReconciliationState(StrEnum):
"""Closed reconciliation outcomes."""
UNRESOLVED = "unresolved"
KEEP_LOCAL = "keep_local"
ACCEPT_AEAT = "accept_aeat"
DISMISSED = "dismissed"
NO_ACTION = "no_action"
[docs]
class AeatSyncWorkspaceSourceObservationV1(BaseModel):
"""One authority's independent availability/freshness/count axis."""
model_config = STRICT_FROZEN_CONFIG
source: AeatSyncWorkspaceSource
availability: AeatSyncWorkspaceAvailability
observed_at: UtcInstant | None = None
refusal: NamespacedId | None = None
item_count: NonNegativeInt | None = None
@model_validator(mode="after")
@pydantic_validation_boundary
def _coherent(self) -> Self:
observable = _observable(self.availability)
if observable != (self.observed_at is not None and self.item_count is not None):
raise ValueError("source observability contradicts its time/count")
if self.availability is AeatSyncWorkspaceAvailability.AVAILABLE:
if self.refusal is not None:
raise ValueError("available source cannot carry refusal")
elif self.refusal is None:
raise ValueError("non-available source requires refusal")
return self
[docs]
class AeatSyncWorkspaceZoneObservationV1(BaseModel):
"""Independent source observations for one zone."""
model_config = STRICT_FROZEN_CONFIG
zone: AeatSyncWorkspaceZone
sources: tuple[AeatSyncWorkspaceSourceObservationV1, ...]
@model_validator(mode="after")
@pydantic_validation_boundary
def _unique(self) -> Self:
ids = tuple(item.source for item in self.sources)
if len(ids) != len(set(ids)):
raise ValueError("zone source observations must be unique")
return self
[docs]
class AeatSyncWorkspaceZoneStateV1(BaseModel):
"""Projected zone state retaining its independent sources."""
model_config = STRICT_FROZEN_CONFIG
zone: AeatSyncWorkspaceZone
availability: AeatSyncWorkspaceAvailability
sources: tuple[AeatSyncWorkspaceSourceObservationV1, ...]
item_count: NonNegativeInt | None
[docs]
class AeatSyncWorkspaceActionRowV1(BaseModel):
"""Public immutable capability provenance shared by every workspace row."""
model_config = STRICT_FROZEN_CONFIG
supported_actions: tuple[ActionReference, ...] = ()
supported_operations: tuple[OperationDefinitionId, ...] = ()
[docs]
class AeatSyncWorkspaceOverviewRowV1(AeatSyncWorkspaceActionRowV1):
"""One area's local-versus-AEAT standing, with the actions it admits."""
area: AeatSyncOverviewArea
local_state: AeatSyncSourceState
aeat_state: AeatSyncSourceState
local_observed_at: UtcInstant | None = None
aeat_observed_at: UtcInstant | None = None
discrepancy_kind: AeatSyncDiscrepancyKind
@model_validator(mode="after")
@pydantic_validation_boundary
def _coherent(self) -> Self:
_dual(self.local_state, self.local_observed_at, self.aeat_state, self.aeat_observed_at)
_discrepancy(self.local_state, self.aeat_state, self.discrepancy_kind)
return self
[docs]
class AeatSyncWorkspaceCensusRowV1(AeatSyncWorkspaceActionRowV1):
"""One census field's local-versus-AEAT status.
Carries no value yet, and the docstring no longer claims that as a safety
property: it is a GAP. Nothing produces these rows outside fixtures, and
their AEAT side stays never-captured until a pull happens, so there is no
captured value to carry. When a producer exists the values belong here, on
the same reasoning as every other authenticated surface.
"""
path: str = Field(min_length=1, max_length=256)
category: AeatSyncCensusCategory
status: AeatSyncCensusStatus
local_value: str | None = Field(default=None, max_length=256)
"""What the local profile holds for this field, or nothing when unobserved."""
aeat_value: str | None = Field(default=None, max_length=256)
"""What the AEAT censo holds, or nothing when no pull has observed it.
`None` on either side is UNOBSERVED, never an empty field. A censo entry the
taxpayer has genuinely left blank is the empty string, and collapsing the
two would tell an operator AEAT holds nothing where in truth nobody has
looked -- which is the difference between "correct" and "unchecked" on a
comparison whose entire purpose is to show what differs.
"""
@model_validator(mode="after")
@pydantic_validation_boundary
def _the_status_and_the_values_must_agree(self) -> Self:
"""A verdict must carry what it judged; a non-verdict must not pretend to.
Both directions are defects, and neither is caught by the other.
Reporting a CONFLICT while withholding a side asks the operator to
accept a difference they cannot see -- the same defect the invoice and
entry suggestions carried before they showed the amounts they compared.
UNSET and UNCHANGED stay deliberately unconstrained, as they were:
both are meaningful before either side has been read, and requiring
values there would force a producer to invent them.
NOT_COMPARED asserts that nobody looked at the AEAT. A row carrying an
AEAT value under that status is self-contradicting: someone plainly
did look, and the operator would be told the field is unchecked while
the evidence sits in the same row.
"""
if self.status is AeatSyncCensusStatus.CONFLICT and (self.local_value is None or self.aeat_value is None):
raise ValueError("a census conflict must carry both the local and the AEAT value")
if self.status is AeatSyncCensusStatus.NOT_COMPARED and self.aeat_value is not None:
raise ValueError("a census row that was never compared cannot carry an AEAT value")
return self
[docs]
class AeatSyncWorkspaceFiledDeclarationRowV1(AeatSyncWorkspaceActionRowV1):
"""One declaration this profile filed, against what the AEAT shows for it."""
modelo: ModeloCode
filing_year: FilingYear
period: Period
local_filing_state: AeatSyncLocalFilingState
local_filed_at: UtcInstant | None = None
aeat_observation_state: AeatSyncAeatObservationState
aeat_observed_at: UtcInstant | None = None
justificante_state: AeatSyncJustificanteState
justificante_observed_at: UtcInstant | None = None
@property
def aeat_submission_state(self) -> AeatSyncAeatObservationState:
"""Return the AEAT declaration state."""
return self.aeat_observation_state
@model_validator(mode="after")
@pydantic_validation_boundary
def _coherent(self) -> Self:
if self.period.filing_year != self.filing_year:
raise ValueError("period and filing year disagree")
if (self.local_filing_state is AeatSyncLocalFilingState.FILED) != (self.local_filed_at is not None):
raise ValueError("only filed local state carries filing time")
_optional_time(self.aeat_observation_state, self.aeat_observed_at, AeatSyncAeatObservationState.NOT_OBSERVED)
confident = self.justificante_state in {AeatSyncJustificanteState.AVAILABLE, AeatSyncJustificanteState.VERIFIED}
if confident != (self.justificante_observed_at is not None):
raise ValueError("justificante state contradicts time")
if confident and self.aeat_observation_state is AeatSyncAeatObservationState.NOT_OBSERVED:
raise ValueError("justificante cannot be confident without AEAT observation")
return self
[docs]
class AeatSyncWorkspaceNotificationRowV1(BaseModel):
"""One notification's metadata, and the coordinate that opens its document.
``selection_key`` is populated only by :func:`project_aeat_sync_workspace`.
An admission row may leave it absent; the projector always replaces it
from the fact's private coordinate before exposing the row publicly.
"""
model_config = STRICT_FROZEN_CONFIG
issued_on: date
read_on: date | None = None
read_state: AeatSyncNotificationReadState
category: AeatSyncNotificationCategory
document_custody_state: AeatSyncDocumentCustodyState
document_custody_observed_at: UtcInstant | None = None
selection_key: AeatSyncNotificationSelectionKey | None = None
@property
def issue_date(self) -> date:
"""Return the notification issue date."""
return self.issued_on
@model_validator(mode="after")
@pydantic_validation_boundary
def _coherent(self) -> Self:
if (self.read_state is AeatSyncNotificationReadState.READ) != (self.read_on is not None):
raise ValueError("only read state carries read date")
if self.read_on is not None and self.read_on < self.issued_on:
raise ValueError("read date precedes issue date")
held = self.document_custody_state is AeatSyncDocumentCustodyState.HELD
if held != (self.document_custody_observed_at is not None):
raise ValueError("custody state contradicts time")
if held and self.read_state is not AeatSyncNotificationReadState.READ:
raise ValueError("held document requires read state")
return self
class _DualRow(AeatSyncWorkspaceActionRowV1):
modelo: ModeloCode
filing_year: FilingYear
period: Period
local_state: AeatSyncSourceState
aeat_state: AeatSyncSourceState
local_observed_at: UtcInstant | None = None
aeat_observed_at: UtcInstant | None = None
discrepancy_kind: AeatSyncDiscrepancyKind
local_value: str | None = Field(default=None, max_length=256)
"""What the local record holds for the compared figure, or nothing.
`None` is UNOBSERVED, not empty and not zero. A row can legitimately have
no value on a side that was never read, and the state axis above already
says which side that is.
"""
aeat_value: str | None = Field(default=None, max_length=256)
"""What AEAT holds for the same figure, or nothing when unobserved."""
@model_validator(mode="after")
@pydantic_validation_boundary
def _coherent(self) -> Self:
if self.period.filing_year != self.filing_year:
raise ValueError("period and filing year disagree")
_dual(self.local_state, self.local_observed_at, self.aeat_state, self.aeat_observed_at)
_discrepancy(self.local_state, self.aeat_state, self.discrepancy_kind)
_compared_values(self.discrepancy_kind, self.local_value, self.aeat_value)
return self
[docs]
class AeatSyncWorkspaceEvidenceComparisonRowV1(_DualRow):
"""One declaration figure as the local record holds it and as the AEAT shows it."""
pass
[docs]
class AeatSyncWorkspaceReconciliationRowV1(_DualRow):
"""One reconciliation item, its two sides and how it was resolved."""
reconciliation_state: AeatSyncReconciliationState
@model_validator(mode="after")
@pydantic_validation_boundary
def _reconciled(self) -> Self:
no_action = self.reconciliation_state is AeatSyncReconciliationState.NO_ACTION
if (self.discrepancy_kind is AeatSyncDiscrepancyKind.NONE) != no_action:
raise ValueError("reconciliation state contradicts discrepancy")
if no_action and self.supported_actions:
raise ValueError("NO_ACTION cannot carry actions")
return self
@override
def model_copy(self, *, update: Mapping[str, Any] | None = None, deep: bool = False) -> Self:
"""Keep state/action closure intact across mutation-test copies."""
if update is None:
return super().model_copy(deep=deep)
return type(self).model_validate({**self.model_dump(), **update})
[docs]
@dataclass(frozen=True, slots=True)
class AeatSyncWorkspaceFactV1[RowT: BaseModel]:
"""Admission-only scope/private identity projected away from ``row``."""
bucket_id: BucketId
subject_key: str
row: RowT
private_identity: str | None = None
def __post_init__(self) -> None:
"""Reject absent or blank provenance coordinates."""
TypeAdapter(BucketId).validate_python(self.bucket_id)
if not self.subject_key.strip():
raise ValueError("subject key cannot be blank")
if self.private_identity is not None and not self.private_identity.strip():
raise ValueError("private identity cannot be blank")
[docs]
class AeatSyncWorkspaceProjectionV1(BaseModel):
"""Public result, physically free of protected admission fields."""
model_config = STRICT_FROZEN_CONFIG
contract_version: int = AEAT_SYNC_WORKSPACE_CONTRACT_VERSION
zones: tuple[AeatSyncWorkspaceZoneStateV1, ...]
overview: tuple[AeatSyncWorkspaceOverviewRowV1, ...] = ()
census: tuple[AeatSyncWorkspaceCensusRowV1, ...] = ()
filed_declarations: tuple[AeatSyncWorkspaceFiledDeclarationRowV1, ...] = ()
notifications: tuple[AeatSyncWorkspaceNotificationRowV1, ...] = ()
evidence_comparison: tuple[AeatSyncWorkspaceEvidenceComparisonRowV1, ...] = ()
reconciliation: tuple[AeatSyncWorkspaceReconciliationRowV1, ...] = ()
@model_validator(mode="after")
@pydantic_validation_boundary
def _six_zones(self) -> Self:
if tuple(item.zone for item in self.zones) != tuple(AeatSyncWorkspaceZone):
raise ValueError("zones must cover the closed catalogue")
selection_keys = tuple(row.selection_key for row in self.notifications)
if any(key is None for key in selection_keys):
raise ValueError("projected notifications require selection keys")
if len(selection_keys) != len(set(selection_keys)):
raise ValueError("projected notification selection keys must be unique")
return self
[docs]
def aeat_sync_workspace_sources(zone: AeatSyncWorkspaceZone) -> tuple[AeatSyncWorkspaceSource, ...]:
"""Return the canonical independent sources required by one zone."""
from . import _workspace_projection
return _workspace_projection.aeat_sync_workspace_sources(zone)
_NOTIFICATION_SELECTION_NAMESPACE: Final[str] = "aeat_sync.notification.selection.v1"
[docs]
def notification_selection_key(private_identity: str) -> AeatSyncNotificationSelectionKey:
"""Derive a process-stable public focus key without retaining private data."""
canonical = "\x1f".join((_NOTIFICATION_SELECTION_NAMESPACE, private_identity)).encode("utf-8")
digest = hmac.digest(_NOTIFICATION_SELECTION_KEY, canonical, hashlib.sha256).hex()
return f"{_NOTIFICATION_SELECTION_PREFIX}{digest}"
[docs]
def public_notification_row(
row: AeatSyncWorkspaceNotificationRowV1,
selection_key: AeatSyncNotificationSelectionKey,
) -> AeatSyncWorkspaceNotificationRowV1:
"""Rebuild one notification row and inject only its safe selection key."""
values = row.model_dump(include=set(AeatSyncWorkspaceNotificationRowV1.model_fields))
values["selection_key"] = selection_key
return AeatSyncWorkspaceNotificationRowV1.model_validate(values)
[docs]
def source_observation_is_observable(value: AeatSyncWorkspaceAvailability) -> bool:
"""Return whether a source has an observation that can support a claim."""
return value in {AeatSyncWorkspaceAvailability.AVAILABLE, AeatSyncWorkspaceAvailability.STALE}
_observable = source_observation_is_observable
def _optional_time(state: object, when: UtcInstant | None, missing: object) -> None:
if (state is missing) == (when is not None):
raise ValueError("state contradicts observation time")
def _dual(
local: AeatSyncSourceState, local_at: UtcInstant | None, aeat: AeatSyncSourceState, aeat_at: UtcInstant | None
) -> None:
_optional_time(local, local_at, AeatSyncSourceState.NOT_OBSERVED)
_optional_time(aeat, aeat_at, AeatSyncSourceState.NOT_OBSERVED)
def _compared_values(
kind: AeatSyncDiscrepancyKind,
local_value: str | None,
aeat_value: str | None,
) -> None:
"""Refuse a difference claim that withholds one of the values it compares.
STATE_MISMATCH and CONTRADICTORY_SOURCE both assert that two observed sides
disagree. Making that claim while hiding a side leaves the operator to
accept a difference they cannot inspect -- the same defect the invoice/entry
suggestions carried before they showed their amounts.
The one-sided kinds are deliberately exempt. LOCAL_ONLY and AEAT_ONLY say
the other side is ABSENT, so there is no second value to carry; UNOBSERVED
says nobody looked; NONE says the sides agree and needs no evidence of
difference.
"""
if kind not in {AeatSyncDiscrepancyKind.STATE_MISMATCH, AeatSyncDiscrepancyKind.CONTRADICTORY_SOURCE}:
return
if local_value is None or aeat_value is None:
raise ValueError(f"a {kind.value} row must carry both the local and the AEAT value")
def _discrepancy(local: AeatSyncSourceState, aeat: AeatSyncSourceState, kind: AeatSyncDiscrepancyKind) -> None:
if AeatSyncSourceState.NOT_OBSERVED in {local, aeat}:
expected = AeatSyncDiscrepancyKind.UNOBSERVED
elif local == aeat:
expected = AeatSyncDiscrepancyKind.NONE
elif local is AeatSyncSourceState.ABSENT:
expected = AeatSyncDiscrepancyKind.AEAT_ONLY
elif aeat is AeatSyncSourceState.ABSENT:
expected = AeatSyncDiscrepancyKind.LOCAL_ONLY
elif AeatSyncSourceState.CONFLICT in {local, aeat}:
expected = AeatSyncDiscrepancyKind.CONTRADICTORY_SOURCE
else:
expected = AeatSyncDiscrepancyKind.STATE_MISMATCH
if kind is not expected:
raise ValueError("discrepancy contradicts source states")
class _AeatSyncWorkspaceProjectionArguments(TypedDict):
bucket_id: BucketId
subject_key: str
zone_observations: tuple[AeatSyncWorkspaceZoneObservationV1, ...]
action_catalogue: ActionCatalogue
operation_contracts: OperationPublicContractSetV1
overview: NotRequired[tuple[AeatSyncWorkspaceFactV1[AeatSyncWorkspaceOverviewRowV1], ...]]
census: NotRequired[tuple[AeatSyncWorkspaceFactV1[AeatSyncWorkspaceCensusRowV1], ...]]
filed_declarations: NotRequired[tuple[AeatSyncWorkspaceFactV1[AeatSyncWorkspaceFiledDeclarationRowV1], ...]]
notifications: NotRequired[tuple[AeatSyncWorkspaceFactV1[AeatSyncWorkspaceNotificationRowV1], ...]]
evidence_comparison: NotRequired[tuple[AeatSyncWorkspaceFactV1[AeatSyncWorkspaceEvidenceComparisonRowV1], ...]]
reconciliation: NotRequired[tuple[AeatSyncWorkspaceFactV1[AeatSyncWorkspaceReconciliationRowV1], ...]]
[docs]
def project_aeat_sync_workspace(
**arguments: Unpack[_AeatSyncWorkspaceProjectionArguments],
) -> AeatSyncWorkspaceProjectionV1:
"""Project already-loaded, scoped facts through the single projection owner."""
from ._workspace_projection import project_aeat_sync_workspace as project
return project(**arguments)