Module adcp.reporting.receipts.schema

Isolated ingestion readiness; never widens an A/B/C/B1/B2.1 manifest.

Functions

async def validate_receipt_schema(connection: Any, *, notifications: bool = False) ‑> None

Expand source code

async def validate_receipt_schema(connection: Any, *, notifications: bool = False) -> None:
    ready = False
    try:
        required = json.loads(
            files("adcp.reporting.receipts").joinpath("required_schema.json").read_text()
        )
        installed = await schema_objects(connection)
        if required and all(installed.get(key) == expected for key, expected in required.items()):
            await validate_materializer_schema(connection, notifications=notifications)
            ready = True
    except Exception:
        ready = False
    if not ready:
        raise ReportingReceiptError("RECEIPT_SCHEMA_UNREADY")