topmark.registry.processors

topmark / registry / processors

Advanced public registry for processor definitions.

This module exposes read-oriented views and limited mutation hooks for the composed processor registry used by TopMark. Most callers should prefer the stable facade in topmark.registry.registry.Registry; this module primarily serves advanced integrations, plugins, and tests.

Notes

• Public views such as as_mapping() and iter_meta() are derived from a composed registry (base built-ins + local overlays - removals) and are exposed as MappingProxyType where appropriate.
• register() and unregister() apply overlay-only changes; they do not mutate the base processor-definition registry built from explicit built-in bindings.
• The canonical identity-oriented processor view is keyed by canonical processor key and exposed via as_mapping().
• When the environment variable TOPMARK_VALIDATE is set to a truthy value (1, true, yes), lightweight developer validations run on the composed processor mapping.

HeaderProcessorRegistry

Composed registry view for processor definitions.

Notes

• Only validated HeaderProcessor subclasses are admitted to the effective registry.
• Mutation hooks are intended for plugin authors and test scaffolding. Most integrations should consume metadata and read-only views instead.

qualified_keys classmethod

qualified_keys()

Return qualified keys of all registered processor definitions.

Returns:

Type	Description
tuple[str, ...]	Sorted tuple of processor qualified keys.

Source code in src/topmark/registry/processors.py

@classmethod
def qualified_keys(cls) -> tuple[str, ...]:
    """Return qualified keys of all registered processor definitions.

    Returns:
        Sorted tuple of processor qualified keys.
    """
    with cls._lock:
        # NOTE: keys are already unique
        return tuple(sorted(cls._compose().keys()))

namespaces classmethod

namespaces()

Return namespaces represented in the processor-definition registry.

Returns:

Type	Description
tuple[str, ...]	Sorted tuple of unique processor namespaces.

Source code in src/topmark/registry/processors.py

@classmethod
def namespaces(cls) -> tuple[str, ...]:
    """Return namespaces represented in the processor-definition registry.

    Returns:
        Sorted tuple of unique processor namespaces.
    """
    with cls._lock:
        # Use set comprehension to return "sorted set" of namespaces
        return tuple(sorted({proc.namespace for proc in cls._compose().values()}))

get classmethod

get(processor_key)

Return a processor definition by canonical processor key.

Parameters:

Name	Type	Description	Default
processor_key	str	Canonical processor key.	required

Returns:

Type	Description
ProcessorDefinition | None	Matching ProcessorDefinition, or None if not found.

Source code in src/topmark/registry/processors.py

@classmethod
def get(cls, processor_key: str) -> ProcessorDefinition | None:
    """Return a processor definition by canonical processor key.

    Args:
        processor_key: Canonical processor key.

    Returns:
        Matching `ProcessorDefinition`, or ``None`` if not found.
    """
    with cls._lock:
        return cls._compose().get(processor_key)

as_mapping classmethod

as_mapping()

Return the canonical processor-definition mapping keyed by qualified key.

Returns:

Type	Description
Mapping[str, ProcessorDefinition]	Mapping of canonical processor key to ProcessorDefinition objects.

Notes

The returned mapping is a MappingProxyType and must not be mutated.

Source code in src/topmark/registry/processors.py

@classmethod
def as_mapping(cls) -> Mapping[str, ProcessorDefinition]:
    """Return the canonical processor-definition mapping keyed by qualified key.

    Returns:
        Mapping of canonical processor key to `ProcessorDefinition` objects.

    Notes:
        The returned mapping is a ``MappingProxyType`` and must not be mutated.
    """
    with cls._lock:
        cached: Mapping[str, ProcessorDefinition] | None = cls._cache_by_qualified_key
        if cached is not None:
            return cached

        composed: dict[str, ProcessorDefinition] = cls._compose()
        cls._cache_by_qualified_key = MappingProxyType(composed)
        return cls._cache_by_qualified_key

iter_meta classmethod

iter_meta()

Iterate over stable metadata for processor definitions.

Yields:

