Module adcp.reporting.feed.store

Optional feed participant. All legacy required protocols remain unchanged.

Classes

class ReportingFeedStore (*args, **kwargs)

Expand source code

@runtime_checkable
class ReportingFeedStore(Protocol):
    """Trusted caller seam. The mounted handler reauthorizes every request.

    The request-local callback rechecks the same canonical principal before
    publishing a prepared snapshot or returning a continuation. It is never
    retained in a store, snapshot, or shared handler state.
    """

    async def read_reporting_feed(
        self,
        request: dict[str, Any],
        *,
        caller: ReportingDeliveryPrincipal,
        consumer_status_enabled: bool = False,
        reauthorize: Callable[[], Awaitable[None]] | None = None,
    ) -> dict[str, Any]: ...

    async def read_reporting_feed_snapshot(
        self,
        snapshot_id: str,
        *,
        caller: ReportingDeliveryPrincipal,
    ) -> ReportingFeedSnapshot | None: ...

    async def reporting_feed_ready(self) -> bool: ...

Trusted caller seam. The mounted handler reauthorizes every request.

The request-local callback rechecks the same canonical principal before publishing a prepared snapshot or returning a continuation. It is never retained in a store, snapshot, or shared handler state.

Ancestors

• typing.Protocol
• typing.Generic

Methods

async def read_reporting_feed(self,
request: dict[str, Any],
*,
caller: ReportingDeliveryPrincipal,
consumer_status_enabled: bool = False,
reauthorize: Callable[[], Awaitable[None]] | None = None) ‑> dict[str, typing.Any]

Expand source code

async def read_reporting_feed(
    self,
    request: dict[str, Any],
    *,
    caller: ReportingDeliveryPrincipal,
    consumer_status_enabled: bool = False,
    reauthorize: Callable[[], Awaitable[None]] | None = None,
) -> dict[str, Any]: ...

async def read_reporting_feed_snapshot(self, snapshot_id: str, *, caller: ReportingDeliveryPrincipal) ‑> ReportingFeedSnapshot | None

Expand source code

async def read_reporting_feed_snapshot(
    self,
    snapshot_id: str,
    *,
    caller: ReportingDeliveryPrincipal,
) -> ReportingFeedSnapshot | None: ...

async def reporting_feed_ready(self) ‑> bool

Expand source code

async def reporting_feed_ready(self) -> bool: ...