Module adcp.reporting.production.contracts

Provider declarations and trusted, immutable source generation bindings.

Classes

class ReportingProductionDestinationBinding (binding: ReportingDestinationBinding,
method: ReportingProductionMethod,
configuration: Mapping[str, Any])

Expand source code

@dataclass(frozen=True)
class ReportingProductionDestinationBinding:
    """The provider's authorized, immutable destination contract for a caller.

    ``configuration`` is the complete secret-free public delivery method,
    including the selected destination. The provider resolves it from its
    trusted binding, not from a buyer's claim. Credentials remain in the
    independently authorized write/readback sessions. The SDK freezes these
    bytes at admission and compares them again before materialization.
    """

    binding: ReportingDestinationBinding = field(repr=False)
    method: ReportingProductionMethod
    configuration: Mapping[str, Any] = field(repr=False, compare=False)
    _wire: bytes = field(init=False, repr=False)

    def __post_init__(self) -> None:
        from adcp.reporting.evidence import consumer_reference, resource_location
        from adcp.validation.schema_loader import get_named_validator

        if (
            type(self.binding) is not ReportingDestinationBinding
            or type(self.method) is not ReportingProductionMethod
        ):
            raise ValueError("destination requires the exact provider binding and method")
        raw = json.loads(canonical_json_utf8_v1(dict(self.configuration)))
        validator = get_named_validator("core/reporting-delivery-method.json")
        offered = self.method.wire()
        if (
            validator is None
            or next(validator.iter_errors(raw), None) is not None
            or any(raw.get(k) != offered.get(k) for k in ("pattern", "transport", "orchestration"))
            or (raw["pattern"] == "file_transfer" and raw["format"] != offered.get("format"))
            or raw["destination"]["mode"] not in offered["destination_modes"]
        ):
            raise ValueError("destination must match the provider's complete method")
        destination = raw["destination"]
        if destination["mode"] == "existing":
            if destination["destination_ref"] != self.binding.destination_ref:
                raise ValueError("destination must match the immutable binding")
        elif destination.get("provider") != offered.get("provider") or destination.get(
            "access_mode"
        ) != offered.get("access_mode"):
            raise ValueError("destination must match the provider's complete method")
        if "location" in destination:
            resource_location(destination["location"])
        if "recipient" in destination:
            consumer_reference(destination["recipient"]["identity"])
        object.__setattr__(self, "_wire", canonical_json_utf8_v1(raw))

    def wire(self) -> dict[str, Any]:
        return dict(json.loads(self._wire))

The provider's authorized, immutable destination contract for a caller.

configuration is the complete secret-free public delivery method, including the selected destination. The provider resolves it from its trusted binding, not from a buyer's claim. Credentials remain in the independently authorized write/readback sessions. The SDK freezes these bytes at admission and compares them again before materialization.

Instance variables

var binding : ReportingDestinationBinding

var configuration : Mapping[str, typing.Any]

var method : ReportingProductionMethod

Methods

def wire(self) ‑> dict[str, typing.Any]

Expand source code

def wire(self) -> dict[str, Any]:
    return dict(json.loads(self._wire))

class ReportingProductionMethod (capability: ReportingWriterCapability, method: Mapping[str, Any])

Expand source code

@dataclass(frozen=True)
class ReportingProductionMethod:
    """The actual provider's complete public method for one writer capability.

    This includes the provider, destination modes, access/producer identity and
    reader requirements. A method advertised by an offering must equal this
    declaration. The copied bytes cannot change through an adopter's mapping.
    Runtime writer, resolver, verifier and authorization checks remain required.
    """

    capability: ReportingWriterCapability
    method: Mapping[str, Any] = field(repr=False, compare=False)
    _wire: bytes = field(init=False, repr=False)

    def __post_init__(self) -> None:
        from adcp.validation.schema_loader import get_named_validator

        raw = json.loads(canonical_json_utf8_v1(dict(self.method)))
        validator = get_named_validator("core/reporting-delivery-offering.json")
        if (
            validator is None
            or next(
                validator.evolve(schema=validator.schema["properties"]["method"]).iter_errors(raw),
                None,
            )
            is not None
            or raw.get("orchestration") != "producer_managed"
            or (raw.get("pattern"), raw.get("transport"), raw.get("format"))
            != (self.capability.method, self.capability.transport, self.capability.format)
        ):
            raise ValueError("production method must match the provider's exact writer capability")
        object.__setattr__(self, "_wire", canonical_json_utf8_v1(raw))

    def wire(self) -> dict[str, Any]:
        return dict(json.loads(self._wire))

The actual provider's complete public method for one writer capability.

This includes the provider, destination modes, access/producer identity and reader requirements. A method advertised by an offering must equal this declaration. The copied bytes cannot change through an adopter's mapping. Runtime writer, resolver, verifier and authorization checks remain required.

Instance variables

var capability : ReportingWriterCapability

var method : Mapping[str, typing.Any]

Methods

def wire(self) ‑> dict[str, typing.Any]

Expand source code

def wire(self) -> dict[str, Any]:
    return dict(json.loads(self._wire))