Type	Description
ProcessorMeta	Serializable ProcessorMeta values describing each registered
ProcessorMeta	processor definition.

Source code in src/topmark/registry/processors.py

@classmethod
def iter_meta(cls) -> Iterator[ProcessorMeta]:
    """Iterate over stable metadata for processor definitions.

    Yields:
        Serializable `ProcessorMeta` values describing each registered
        processor definition.
    """
    with cls._lock:
        for proc_def in cls._compose().values():
            proc_obj: HeaderProcessor = proc_def.processor_class()
            yield ProcessorMeta(
                namespace=proc_def.namespace,
                local_key=proc_def.local_key,
                description=getattr(proc_obj, "description", "") or "",
                block_prefix=getattr(proc_obj, "block_prefix", "") or "",
                block_suffix=getattr(proc_obj, "block_suffix", "") or "",
                line_indent=getattr(proc_obj, "line_indent", "") or "",
                line_prefix=getattr(proc_obj, "line_prefix", "") or "",
                line_suffix=getattr(proc_obj, "line_suffix", "") or "",
            )

register classmethod

register(*, processor_class)

Register a processor definition.

Parameters:

Name	Type	Description	Default
processor_class	type[HeaderProcessor]	Concrete HeaderProcessor subclass to register.	required

Returns:

Type	Description
ProcessorDefinition	The registered ProcessorDefinition.

Raises:

Type	Description
DuplicateProcessorRegistrationError	If the effective registry already contains a processor for the same qualified key.
TypeError	If processor_class is not a valid HeaderProcessor subclass or if its identity is malformed.
ReservedNamespaceError	If the reserved built-in topmark namespace is used by an ineligible external processor class.

Source code in src/topmark/registry/processors.py

@classmethod
def register(
    cls,
    *,
    processor_class: type[HeaderProcessor],
) -> ProcessorDefinition:
    """Register a processor definition.

    Args:
        processor_class: Concrete `HeaderProcessor` subclass to register.

    Returns:
        The registered `ProcessorDefinition`.

    Raises:
        DuplicateProcessorRegistrationError: If the effective registry
            already contains a processor for the same qualified key.
        TypeError: If `processor_class` is not a valid `HeaderProcessor`
            subclass or if its identity is malformed.
        ReservedNamespaceError: If the reserved built-in ``topmark``
            namespace is used by an ineligible external processor class.
    """  # noqa: DOC503 - documents propagated exceptions from underlying registry helpers
    with cls._lock:
        proc_cls: type[HeaderProcessor] = _validate_processor_class(processor_class)
        proc_def = ProcessorDefinition(
            namespace=proc_cls.namespace,
            local_key=proc_cls.local_key,
            processor_class=proc_cls,
        )
        qualified_key: str = proc_def.qualified_key

        if qualified_key in cls._compose():
            raise DuplicateProcessorRegistrationError(qualified_key=qualified_key)

        cls._removals.discard(qualified_key)
        cls._overrides[qualified_key] = proc_def
        cls._clear_cache()
        return proc_def

unregister classmethod

unregister(processor_key)

Remove a processor definition from the effective registry by canonical processor key.

Parameters:

Name	Type	Description	Default
processor_key	str	Canonical processor key to remove.	required

Returns:

Type	Description
bool	True if the processor definition was present in the effective
bool	registry, else False.

Notes

This mutates process-global registry overlay state.

Source code in src/topmark/registry/processors.py

@classmethod
def unregister(cls, processor_key: str) -> bool:
    """Remove a processor definition from the effective registry by canonical processor key.

    Args:
        processor_key: Canonical processor key to remove.

    Returns:
        ``True`` if the processor definition was present in the effective
        registry, else ``False``.

    Notes:
        This mutates process-global registry overlay state.
    """
    with cls._lock:
        existed = False
        if processor_key in cls._overrides:
            cls._overrides.pop(processor_key, None)
            existed = True
        if processor_key in cls._compose():
            cls._removals.add(processor_key)
            existed = True
        cls._clear_cache()
        return existed