Skip to content

topmark.registry.machine.payloads

topmark / registry / machine / payloads

Payload builders for registry-related machine-readable output.

This module builds JSON-serializable payload objects (plain Python dicts/lists) for registry-focused CLI commands:

  • topmark registry filetypes
  • topmark registry processors
  • topmark registry bindings

The builders here are: - Click-free and console-free. - Deterministic (sorted) so output is stable for tests and downstream tooling. - Focused on producing payloads only (no envelope/record shaping, no serialization).

Envelope/record shaping lives in topmark.registry.machine.envelopes. Serialization to JSON/NDJSON strings lives in topmark.registry.machine.serializers.

build_filetypes_payload

build_filetypes_payload(*, show_details)

Build the registry payload for topmark registry filetypes.

Parameters:

Name Type Description Default
show_details bool

If True, include matching rules, binding state, and structured policy metadata.

required

Returns:

Type Description
FileTypesPayload

List of file type entries, sorted by canonical file type key.

Source code in src/topmark/registry/machine/payloads.py
def build_filetypes_payload(*, show_details: bool) -> FileTypesPayload:
    """Build the registry payload for `topmark registry filetypes`.

    Args:
        show_details: If True, include matching rules, binding state, and
            structured policy metadata.

    Returns:
        List of file type entries, sorted by canonical file type key.
    """
    raw_items: list[FileTypeInfo] = list_filetypes()

    payload: FileTypesPayload = []
    for item in sorted(raw_items, key=lambda it: str(it["qualified_key"])):
        if show_details:
            payload.append(
                FileTypeDetailEntry(
                    local_key=item["local_key"],
                    namespace=item["namespace"],
                    qualified_key=item["qualified_key"],
                    description=item["description"],
                    bound=item["bound"],
                    extensions=list(item["extensions"]),
                    filenames=list(item["filenames"]),
                    patterns=list(item["patterns"]),
                    skip_processing=item["skip_processing"],
                    has_content_matcher=item["has_content_matcher"],
                    has_insert_checker=item["has_insert_checker"],
                    policy=_serialize_filetype_policy(item["policy"]),
                )
            )
        else:
            payload.append(
                FileTypeBriefEntry(
                    local_key=item["local_key"],
                    namespace=item["namespace"],
                    qualified_key=item["qualified_key"],
                    description=item["description"],
                )
            )

    return payload

build_processors_payload

build_processors_payload(*, show_details)

Build the registry payload for topmark registry processors.

Parameters:

Name Type Description Default
show_details bool

If True, include expanded identity and delimiter fields.

required

Returns:

Type Description
ProcessorsPayload

List of processor entries, sorted by canonical processor key.

Source code in src/topmark/registry/machine/payloads.py
def build_processors_payload(*, show_details: bool) -> ProcessorsPayload:
    """Build the registry payload for `topmark registry processors`.

    Args:
        show_details: If True, include expanded identity and delimiter fields.

    Returns:
        List of processor entries, sorted by canonical processor key.
    """
    raw_items: list[ProcessorInfo] = list_processors()

    processors: ProcessorsPayload = []
    for item in sorted(raw_items, key=lambda it: str(it["qualified_key"])):
        if show_details:
            processors.append(
                ProcessorDetailEntry(
                    local_key=item["local_key"],
                    namespace=item["namespace"],
                    qualified_key=item["qualified_key"],
                    description=item["description"],
                    bound=item["bound"],
                    line_indent=item["line_indent"],
                    line_prefix=item["line_prefix"],
                    line_suffix=item["line_suffix"],
                    block_prefix=item["block_prefix"],
                    block_suffix=item["block_suffix"],
                )
            )
        else:
            processors.append(
                ProcessorBriefEntry(
                    local_key=item["local_key"],
                    namespace=item["namespace"],
                    qualified_key=item["qualified_key"],
                    description=item["description"],
                )
            )

    return processors

build_bindings_payload

build_bindings_payload(*, show_details)

Build the registry payload for topmark registry bindings.

Shape

{ "bindings": [...], "unbound_filetypes": [...], "unused_processors": [...], }

Parameters:

Name Type Description Default
show_details bool

If True, expand bindings and reference entries with identity/description objects.

required

Returns:

Type Description
BindingsPayload

A BindingsPayload dict whose lists are deterministically sorted.

Source code in src/topmark/registry/machine/payloads.py
def build_bindings_payload(*, show_details: bool) -> BindingsPayload:
    """Build the registry payload for `topmark registry bindings`.

    Shape:
        {
          "bindings": [...],
          "unbound_filetypes": [...],
          "unused_processors": [...],
        }

    Args:
        show_details: If True, expand bindings and reference entries with
            identity/description objects.

    Returns:
        A `BindingsPayload` dict whose lists are deterministically sorted.
    """
    raw_bindings: list[BindingInfo] = list_bindings()
    raw_filetypes: list[FileTypeInfo] = list_filetypes()
    raw_processors: list[ProcessorInfo] = list_processors()

    bindings: BindingEntriesPayload = []
    for item in sorted(raw_bindings, key=lambda it: str(it["file_type_key"])):
        if show_details:
            bindings.append(
                BindingDetailEntry(
                    file_type_key=item["file_type_key"],
                    file_type_local_key=item["file_type_local_key"],
                    file_type_namespace=item["file_type_namespace"],
                    processor_key=item["processor_key"],
                    processor_local_key=item["processor_local_key"],
                    processor_namespace=item["processor_namespace"],
                    file_type_description=item["file_type_description"],
                    processor_description=item["processor_description"],
                )
            )
        else:
            bindings.append(
                BindingBriefEntry(
                    file_type_key=item["file_type_key"],
                    processor_key=item["processor_key"],
                )
            )

    bound_filetype_keys: set[str] = {item["file_type_key"] for item in raw_bindings}
    unbound_filetypes: list[str | FileTypeRefEntry] = []
    for item in sorted(raw_filetypes, key=lambda it: str(it["qualified_key"])):
        if item["qualified_key"] in bound_filetype_keys:
            continue
        if show_details:
            unbound_filetypes.append(
                FileTypeRefEntry(
                    local_key=item["local_key"],
                    namespace=item["namespace"],
                    qualified_key=item["qualified_key"],
                    description=item["description"],
                )
            )
        else:
            unbound_filetypes.append(item["qualified_key"])

    unused_processors: list[str | ProcessorRefEntry] = []
    for item in sorted(raw_processors, key=lambda it: str(it["qualified_key"])):
        if item["bound"]:
            continue
        if show_details:
            unused_processors.append(_serialize_processor_ref(item))
        else:
            unused_processors.append(item["qualified_key"])

    return BindingsPayload(
        bindings=bindings,
        unbound_filetypes=unbound_filetypes,
        unused_processors=unused_processors,
    )