aeat.core.json_contract module

Shared primitives for the CLI’s strict --json output contract.

Defines the strict pydantic v2 bases (OutputSchema, OutputRootSchema), the canonical success envelope (SchemaEnvelope), the typed diagnostic channel (Notice), the schema registry (SCHEMA_REGISTRY), and the emit helpers (emit_json_document(), emit_json_success()) used by every registered machine-output path. CLI payload modules import these primitives through entrypoints.cli._schemas, register result models with register_schema(), and route JSON mode through entrypoints.cli._common._emit_envelope().

emit_json_success() derives EnvelopeStatus from supplied Notice values via derive_status() and applies core.redaction.redact_structured_for_cli_output() to the entire envelope before writing stdout. Text output remains owned by core.output_rendering.render_command_output(), so redaction and the reveal_cli_identifiers_opt_in switch stay consistent across text and JSON surfaces.

Living in core keeps domain and adapter packages free of any dependency on entrypoints.cli: a wrapped command emits its strict-validated payload through emit_json_success() without having to know how the CLI itself wires Click options.

This module owns the stdout success contract and schema registry. The stderr failure document is rendered by core.errors using the same ENVELOPE_SCHEMA_VERSION, and text-mode layout remains owned by core.output_rendering.

ENVELOPE_SCHEMA_VERSION

Envelope contract version shared by the success SchemaEnvelope and the stderr error envelope. Both documents carry the same outer spine (schema_version, command, status, notices), so the version is pinned once here and bumped only on a backwards-incompatible change to that spine.

class EnvelopeStatus(*values)[source]

Bases: StrEnum

Outcome discriminator carried on every CLI return document.

success and warning ride on the stdout SchemaEnvelope (warning when the command attached at least one warning-severity Notice); error rides on the stderr error envelope. A machine consumer reads this single field to learn the outcome instead of branching on stdout-vs-stderr, and derive_status() is the success-envelope authority for computing it.

emit_json_success() never emits ERROR; blocking failures route through the shared AeatError boundary instead of being smuggled into stdout notices.

SUCCESS
WARNING
ERROR
class NoticeSeverity(*values)[source]

Bases: StrEnum

Severity of a single operator-facing Notice.

info is a non-fatal next-step hint or informational advisory; warning is a non-blocking advisory the operator should act on. A command that attaches any warning notice resolves to EnvelopeStatus.WARNING through derive_status().

INFO
WARNING
class Notice(**data)[source]

Bases: BaseModel

One typed, non-blocking diagnostic on the envelope notices channel.

The single uniform surface for operator-facing warnings, advisories, and next-step hints across every command. Domain diagnostics (e.g. ModeloFinding, source-resolution advisories) are projected into this shape rather than re-modelled as bespoke per-command payload fields. CLI helpers such as entrypoints.cli._common._emit_envelope() pass these values to emit_json_success(), while text renderers fold equivalent prose into their line output.

Variables:
  • severityinfo or warning; drives the envelope status.

  • code – Stable machine-readable notice identifier (e.g. "modelo.calculate.unconsumed_iva").

  • message – Operator-facing rendered text for the notice.

  • suggestion – Optional copy-paste command or next-step action, mirroring the error envelope’s suggestion field.

  • context – Optional structured provenance for the notice (e.g. the source-resolution reason / source_kind), mirroring the error envelope’s context so a migrated advisory keeps its machine-queryable sub-fields without a bespoke payload model.

Parameters:

Blocking failures are not notices; they raise an AeatError and emit on stderr. Command payload schemas should also avoid reintroducing bespoke advisory, hint, or warning fields inside result when a Notice can carry the same non-blocking diagnostic.

severity: NoticeSeverity
code: str
message: str
suggestion: str | None
context: dict[str, str] | None
derive_status(notices)[source]

Return EnvelopeStatus.WARNING if any notice is warning-severity.

Success documents never carry EnvelopeStatus.ERROR; that status is reserved for the stderr error envelope. The returned EnvelopeStatus is the stdout SchemaEnvelope status used by emit_json_success().

Return type:

EnvelopeStatus

Parameters:

notices (Sequence[Notice])

exception OutputSchemaError(message=None, *, context=None, suggestion=None, translated_message=None)[source]

Bases: AeatError

Raised when the CLI output-schema registry is misconfigured.

Triggered by register_schema() when a non-schema class is decorated, when a command path is registered twice with different schemas, or when the command path is blank. It deliberately inherits core.errors.AeatError so registry defects route through the shared CLI error boundary instead of bypassing structured output.

Parameters:
  • message (str | None)

  • context (dict[str, object] | None)

  • suggestion (str | None)

  • translated_message (str | None)

Return type:

None

code: ClassVar[ErrorCode]
class OutputSchema(**data)[source]

Bases: BaseModel

Strict, frozen base class for every command-specific --json payload.

Subclasses inherit extra="forbid", frozen=True, strict=True, and validate_assignment=True so accidental field drift between contract and implementation surfaces as a validation error rather than a silently-extended payload. Each concrete result model should be decorated with register_schema() so the CLI conformance gate can match command leaves against SCHEMA_REGISTRY.

The class describes the inner result payload only; the outer SchemaEnvelope is applied later by emit_json_success().

class OutputRootSchema(root=PydanticUndefined, **data)[source]

Bases: RootModel[TypeVar], Generic

Strict root/list base class for --json payloads with a non-mapping root.

Use this for commands whose top-level JSON value is a list or scalar rather than an object. Carries the same strict / frozen / validate-on- assignment configuration as OutputSchema, and participates in the same register_schema() registry contract.

Parameters:

root (RootModelRootType)

