Module adcp.reporting.materializer.reference

Deterministic memory destination for tests/development. NEVER production support.

Functions

def reference_digest(verifier: ReportingRevisionVerifier, rows: list[object]) ‑> ReportingCanonicalDigest

Expand source code

def reference_digest(
    verifier: ReportingRevisionVerifier, rows: list[object]
) -> ReportingCanonicalDigest:
    encoded, _ = verifier.canonicalize(rows)
    contract = verifier.key.canonicalization
    return ReportingCanonicalDigest(
        hashlib.sha256(b"[" + b",".join(encoded) + b"]").hexdigest(),
        contract.canonicalization_id,
        contract.canonicalization_uri,
        contract.canonicalization_sha256,
    )

def reference_verifier(capability: ReportingWriterCapability | None = None) ‑> ReportingRevisionVerifier

Expand source code

def reference_verifier(
    capability: ReportingWriterCapability | None = None,
) -> ReportingRevisionVerifier:
    """One installed example definition/canonicalization; no network resolution."""
    root = files("adcp.reporting.materializer").joinpath("assets")
    definition = root.joinpath("reference-definition.json").read_bytes()
    schema = root.joinpath("reference-row-schema.json").read_bytes()
    contract = root.joinpath("reference-canonicalization.json").read_bytes()
    capability = capability or ReportingWriterCapability(
        "file_transfer",
        "reference-memory",
        "jsonl",
        "canonical_digest",
        "producer",
        "immutable_location",
        "sha256",
        "conditional_create",
    )
    return ReportingRevisionVerifier(
        ReportingVerificationKey(
            "reference-report-v1",
            "paid_media_delivery",
            ReportingDefinitionBinding(
                "https://contracts.example.test/reference-definition.json",
                hashlib.sha256(definition).hexdigest(),
                "1.0.0",
                "https://contracts.example.test/reference-row-schema.json",
                hashlib.sha256(schema).hexdigest(),
                monetary_metric_units=(("spend", "USD"),),
                monetary_control_total_units=(("spend", "USD"),),
            ),
            ReportingCanonicalization(
                "reference-jcs-rows-v1",
                "https://contracts.example.test/reference-canonicalization.json",
                hashlib.sha256(contract).hexdigest(),
            ),
            capability,
        ),
        definition,
        schema,
        contract,
    )

One installed example definition/canonicalization; no network resolution.

Classes

class ReferenceReportingDestinationWriter (capabilities: tuple[ReportingWriterCapability, ...])

Expand source code

@final
class ReferenceReportingDestinationWriter:
    """Conditional, deterministic external identities in shared process memory.

    Losing this object loses its artifacts. No durability or Managed capability
    follows from this type or its descriptors. The production flag is a constant
    property, with no constructor/config override.
    """

    def __init_subclass__(cls, **kwargs: Any) -> None:
        raise TypeError("the non-production reference writer cannot be promoted by subclassing")

    def __init__(self, capabilities: tuple[ReportingWriterCapability, ...]) -> None:
        if type(capabilities) is not tuple or any(
            type(c) is not ReportingWriterCapability for c in capabilities
        ):
            raise ValueError("reference capabilities require an immutable typed tuple")
        self._capabilities = capabilities
        self._artifacts: dict[str, _Artifact] = {}
        self.write_effects = 0
        self.open_count = 0
        self.close_count = 0

    @property
    def capabilities(self) -> tuple[ReportingWriterCapability, ...]:
        return self._capabilities

    @property
    def production_eligible(self) -> Literal[False]:
        return False

    def __repr__(self) -> str:
        return "<ReferenceReportingDestinationWriter non-production>"

Conditional, deterministic external identities in shared process memory.

Losing this object loses its artifacts. No durability or Managed capability follows from this type or its descriptors. The production flag is a constant property, with no constructor/config override.

Instance variables

prop capabilities : tuple[ReportingWriterCapability, ...]

Expand source code

@property
def capabilities(self) -> tuple[ReportingWriterCapability, ...]:
    return self._capabilities

prop production_eligible : Literal[False]

Expand source code

@property
def production_eligible(self) -> Literal[False]:
    return False

class ReferenceReportingResolver (writer: ReferenceReportingDestinationWriter,
registry: ReportingRevisionVerifierRegistry,
bindings: tuple[ReportingDestinationBinding, ...])

Expand source code

@dataclass(frozen=True)
class ReferenceReportingResolver:
    writer: ReferenceReportingDestinationWriter
    registry: ReportingRevisionVerifierRegistry
    bindings: tuple[ReportingDestinationBinding, ...] = field(repr=False)
    _revoked: set[ReportingDeliveryPrincipal] = field(
        default_factory=set, init=False, repr=False, compare=False
    )
    _rotations: list[int] = field(
        default_factory=lambda: [0], init=False, repr=False, compare=False
    )

    def __post_init__(self) -> None:
        if type(self.bindings) is not tuple or any(
            type(b) is not ReportingDestinationBinding for b in self.bindings
        ):
            raise ValueError("reference resolver requires frozen bindings")

    def revoke(self, principal: ReportingDeliveryPrincipal) -> None:
        self._revoked.add(principal)

    def rotate(self) -> None:
        self._rotations[0] += 1

    def resolve(
        self,
        request: ReportingDestinationRequest,
        *,
        phase: ReportingIOPhase,
        context: ReportingIOContext,
    ) -> ReportingDestinationSession:
        return _Session(self, request, phase, context)

ReferenceReportingResolver(writer: 'ReferenceReportingDestinationWriter', registry: 'ReportingRevisionVerifierRegistry', bindings: 'tuple[ReportingDestinationBinding, …]')

Instance variables

var bindings : tuple[ReportingDestinationBinding, ...]

var registry : ReportingRevisionVerifierRegistry

var writer : ReferenceReportingDestinationWriter

Methods

def resolve(self,
request: ReportingDestinationRequest,
*,
phase: ReportingIOPhase,
context: ReportingIOContext) ‑> ReportingDestinationSession

Expand source code

def resolve(
    self,
    request: ReportingDestinationRequest,
    *,
    phase: ReportingIOPhase,
    context: ReportingIOContext,
) -> ReportingDestinationSession:
    return _Session(self, request, phase, context)

def revoke(self, principal: ReportingDeliveryPrincipal) ‑> None

Expand source code

def revoke(self, principal: ReportingDeliveryPrincipal) -> None:
    self._revoked.add(principal)

def rotate(self) ‑> None

Expand source code

def rotate(self) -> None:
    self._rotations[0] += 1