class ReportingProductionSource (*args, **kwargs)

Expand source code

@runtime_checkable
class ReportingProductionSource(ReportingSourceExecutor, Protocol):
    """A source with an authenticated generation mapping available before I/O.

    Discovery may precede any account binding. Admission and every source turn
    require the applicable binding, obtained from the source's trusted account
    configuration. The SDK calls ``configuration_binding`` immediately before
    dispatch and again under the account lock before sealing or publishing the
    result, including replay after a restart. Returning ``None`` discards that
    result without a success checkpoint and stops this account's remaining
    work for the turn. Restoring the binding permits work on the next turn.

    An in-flight fetch is allowed to finish: revocation takes effect at the next
    dispatch or publish. Adopters own any caching or latency inside their
    callback; the SDK does not retain an authorization grant.
    """

    def configuration_binding(
        self, configuration: ReportingConfiguration
    ) -> ReportingProductionSourceBinding | None:
        """Return the current trusted binding, or ``None`` to deny source work.

        Adopters own callback caching and latency. In-flight fetches finish;
        revocation takes effect at the next dispatch or publish.
        """
        raise NotImplementedError

A source with an authenticated generation mapping available before I/O.

Discovery may precede any account binding. Admission and every source turn require the applicable binding, obtained from the source's trusted account configuration. The SDK calls configuration_binding immediately before dispatch and again under the account lock before sealing or publishing the result, including replay after a restart. Returning None discards that result without a success checkpoint and stops this account's remaining work for the turn. Restoring the binding permits work on the next turn.

An in-flight fetch is allowed to finish: revocation takes effect at the next dispatch or publish. Adopters own any caching or latency inside their callback; the SDK does not retain an authorization grant.

Ancestors

• ReportingSourceExecutor
• typing.Protocol
• typing.Generic

Methods

def configuration_binding(self, configuration: ReportingConfiguration) ‑> ReportingProductionSourceBinding | None

Expand source code

def configuration_binding(
    self, configuration: ReportingConfiguration
) -> ReportingProductionSourceBinding | None:
    """Return the current trusted binding, or ``None`` to deny source work.

    Adopters own callback caching and latency. In-flight fetches finish;
    revocation takes effect at the next dispatch or publish.
    """
    raise NotImplementedError

Return the current trusted binding, or None to deny source work.

Adopters own callback caching and latency. In-flight fetches finish; revocation takes effect at the next dispatch or publish.

Inherited members

• ReportingSourceExecutor:
  • capabilities

class ReportingProductionSourceBinding (generation_key: ReportingConfigurationGenerationKey,
capabilities_sha256: str,
media_buy_products: tuple[tuple[str, str], ...],
*,
configuration_sha256: str,
service_context: ReportingProductionSourceContext | None = None)

Expand source code

@dataclass(frozen=True)
class ReportingProductionSourceBinding:
    """Trusted account-to-source mapping, fixed for a configuration generation.

    ``media_buy_products`` comes from the authenticated source/account mapping,
    never from buyer JSON or a report-definition identifier. The SDK persists
    it with admission, checks the effective capability digest on every source
    turn, and refuses a changed mapping. Reauthorization can withdraw a binding;
    it cannot silently change historical scope. No credentials belong here.
    """

    generation_key: ReportingConfigurationGenerationKey
    capabilities_sha256: str
    media_buy_products: tuple[tuple[str, str], ...]
    configuration_sha256: str = field(kw_only=True)
    service_context: ReportingProductionSourceContext | None = field(
        default=None, kw_only=True, repr=False
    )

    def __post_init__(self) -> None:
        from adcp.reporting.evidence import reporting_identifier, sha256_value

        if type(self.generation_key) is not ReportingConfigurationGenerationKey:
            raise ValueError("source binding requires an exact configuration generation")
        sha256_value(self.capabilities_sha256)
        sha256_value(self.configuration_sha256)
        if (
            self.service_context is not None
            and type(self.service_context) is not ReportingProductionSourceContext
        ):
            raise ValueError("source binding requires an exact service source context")
        pairs = tuple(tuple(pair) for pair in self.media_buy_products)
        if any(len(pair) != 2 for pair in pairs):
            raise ValueError("source binding requires media-buy/product pairs")
        for media_buy_id, product_id in pairs:
            reporting_identifier(media_buy_id, maximum=255)
            reporting_identifier(product_id, maximum=255)
        if len({pair[0] for pair in pairs}) != len(pairs):
            raise ValueError("source binding media buys must be unique")
        object.__setattr__(self, "media_buy_products", tuple(sorted(pairs)))

    @classmethod
    def for_configuration(
        cls,
        configuration: ReportingConfiguration,
        *,
        capabilities_sha256: str,
        media_buy_products: tuple[tuple[str, str], ...],
    ) -> ReportingProductionSourceBinding:
        """Freeze the exact generation semantics and explicitly resolved products.

        Lifecycle changes retain the same semantic generation, matching the
        ledger's immutable configuration contract. Captured projection inputs
        independently retain each historical activation/deactivation boundary.
        """
        return cls(
            configuration.generation_key,
            capabilities_sha256,
            media_buy_products,
            configuration_sha256=hashlib.sha256(
                canonical_json_utf8_v1(_config_payload(configuration))
            ).hexdigest(),
        )

    def document(self) -> dict[str, Any]:
        key = self.generation_key
        document = {
            "account_id": key.account_id,
            "consumer_id": key.consumer_id,
            "delivery_config_id": key.delivery_config_id,
            "delivery_config_version": key.delivery_config_version,
            "capabilities_sha256": self.capabilities_sha256,
            "configuration_sha256": self.configuration_sha256,
            "media_buy_products": [list(pair) for pair in self.media_buy_products],
        }
        if self.service_context is not None:
            document["service_context"] = self.service_context.document()
            document["service_context_sha256"] = hashlib.sha256(
                self.service_context._wire
            ).hexdigest()
        return document

    def check(
        self,
        configuration: ReportingConfiguration,
        capabilities: ReportingSourceCapabilitiesV1,
        offering_id: str,
    ) -> None:
        offering = capabilities.offering(offering_id)
        if (
            self.generation_key != configuration.generation_key
            or self.configuration_sha256
            != hashlib.sha256(canonical_json_utf8_v1(_config_payload(configuration))).hexdigest()
            or capabilities.scope != "effective_account"
            or self.capabilities_sha256 != capabilities.capabilities_sha256
            or {pair[0] for pair in self.media_buy_products} != set(configuration.media_buy_ids)
            or (self.media_buy_products and "media_buy" not in offering.constituent_kinds)
            or any(pair[1] not in offering.product_ids for pair in self.media_buy_products)
        ):
            raise failure("BINDING_MISMATCH")

    def constituents(self) -> tuple[ReportingConstituent, ...]:
        return tuple(
            MediaBuyConstituentV1(
                constituent_id=media_buy_id, media_buy_id=media_buy_id, product_id=product_id
            )
            for media_buy_id, product_id in self.media_buy_products
        )

