topmark.registry.machine.envelopes

topmark / registry / machine / envelopes

Envelope/record builders for registry machine-readable output.

This module is responsible for shaping registry payloads into TopMark's canonical machine-readable output envelopes.

Layers: - topmark.registry.machine.payloads builds plain JSON-serializable payloads. - This module wraps those payloads into canonical envelopes / NDJSON record objects. - topmark.registry.machine.serializers converts shaped objects into JSON/NDJSON wire strings.

Conventions: - JSON: one envelope object: {"meta": ..., <payload_key>: ...} - NDJSON: one record per entity: {"kind": <kind>, "meta": ..., <payload>}

This module is Click-free and console-free.

build_filetypes_json_envelope

build_filetypes_json_envelope(*, meta, payload)

Build the JSON envelope for topmark registry filetypes.

Parameters:

Name	Type	Description	Default
meta	MetaPayload	Machine metadata payload.	required
payload	FileTypesPayload	List of file type entries.	required

Returns:

Type	Description
dict[str, object]	JSON envelope with keys meta and filetypes.

Source code in src/topmark/registry/machine/envelopes.py

def build_filetypes_json_envelope(
    *,
    meta: MetaPayload,
    payload: FileTypesPayload,
) -> dict[str, object]:
    """Build the JSON envelope for `topmark registry filetypes`.

    Args:
        meta: Machine metadata payload.
        payload: List of file type entries.

    Returns:
        JSON envelope with keys `meta` and `filetypes`.
    """
    return build_json_envelope(
        meta=meta,
        filetypes=payload,
    )

iter_filetypes_ndjson_records

iter_filetypes_ndjson_records(*, meta, payload)

Iterate NDJSON record objects for topmark registry filetypes.

Produces one filetype record per file type entry.

Parameters:

Name	Type	Description	Default
meta	MetaPayload	Machine metadata payload.	required
payload	FileTypesPayload	List of file type entries.	required

Yields:

Type	Description
dict[str, object]	NDJSON record objects (not yet serialized to strings).

Source code in src/topmark/registry/machine/envelopes.py

def iter_filetypes_ndjson_records(
    *,
    meta: MetaPayload,
    payload: FileTypesPayload,
) -> Iterator[dict[str, object]]:
    """Iterate NDJSON record objects for `topmark registry filetypes`.

    Produces one `filetype` record per file type entry.

    Args:
        meta: Machine metadata payload.
        payload: List of file type entries.

    Yields:
        NDJSON record objects (not yet serialized to strings).
    """
    for item in payload:
        entry: FileTypeEntry = item
        yield build_ndjson_record(
            kind=RegistryKind.FILETYPE,
            meta=meta,
            payload=entry,
        )

build_processors_json_envelope

build_processors_json_envelope(*, meta, payload)

Build the JSON envelope for topmark registry processors.

Parameters:

Name	Type	Description	Default
meta	MetaPayload	Machine metadata payload.	required
payload	ProcessorsPayload	List of processor entries.	required

Returns:

Type	Description
dict[str, object]	JSON envelope with keys meta and processors.

Source code in src/topmark/registry/machine/envelopes.py

def build_processors_json_envelope(
    *,
    meta: MetaPayload,
    payload: ProcessorsPayload,
) -> dict[str, object]:
    """Build the JSON envelope for `topmark registry processors`.

    Args:
        meta: Machine metadata payload.
        payload: List of processor entries.

    Returns:
        JSON envelope with keys `meta` and `processors`.
    """
    return build_json_envelope(
        meta=meta,
        processors=payload,
    )

iter_processors_ndjson_records

iter_processors_ndjson_records(*, meta, payload)

Iterate NDJSON record objects for topmark registry processors.

Emits: - one processor record per processor entry

Parameters:

Name	Type	Description	Default
meta	MetaPayload	Machine metadata payload.	required
payload	ProcessorsPayload	Processors payload.	required

Yields:

Type	Description
dict[str, object]	NDJSON record objects (not yet serialized to strings).

Source code in src/topmark/registry/machine/envelopes.py

def iter_processors_ndjson_records(
    *,
    meta: MetaPayload,
    payload: ProcessorsPayload,
) -> Iterator[dict[str, object]]:
    """Iterate NDJSON record objects for `topmark registry processors`.

    Emits:
    - one `processor` record per processor entry

    Args:
        meta: Machine metadata payload.
        payload: Processors payload.

    Yields:
        NDJSON record objects (not yet serialized to strings).
    """
    for proc_item in payload:
        proc_entry: ProcessorEntry = proc_item
        yield build_ndjson_record(
            kind=RegistryKind.PROCESSOR,
            meta=meta,
            payload=proc_entry,
        )

build_bindings_json_envelope

build_bindings_json_envelope(*, meta, payload)

Build the JSON envelope for topmark registry bindings.

Parameters:

Name	Type	Description	Default
meta	MetaPayload	Machine metadata payload.	required
payload	BindingsPayload	Bindings payload.	required

Returns:

Type	Description
dict[str, object]	JSON envelope with keys meta, bindings, unbound_filetypes, and unused_processors.

Source code in src/topmark/registry/machine/envelopes.py

def build_bindings_json_envelope(
    *,
    meta: MetaPayload,
    payload: BindingsPayload,
) -> dict[str, object]:
    """Build the JSON envelope for `topmark registry bindings`.

    Args:
        meta: Machine metadata payload.
        payload: Bindings payload.

    Returns:
        JSON envelope with keys `meta`, `bindings`, `unbound_filetypes`, and `unused_processors`.
    """
    return build_json_envelope(
        meta=meta,
        bindings=payload["bindings"],
        unbound_filetypes=payload["unbound_filetypes"],
        unused_processors=payload["unused_processors"],
    )

iter_bindings_ndjson_records

iter_bindings_ndjson_records(*, meta, payload)

Iterate NDJSON record objects for topmark registry bindings.

Emits: - one binding record per binding entry - one unbound_filetype record per unbound file type reference - one unused_processor record per unused processor reference

Parameters:

Name	Type	Description	Default
meta	MetaPayload	Machine metadata payload.	required
payload	BindingsPayload	Bindings payload.	required

Yields:

Type	Description
dict[str, object]	NDJSON record objects (not yet serialized to strings).

Source code in src/topmark/registry/machine/envelopes.py

def iter_bindings_ndjson_records(
    *,
    meta: MetaPayload,
    payload: BindingsPayload,
) -> Iterator[dict[str, object]]:
    """Iterate NDJSON record objects for `topmark registry bindings`.

    Emits:
    - one `binding` record per binding entry
    - one `unbound_filetype` record per unbound file type reference
    - one `unused_processor` record per unused processor reference

    Args:
        meta: Machine metadata payload.
        payload: Bindings payload.

    Yields:
        NDJSON record objects (not yet serialized to strings).
    """
    for binding_item in payload["bindings"]:
        binding_entry: BindingEntry = binding_item
        yield build_ndjson_record(
            kind=RegistryKind.BINDING,
            meta=meta,
            payload=binding_entry,
        )

    for uft_item in payload["unbound_filetypes"]:
        uft_entry: FileTypeRef = uft_item
        yield build_ndjson_record(
            kind=RegistryKind.UNBOUND_FILETYPE,
            meta=meta,
            payload=uft_entry,
        )

    for unused_item in payload["unused_processors"]:
        proc_entry: ProcessorRef = unused_item
        yield build_ndjson_record(
            kind=RegistryKind.UNUSED_PROCESSOR,
            meta=meta,
            payload=proc_entry,
        )