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,
)
|