Module adcp.reporting.projection.wire

Tier-correct exact views over one captured, caller-private financial history.

Functions

def exact_revision_evidence(core: ReportingStatusSnapshot,
revision: ReportingRevisionRecord,
owner: ReportingObligationRecord | None,
records: tuple[ReportingDeliveryRecord, ...],
caller: ReportingStatusCaller) ‑> dict[str, typing.Any]

Expand source code

def exact_revision_evidence(
    core: ReportingStatusSnapshot,
    revision: ReportingRevisionRecord,
    owner: ReportingObligationRecord | None,
    records: tuple[ReportingDeliveryRecord, ...],
    caller: ReportingStatusCaller,
) -> dict[str, Any]:
    if owner is None or owner.account_id != caller.account_id:
        raise LedgerConflictError(
            "LOOKUP_UNAVAILABLE", "no such revision is available to this caller"
        )
    history = tuple(
        r for r in core.revisions if r.reporting_obligation_id == owner.reporting_obligation_id
    )
    project_reconciliation(
        owner, history, core.adjustments, records, consumer_id=caller.consumer_id, as_of=core.as_of
    )
    binding = next(
        (
            r
            for r in records
            if isinstance(r, ReportingDestinationBinding)
            and r.generation_key == owner.generation_key
        ),
        None,
    )
    if binding is None:
        return {}
    attempts = {
        r.reporting_materialization_id: r
        for r in records
        if isinstance(r, ReportingMaterializationAttempt)
        and r.reporting_revision_id == revision.reporting_revision_id
    }
    artifacts = [
        r
        for r in records
        if isinstance(r, ReportingMaterializationRecord)
        and r.reporting_revision_id == revision.reporting_revision_id
    ]
    adjustments = tuple(
        a
        for a in core.adjustments
        if a.adjusts_reporting_revision_id == revision.reporting_revision_id
    )
    adjustment_ids = {a.reporting_adjustment_id for a in adjustments}
    receipts = [
        r
        for r in records
        if isinstance(r, ReportingRevisionReceiptRecord)
        and r.reporting_revision_id == revision.reporting_revision_id
    ]
    adjustment_receipts = [
        r
        for r in records
        if isinstance(r, ReportingAdjustmentReceiptRecord)
        and r.reporting_adjustment_id in adjustment_ids
    ]
    return {
        "revision": revision_to_wire(revision, obligation=owner),
        "adjustments": [adjustment_to_wire(a) for a in adjustments],
        "materializations": [
            materialization_to_wire(
                ReportingMaterializationView(
                    attempts[r.reporting_materialization_id],
                    binding,
                    r,
                    tuple(
                        c
                        for c in records
                        if isinstance(c, ReportingMaterializationCheck)
                        and c.reporting_materialization_id == r.reporting_materialization_id
                    ),
                ),
                obligation=owner,
            )
            for r in artifacts
        ],
        "receipts": [receipt_to_wire(r) for r in receipts],
        "adjustment_receipts": [receipt_to_wire(r) for r in adjustment_receipts],
        "pagination": {
            "total_count": 1
            + len(adjustments)
            + len(artifacts)
            + len(receipts)
            + len(adjustment_receipts),
            "has_more": False,
        },
    }

def render_tier_status(store: Any,
request: dict[str, Any],
value: ReportingProjectionInput,
caller: ReportingDeliveryPrincipal,
policy: dict[str, Any],
consumer_status_enabled: bool) ‑> dict[str, typing.Any]

Expand source code

def render_tier_status(
    store: Any,
    request: dict[str, Any],
    value: ReportingProjectionInput,
    caller: ReportingDeliveryPrincipal,
    policy: dict[str, Any],
    consumer_status_enabled: bool,
) -> dict[str, Any]:
    if (
        getattr(store, "_projection_read_policy", None) != policy
        or policy["consumer_status_enabled"] != consumer_status_enabled
    ):
        raise LedgerConflictError(
            "STATUS_PROJECTION_UNAVAILABLE", "reporting projection is unavailable"
        )
    return ReportingStatusHandler(
        store,
        consumer_status_enabled=consumer_status_enabled,
        escalation=getattr(store, "_projection_read_escalation", None),
    ).render_snapshot(
        request,
        caller=ReportingStatusCaller(caller.account_id, caller.consumer_id),
        snapshot=settled_replay(private_snapshot(value.core, caller)),
        reconciliation=value.reconciliation,
        revision_ownership=policy["ownership_enabled"],
    )

Classes

class ReportingTierStatusStore (*args, **kwargs)

Expand source code

@runtime_checkable
class ReportingTierStatusStore(Protocol):
    async def read_tier_status(
        self,
        request: dict[str, Any],
        *,
        caller: ReportingDeliveryPrincipal,
        consumer_status_enabled: bool = False,
    ) -> dict[str, Any]: ...

Base class for protocol classes.

Protocol classes are defined as::

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).

For example::

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as::

class GenProto(Protocol[T]):
    def meth(self) -> T:
        ...

Ancestors

• typing.Protocol
• typing.Generic

Methods

async def read_tier_status(self,
request: dict[str, Any],
*,
caller: ReportingDeliveryPrincipal,
consumer_status_enabled: bool = False) ‑> dict[str, typing.Any]

Expand source code

async def read_tier_status(
    self,
    request: dict[str, Any],
    *,
    caller: ReportingDeliveryPrincipal,
    consumer_status_enabled: bool = False,
) -> dict[str, Any]: ...