Trusted account-to-source mapping, fixed for a configuration generation.

media_buy_products comes from the authenticated source/account mapping, never from buyer JSON or a report-definition identifier. The SDK persists it with admission, checks the effective capability digest on every source turn, and refuses a changed mapping. Reauthorization can withdraw a binding; it cannot silently change historical scope. No credentials belong here.

Static methods

def for_configuration(configuration: ReportingConfiguration,
*,
capabilities_sha256: str,
media_buy_products: tuple[tuple[str, str], ...]) ‑> ReportingProductionSourceBinding

Freeze the exact generation semantics and explicitly resolved products.

Lifecycle changes retain the same semantic generation, matching the ledger's immutable configuration contract. Captured projection inputs independently retain each historical activation/deactivation boundary.

Instance variables

var capabilities_sha256 : str

var configuration_sha256 : str

var generation_key : ReportingConfigurationGenerationKey

var media_buy_products : tuple[tuple[str, str], ...]

var service_context : ReportingProductionSourceContext | None

Methods

def check(self,
configuration: ReportingConfiguration,
capabilities: ReportingSourceCapabilitiesV1,
offering_id: str) ‑> None

Expand source code

def check(
    self,
    configuration: ReportingConfiguration,
    capabilities: ReportingSourceCapabilitiesV1,
    offering_id: str,
) -> None:
    offering = capabilities.offering(offering_id)
    if (
        self.generation_key != configuration.generation_key
        or self.configuration_sha256
        != hashlib.sha256(canonical_json_utf8_v1(_config_payload(configuration))).hexdigest()
        or capabilities.scope != "effective_account"
        or self.capabilities_sha256 != capabilities.capabilities_sha256
        or {pair[0] for pair in self.media_buy_products} != set(configuration.media_buy_ids)
        or (self.media_buy_products and "media_buy" not in offering.constituent_kinds)
        or any(pair[1] not in offering.product_ids for pair in self.media_buy_products)
    ):
        raise failure("BINDING_MISMATCH")

def constituents(self) ‑> tuple[ProductConstituentV1 | PackageItemConstituentV1 | MediaBuyConstituentV1, ...]

Expand source code

def constituents(self) -> tuple[ReportingConstituent, ...]:
    return tuple(
        MediaBuyConstituentV1(
            constituent_id=media_buy_id, media_buy_id=media_buy_id, product_id=product_id
        )
        for media_buy_id, product_id in self.media_buy_products
    )

def document(self) ‑> dict[str, typing.Any]

Expand source code

def document(self) -> dict[str, Any]:
    key = self.generation_key
    document = {
        "account_id": key.account_id,
        "consumer_id": key.consumer_id,
        "delivery_config_id": key.delivery_config_id,
        "delivery_config_version": key.delivery_config_version,
        "capabilities_sha256": self.capabilities_sha256,
        "configuration_sha256": self.configuration_sha256,
        "media_buy_products": [list(pair) for pair in self.media_buy_products],
    }
    if self.service_context is not None:
        document["service_context"] = self.service_context.document()
        document["service_context_sha256"] = hashlib.sha256(
            self.service_context._wire
        ).hexdigest()
    return document