Skip to content

topmark.pipeline.adapters

topmark / pipeline / adapters

Adapters bridging between high-level ProcessingContext and lightweight view protocols.

This module provides helper classes and functions that adapt a full topmark.pipeline.context.model.ProcessingContext into protocol-compatible view objects such as topmark.filetypes.model.PreInsertContextView. These adapters allow downstream components like InsertChecker or other file-type-specific validation utilities to operate without depending on the entire pipeline internals.

The design ensures a stable and memory-efficient interface by exposing only the minimal attributes or iterators required for read-only inspection of file contents and metadata.

All adapters defined here are intentionally lightweight and immutable within their lifecycle.

PreInsertViewAdapter dataclass 

PreInsertViewAdapter(ctx)

Adapter that exposes a ProcessingContext as a PreInsertContextView.

This class extracts only the fields required by the topmark.filetypes.model.PreInsertContextView protocol and presents them through lightweight, read-only attributes.

Attributes:

Name Type Description
lines Iterable[str]

Streaming access to the file lines as provided by the ProcessingContext image view.
newline_style str

Detected newline style ("LF", "CR", "CRLF").
header_processor HeaderProcessor | None

The file-type-specific HeaderProcessor instance.
file_type FileType | None

The resolved FileType instance or None if unresolved.

Parameters:

Name Type Description Default
ctx ProcessingContext

The processing context to wrap.

 required
Source code in src/topmark/pipeline/adapters.py 
67
68
69
70
71
def __init__(self, ctx: ProcessingContext) -> None:
    object.__setattr__(self, "lines", ctx.iter_image_lines())
    object.__setattr__(self, "newline_style", ctx.newline_style)
    object.__setattr__(self, "header_processor", ctx.header_processor)
    object.__setattr__(self, "file_type", ctx.file_type)

as_sequence 

as_sequence(lines)

Convert an iterable or sequence of lines into a list.

This utility ensures that downstream components expecting a concrete list[str] receive one, while gracefully handling None and already-list inputs without unnecessary copying.

Parameters:

Name Type Description Default
lines Sequence[str] | Iterable[str] | None

The source of line strings, which may be a list, tuple, generator, or None.

 required

Returns:

Type Description
list[str]

list[str]: A list of line strings. Returns an empty list when
list[str]

lines is None.
Source code in src/topmark/pipeline/adapters.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def as_sequence(lines: Sequence[str] | Iterable[str] | None) -> list[str]:
    """Convert an iterable or sequence of lines into a list.

    This utility ensures that downstream components expecting a concrete
    ``list[str]`` receive one, while gracefully handling ``None`` and
    already-list inputs without unnecessary copying.

    Args:
        lines: The source of line strings, which may be a list, tuple, generator, or ``None``.

    Returns:
        list[str]: A list of line strings. Returns an empty list when
        ``lines`` is ``None``.
    """
    if lines is None:
        return []
    return list(lines) if not isinstance(lines, list) else lines