topmark.registry.bindings

topmark / registry / bindings

Advanced registry for file type to processor bindings.

This module owns the effective relationship layer between registered file types and registered processor definitions. It is intentionally separate from both the file type and processor registries so identity and relationships can be modeled independently.

Binding dataclass

Binding(*, filetype, processor)

Joined metadata view of one file type and its optionally bound processor.

Attributes:

Name	Type	Description
filetype	FileTypeMeta	Serializable metadata for the file type.
processor	ProcessorMeta | None	Serializable metadata for the bound processor, or None if the file type is currently unbound.

BindingRegistry

Composed registry of effective file-type-key to processor-key bindings.

The base binding view is derived from the explicit built-in processor bindings declared in topmark.processors.instances. Local overrides and removals are then layered on top.

as_mapping classmethod

as_mapping()

Return a read-only mapping of effective file type to processor bindings.

Returns:

Type	Description
Mapping[str, str]	Mapping of canonical file type key to canonical processor key.

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

@classmethod
def as_mapping(cls) -> Mapping[str, str]:
    """Return a read-only mapping of effective file type to processor bindings.

    Returns:
        Mapping of canonical file type key to canonical processor key.
    """
    with cls._lock:
        composed: dict[str, str] = cls._compose()
        return MappingProxyType(composed)

get_processor_key classmethod

get_processor_key(file_type_key)

Return the bound canonical processor key for a file type key.

Parameters:

Name	Type	Description	Default
file_type_key	str	Canonical file type key.	required

Returns:

Type	Description
str | None	The bound canonical processor key, or None if the file type is
str | None	currently unbound.

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

@classmethod
def get_processor_key(cls, file_type_key: str) -> str | None:
    """Return the bound canonical processor key for a file type key.

    Args:
        file_type_key: Canonical file type key.

    Returns:
        The bound canonical processor key, or ``None`` if the file type is
        currently unbound.
    """
    with cls._lock:
        return cls._compose().get(file_type_key)

get_filetype_keys classmethod

get_filetype_keys(processor_key)

Return canonical file type keys currently bound to a processor.

Parameters:

Name	Type	Description	Default
processor_key	str	Canonical processor key.	required

Returns:

Type	Description
str	Sorted tuple of canonical file type keys currently bound to the
...	processor.

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

@classmethod
def get_filetype_keys(cls, processor_key: str) -> tuple[str, ...]:
    """Return canonical file type keys currently bound to a processor.

    Args:
        processor_key: Canonical processor key.

    Returns:
        Sorted tuple of canonical file type keys currently bound to the
        processor.
    """
    with cls._lock:
        return tuple(
            sorted(
                filetype_qk
                for filetype_qk, processor_qk in cls._compose().items()
                if processor_qk == processor_key
            )
        )

is_bound classmethod

is_bound(file_type_key)

Return whether a canonical file type key currently has a binding.

Parameters:

Name	Type	Description	Default
file_type_key	str	Canonical file type key.	required

Returns:

Type	Description
bool	True if the file type currently has a binding, else False.

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

@classmethod
def is_bound(cls, file_type_key: str) -> bool:
    """Return whether a canonical file type key currently has a binding.

    Args:
        file_type_key: Canonical file type key.

    Returns:
        ``True`` if the file type currently has a binding, else ``False``.
    """
    with cls._lock:
        return file_type_key in cls._compose()

is_processor_bound classmethod

is_processor_bound(processor_key)

Return whether a canonical processor key is referenced by any binding.

Parameters:

Name	Type	Description	Default
processor_key	str	Canonical processor key.	required

Returns:

Type	Description
bool	True if at least one file type is currently bound to the
bool	processor, else False.

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

@classmethod
def is_processor_bound(cls, processor_key: str) -> bool:
    """Return whether a canonical processor key is referenced by any binding.

    Args:
        processor_key: Canonical processor key.

    Returns:
        ``True`` if at least one file type is currently bound to the
        processor, else ``False``.
    """
    with cls._lock:
        return any(processor_qk == processor_key for processor_qk in cls._compose().values())

bind classmethod

bind(*, file_type_key, processor_key)

Bind a registered file type key to a registered processor key.

Parameters:

Name	Type	Description	Default
file_type_key	str	Canonical file type key to bind.	required
processor_key	str	Canonical processor key to bind to the file type.	required

Raises:

Type	Description
UnknownFileTypeError	If file_type_key does not resolve to a registered file type.
ProcessorBindingError	If processor_key is unknown or if the file type is already bound.

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

@classmethod
def bind(cls, *, file_type_key: str, processor_key: str) -> None:
    """Bind a registered file type key to a registered processor key.

    Args:
        file_type_key: Canonical file type key to bind.
        processor_key: Canonical processor key to bind to the file type.

    Raises:
        UnknownFileTypeError: If `file_type_key` does not resolve to
            a registered file type.
        ProcessorBindingError: If `processor_key` is unknown or if
            the file type is already bound.
    """
    with cls._lock:
        file_type: FileType | None = FileTypeRegistry.get(
            file_type_key,
        )
        if file_type is None:
            raise UnknownFileTypeError(file_type=file_type_key)

        processor: ProcessorDefinition | None = HeaderProcessorRegistry.get(
            processor_key,
        )
        if processor is None:
            raise ProcessorBindingError(
                message=f"Unknown processor qualified key: {processor_key}",
                file_type=file_type_key,
            )

        existing: str | None = cls._compose().get(file_type_key)
        if existing is not None:
            raise ProcessorBindingError(
                message=(
                    f"File type '{file_type_key}' is already bound to processor '{existing}'."
                ),
                file_type=file_type_key,
            )

        cls._removals.discard(file_type_key)
        cls._overrides[file_type_key] = processor.qualified_key