class SchemaEnvelope(**data)[source]

Bases: BaseModel, Generic

Stable outer envelope wrapping a successful command’s payload.

Every successful --json response is rendered through this envelope so consumers can rely on the same outer keys regardless of the inner payload shape. The outer spine (schema_version, command, status, notices) is shared with the stderr error envelope so one shape describes success, warning, and error outcomes. emit_json_success() constructs the runtime mapping and the JSON-contract conformance gate specialises this generic envelope over every schema in SCHEMA_REGISTRY.

The envelope is a wire contract, not the command dispatcher. It does not discover Click/Typer leaves, choose text output, or own command authorization; those layers supply a strict result and stable command path before entering this contract.

Variables:
  • schema_version – Envelope version; bumped only on backwards-incompatible changes to the shared spine.

  • command – Stable command path string (e.g. "workflow list").

  • active_profile – Human-readable label of the active taxpayer profile (the operator-chosen display name), or None before any profile exists and for non-profile-bound commands. The identity anchor a caller reconciles against; it is the label, never the redacted profile/bucket UUID. Resolved and injected at the CLI transport boundary (the core layer never scans profile manifests), so it stays None for any emitter that does not supply it.

  • status – Outcome discriminator (success or warning here).

  • result – The strict-validated command result.

  • notices – Typed non-blocking diagnostics (warnings, advisories, next-step hints) surfaced to the caller. Replaces the former free-form warnings string list.

Parameters:
schema_version: str
command: str
active_profile: str | None
status: EnvelopeStatus
result: ResultT
notices: list[Notice]
SCHEMA_REGISTRY: dict[str, RegisteredSchema]

Process-global registry mapping command-path strings to their result schema.

Populated by the register_schema() decorator at import time. Consumers (notably the doc generator and the JSON-contract conformance tests) iterate over this mapping to enumerate every contract a release exposes.

emit_json_document(payload, *, indent=2, sort_keys=False, stream=None)[source]

Serialise payload and write a single UTF-8 JSON document followed by \\n.

When stream exposes _ReconfigurableStream.reconfigure, the helper pins it to encoding="utf-8", errors="strict" first so downstream cp1252 consoles can not silently corrupt non-ASCII characters in the rendered output. This is the low-level writer used by emit_json_success(); it does not itself apply the envelope or redaction policy.

Use this for already-shaped JSON documents. Registered CLI success payloads should normally enter through emit_json_success() so the envelope, status derivation, and redaction pass remain uniform.

Parameters:
  • payload (object) – Any object reachable by _jsonable_payload() (typically a pydantic.BaseModel, a mapping, or a collection thereof).

  • indent (int | None) – Indent width passed to json.dumps(); None produces a single-line document.

  • sort_keys (bool) – Whether to render mapping keys in lexicographic order.

  • stream (Optional[IO[str]]) – Target text stream; defaults to sys.stdout.

Return type:

None

emit_json_success(command, result, *, notices=None, active_profile=None, indent=2, sort_keys=False, stream=None)[source]

Wrap result in the success spine and emit it via emit_json_document().

The envelope’s schema_version is pinned to ENVELOPE_SCHEMA_VERSION; bumping it is a contract-breaking change handled by the JSON-contract test suite, not a casual edit. The status is derived from the supplied notices (derive_status()) so the JSON outcome and the shell exit code never disagree. The assembled envelope is redacted through core.redaction.redact_structured_for_cli_output() before emit_json_document() writes it.

This helper is stdout-only. Any raised AeatError is handled by the CLI error boundary, which renders the sibling stderr envelope instead of returning a success document with an error-shaped result.

Parameters:
  • command (str) – Stable command path string (e.g. "workflow list").

  • result (object) – The strict-validated command payload to surface as envelope.result.

  • notices (Sequence[Notice] | None) – Optional typed Notice diagnostics (warnings, advisories, next-step hints); defaults to an empty list.

  • active_profile (str | None) – Optional human label of the active taxpayer profile placed on the shared spine (the identity anchor). The core layer never scans profile manifests, so the CLI transport resolves the label and passes it here; None for non-profile-bound emitters. It rides through the same redaction pass as the rest of the envelope, but it is the non-secret display name, not the redacted profile/bucket UUID.

  • indent (int | None) – Indent width forwarded to emit_json_document().

  • sort_keys (bool) – Sort-keys flag forwarded to emit_json_document().

  • stream (Optional[IO[str]]) – Target text stream; defaults to sys.stdout.

Return type:

None

register_schema(command_path)[source]

Decorator that binds a strict schema to a stable command_path.

Usage:

@register_schema("workflow list")
class WorkflowListResult(OutputSchema):
    ...

The same schema may register the same path more than once (idempotent re-import); registering a different schema under an existing path raises OutputSchemaError. Registered paths are the authoritative command strings emitted as SchemaEnvelope.command and compared against the Typer command tree by the JSON-schema conformance tests.

Register concrete command leaves only. Aliases, helper functions, and text-only utilities do not belong in SCHEMA_REGISTRY unless a real CLI path emits their strict payload through the envelope.

Parameters:

command_path (str) – Stable command-path string used both as the registry key and as the value emitted under SchemaEnvelope.command.

Return type:

Callable[[type[TypeVar(RegisteredSchemaT, bound= OutputSchema | OutputRootSchema[Any])]], type[TypeVar(RegisteredSchemaT, bound= OutputSchema | OutputRootSchema[Any])]]

Returns:

The decorator, returning the schema class unchanged.

Raises:

OutputSchemaError – When command_path is blank, when the decorated class is not a strict schema subclass, or when the path is already bound to a different schema.