unbind classmethod

unbind(file_type_key)

Remove the effective binding for a canonical file type key.

Parameters:

Name	Type	Description	Default
file_type_key	str	Canonical file type key whose binding should be removed.	required

Returns:

Type	Description
bool	True if a binding existed in the effective view, else False.

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

@classmethod
def unbind(cls, file_type_key: str) -> bool:
    """Remove the effective binding for a canonical file type key.

    Args:
        file_type_key: Canonical file type key whose binding should be removed.

    Returns:
        ``True`` if a binding existed in the effective view, else ``False``.
    """
    with cls._lock:
        existed: bool = file_type_key in cls._overrides or file_type_key in cls._compose()
        cls._overrides.pop(file_type_key, None)
        cls._removals.add(file_type_key)
        return existed

unbind_processor classmethod

unbind_processor(processor_key)

Remove all bindings that currently reference a processor.

Parameters:

Name	Type	Description	Default
processor_key	str	Canonical processor key.	required

Returns:

Type	Description
tuple[str, ...]	Sorted tuple of canonical file type keys that were unbound.

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

@classmethod
def unbind_processor(cls, processor_key: str) -> tuple[str, ...]:
    """Remove all bindings that currently reference a processor.

    Args:
        processor_key: Canonical processor key.

    Returns:
        Sorted tuple of canonical file type keys that were unbound.
    """
    with cls._lock:
        filetype_qks: tuple[str, ...] = tuple(
            sorted(
                filetype_qk
                for filetype_qk, processor_qk in cls._compose().items()
                if processor_qk == processor_key
            )
        )
        for filetype_qk in filetype_qks:
            cls._overrides.pop(filetype_qk, None)
            cls._removals.add(filetype_qk)
        return filetype_qks

iter_bindings classmethod

iter_bindings()

Iterate the effective joined bindings of file types and processors.

Yields:

Name	Type	Description
Binding	Binding	A pair containing file type metadata and the optionally
	Binding	bound processor metadata (None for unbound types).

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

@classmethod
def iter_bindings(cls) -> Iterator[Binding]:
    """Iterate the effective joined bindings of file types and processors.

    Yields:
        Binding: A pair containing file type metadata and the optionally
        bound processor metadata (``None`` for unbound types).
    """
    filetypes_by_qualified_key: dict[str, FileTypeMeta] = {
        meta.qualified_key: meta for meta in FileTypeRegistry.iter_meta()
    }
    processors_by_qualified_key: dict[str, ProcessorMeta] = {
        meta.qualified_key: meta for meta in HeaderProcessorRegistry.iter_meta()
    }

    for filetype_qualified_key, filetype_meta in sorted(filetypes_by_qualified_key.items()):
        processor_qualified_key: str | None = cls.get_processor_key(
            filetype_qualified_key,
        )
        yield Binding(
            filetype=filetype_meta,
            processor=(
                processors_by_qualified_key.get(processor_qualified_key)
                if processor_qualified_key is not None
                else None
            ),
        )

iter_meta classmethod

iter_meta()

Iterate over effective (file_type_key, processor_key) pairs.

Yields:

Type	Description
tuple[str, str]	Tuples of (filetype_key, processor_key) in sorted order.

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

@classmethod
def iter_meta(cls) -> Iterator[tuple[str, str]]:
    """Iterate over effective ``(file_type_key, processor_key)`` pairs.

    Yields:
        Tuples of ``(filetype_key, processor_key)`` in sorted order.
    """
    with cls._lock:
        yield from sorted(cls._compose().items())

bound_filetype_local_keys classmethod

bound_filetype_local_keys()

Return sorted local keys of file types that currently have a binding.

Returns:

Type	Description
tuple[str, ...]	Sorted tuple of file type local keys that are currently bound.

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

@classmethod
def bound_filetype_local_keys(cls) -> tuple[str, ...]:
    """Return sorted local keys of file types that currently have a binding.

    Returns:
        Sorted tuple of file type local keys that are currently bound.
    """
    with cls._lock:
        return tuple(
            sorted(
                {  # Use set comprehension to remove local key duplicates
                    ft.local_key
                    for ft in FileTypeRegistry.iter_meta_by_local_key()
                    if cls.is_bound(ft.qualified_key)
                }
            )
        )

unbound_filetype_local_keys classmethod

unbound_filetype_local_keys()

Return sorted local keys of recognized file types that currently lack a binding.

Returns:

Type	Description
tuple[str, ...]	Sorted tuple of file type local keys that are currently unbound.

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

@classmethod
def unbound_filetype_local_keys(cls) -> tuple[str, ...]:
    """Return sorted local keys of recognized file types that currently lack a binding.

    Returns:
        Sorted tuple of file type local keys that are currently unbound.
    """
    with cls._lock:
        return tuple(
            sorted(
                {  # Use set comprehension to remove local key duplicates
                    ft.local_key
                    for ft in FileTypeRegistry.iter_meta_by_local_key()
                    if not cls.is_bound(ft.qualified_key)
                }
            )
        )