Skip to content

Syntaxes

Built-in syntaxes, the base class for custom syntaxes, and the syntax factory. See Syntaxes for wire formats and usage.

hother.streamblocks.syntaxes.base

Base syntax class and utilities for StreamBlocks syntax implementations.

BaseSyntax

Bases: ABC

Abstract base class for syntax implementations.

This class provides default implementations and helper methods to reduce code duplication across syntax implementations. It implements the BlockSyntax protocol and provides a template method pattern for parsing blocks.

Custom syntax implementations should: 1. Inherit from this class 2. Implement the abstract methods marked with @abstractmethod 3. Optionally override validate_block() for custom validation

Example

class MySyntax(BaseSyntax): ... def detect_line(self, line: str, candidate: BlockCandidate | None) -> DetectionResult: ... # Implementation here ... pass ... ... def should_accumulate_metadata(self, candidate: BlockCandidate) -> bool: ... # Implementation here ... pass ... ... def extract_block_type(self, candidate: BlockCandidate) -> str | None: ... # Implementation here ... pass ... ... def parse_block(self, candidate: BlockCandidate, block_class: type[Any] | None = None) -> ParseResult[BaseMetadata, BaseContent]: ... # Implementation here ... pass

Source code in src/hother/streamblocks/syntaxes/base.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
class BaseSyntax(ABC):
    """Abstract base class for syntax implementations.

    This class provides default implementations and helper methods to reduce
    code duplication across syntax implementations. It implements the BlockSyntax
    protocol and provides a template method pattern for parsing blocks.

    Custom syntax implementations should:
    1. Inherit from this class
    2. Implement the abstract methods marked with @abstractmethod
    3. Optionally override validate_block() for custom validation

    Example:
        >>> class MySyntax(BaseSyntax):
        ...     def detect_line(self, line: str, candidate: BlockCandidate | None) -> DetectionResult:
        ...         # Implementation here
        ...         pass
        ...
        ...     def should_accumulate_metadata(self, candidate: BlockCandidate) -> bool:
        ...         # Implementation here
        ...         pass
        ...
        ...     def extract_block_type(self, candidate: BlockCandidate) -> str | None:
        ...         # Implementation here
        ...         pass
        ...
        ...     def parse_block(self, candidate: BlockCandidate, block_class: type[Any] | None = None) -> ParseResult[BaseMetadata, BaseContent]:
        ...         # Implementation here
        ...         pass
    """

    # Abstract methods that must be implemented by subclasses

    @abstractmethod
    def detect_line(self, line: str, candidate: BlockCandidate | None) -> DetectionResult:
        """Detect if line is significant for this syntax.

        This method is called for each line in the stream to determine if it's
        an opening marker, closing marker, metadata boundary, or regular content.

        Args:
            line: Current line to check
            candidate: Current candidate if we're inside a block, None if searching

        Returns:
            DetectionResult indicating what was detected
        """
        ...

    @abstractmethod
    def should_accumulate_metadata(self, candidate: BlockCandidate) -> bool:
        """Check if syntax expects more metadata lines.

        This method determines whether the processor should continue accumulating
        metadata lines or move to content accumulation.

        Args:
            candidate: The current block candidate

        Returns:
            True if more metadata lines are expected, False otherwise
        """
        ...

    @abstractmethod
    def extract_block_type(self, candidate: BlockCandidate) -> str | None:
        """Extract block_type from candidate without full parsing.

        This method performs minimal parsing to extract just the block_type,
        which is needed to look up the appropriate block_class from the registry.

        Args:
            candidate: The block candidate to extract block_type from

        Returns:
            The block_type string, or None if it cannot be determined
        """
        ...

    @abstractmethod
    def parse_block(
        self, candidate: BlockCandidate, block_class: type[Any] | None = None
    ) -> ParseResult[BaseMetadata, BaseContent]:
        """Parse a complete block candidate using the specified block class.

        Args:
            candidate: The complete block candidate to parse
            block_class: The Block class to use for parsing (inherits from Block[M, C])
                        If None, uses default base classes

        Returns:
            ParseResult with parsed metadata and content or error
        """
        ...

    # Helper methods for metadata and validation

    def _safe_parse_metadata(
        self,
        metadata_class: type[BaseMetadata],
        data: dict[str, Any],
    ) -> BaseMetadata | ParseResult[BaseMetadata, BaseContent]:
        """Safely parse metadata dict into Pydantic model.

        Helper method to reduce code duplication across syntax implementations.
        Handles ValidationError and returns either the parsed metadata or a
        ParseResult with error information.

        Args:
            metadata_class: The metadata class to instantiate
            data: Dictionary of metadata fields

        Returns:
            Parsed metadata instance, or ParseResult with error on failure
        """
        try:
            return metadata_class(**data)
        except ValidationError as e:
            return ParseResult(
                success=False,
                error=f"Metadata validation error: {e}",
                exception=e,
            )
        except (TypeError, ValueError, KeyError) as e:
            return ParseResult(
                success=False,
                error=f"Invalid metadata: {e}",
                exception=e,
            )

    def _safe_parse_content(
        self,
        content_class: type[BaseContent],
        raw_text: str,
    ) -> BaseContent | ParseResult[BaseMetadata, BaseContent]:
        """Safely parse content into Pydantic model.

        Helper method to reduce code duplication across syntax implementations.
        Handles ValidationError and returns either the parsed content or a
        ParseResult with error information.

        Args:
            content_class: The content class to instantiate
            raw_text: Raw text content to parse

        Returns:
            Parsed content instance, or ParseResult with error on failure
        """
        try:
            return content_class.parse(raw_text)
        except ValidationError as e:
            return ParseResult(
                success=False,
                error=f"Content validation error: {e}",
                exception=e,
            )
        except (TypeError, ValueError, KeyError) as e:
            return ParseResult(
                success=False,
                error=f"Invalid content: {e}",
                exception=e,
            )

    # Default implementations

    def validate_block(self, _block: ExtractedBlock[BaseMetadata, BaseContent]) -> bool:
        """Additional validation after parsing.

        Default implementation always returns True. Override this method
        to add custom validation logic specific to your syntax or block type.

        Args:
            _block: Extracted block to validate

        Returns:
            True if the block is valid, False otherwise
        """
        return True

    def parse_metadata_early(self, candidate: BlockCandidate) -> dict[str, Any] | None:
        """Parse metadata section early, before content accumulation.

        This method is called when the metadata section completes, allowing
        early validation and processing. The result can be cached in the
        candidate for reuse during full block parsing.

        Default implementation returns None (no early parsing).
        Override in subclasses to provide early metadata parsing.

        Args:
            candidate: The current block candidate with metadata accumulated

        Returns:
            Parsed metadata dict if successful, None if parsing not supported
            or failed
        """
        return None

    def parse_content_early(self, candidate: BlockCandidate) -> dict[str, Any] | None:
        """Parse content section early, before final block extraction.

        This method is called when the content section completes (block closes),
        allowing early validation. The result can be cached in the candidate
        for reuse during full block parsing.

        Default implementation returns None (no early parsing).
        Override in subclasses to provide early content parsing.

        Args:
            candidate: The complete block candidate with content accumulated

        Returns:
            Parsed content dict if successful, None if parsing not supported
            or failed
        """
        return None

    # Serialization and documentation (used by prompt generation)

    def serialize_block(self, block: Block[BaseMetadata, BaseContent]) -> str:
        """Serialize a block instance back to this syntax's textual form.

        Used by prompt generation to render examples in the exact format the
        model is expected to produce. The shipped syntaxes override this.
        Custom syntaxes that want example rendering in prompts should override
        it as well.

        Args:
            block: Block instance to serialize

        Returns:
            The block rendered in this syntax's text format

        Raises:
            NotImplementedError: If the concrete syntax does not implement it
        """
        msg = f"{type(self).__name__} does not implement serialize_block()"
        raise NotImplementedError(msg)

    def describe_format(self) -> str:
        """Return a human-readable description of this syntax's block format.

        Used in generated prompts to teach the model the block format. The
        default returns a minimal description; shipped syntaxes override it
        with a concrete format specification.

        Returns:
            A description of the block format for this syntax
        """
        return f"{type(self).__name__}: no format description available."

describe_format 

describe_format() -> str

Return a human-readable description of this syntax's block format.

Used in generated prompts to teach the model the block format. The default returns a minimal description; shipped syntaxes override it with a concrete format specification.

Returns:

Type Description
str

A description of the block format for this syntax
Source code in src/hother/streamblocks/syntaxes/base.py 
321
322
323
324
325
326
327
328
329
330
331
def describe_format(self) -> str:
    """Return a human-readable description of this syntax's block format.

    Used in generated prompts to teach the model the block format. The
    default returns a minimal description; shipped syntaxes override it
    with a concrete format specification.

    Returns:
        A description of the block format for this syntax
    """
    return f"{type(self).__name__}: no format description available."

detect_line abstractmethod 

detect_line(
    line: str, candidate: BlockCandidate | None
) -> DetectionResult

Detect if line is significant for this syntax.

This method is called for each line in the stream to determine if it's an opening marker, closing marker, metadata boundary, or regular content.

Parameters:

Name Type Description Default
line str

Current line to check

 required
candidate BlockCandidate | None

Current candidate if we're inside a block, None if searching

 required

Returns:

Type Description
DetectionResult

DetectionResult indicating what was detected
Source code in src/hother/streamblocks/syntaxes/base.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
@abstractmethod
def detect_line(self, line: str, candidate: BlockCandidate | None) -> DetectionResult:
    """Detect if line is significant for this syntax.

    This method is called for each line in the stream to determine if it's
    an opening marker, closing marker, metadata boundary, or regular content.

    Args:
        line: Current line to check
        candidate: Current candidate if we're inside a block, None if searching

    Returns:
        DetectionResult indicating what was detected
    """
    ...

extract_block_type abstractmethod 

extract_block_type(candidate: BlockCandidate) -> str | None

Extract block_type from candidate without full parsing.

This method performs minimal parsing to extract just the block_type, which is needed to look up the appropriate block_class from the registry.

Parameters:

Name Type Description Default
candidate BlockCandidate

The block candidate to extract block_type from

 required

Returns:

Type Description
str | None

The block_type string, or None if it cannot be determined
Source code in src/hother/streamblocks/syntaxes/base.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
@abstractmethod
def extract_block_type(self, candidate: BlockCandidate) -> str | None:
    """Extract block_type from candidate without full parsing.

    This method performs minimal parsing to extract just the block_type,
    which is needed to look up the appropriate block_class from the registry.

    Args:
        candidate: The block candidate to extract block_type from

    Returns:
        The block_type string, or None if it cannot be determined
    """
    ...

parse_block abstractmethod 

parse_block(
    candidate: BlockCandidate,
    block_class: type[Any] | None = None,
) -> ParseResult[BaseMetadata, BaseContent]

Parse a complete block candidate using the specified block class.

Parameters:

Name Type Description Default
candidate BlockCandidate

The complete block candidate to parse

 required
block_class type[Any] | None

The Block class to use for parsing (inherits from Block[M, C]) If None, uses default base classes

 None

Returns:

Type Description
ParseResult[BaseMetadata, BaseContent]

ParseResult with parsed metadata and content or error
Source code in src/hother/streamblocks/syntaxes/base.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
@abstractmethod
def parse_block(
    self, candidate: BlockCandidate, block_class: type[Any] | None = None
) -> ParseResult[BaseMetadata, BaseContent]:
    """Parse a complete block candidate using the specified block class.

    Args:
        candidate: The complete block candidate to parse
        block_class: The Block class to use for parsing (inherits from Block[M, C])
                    If None, uses default base classes

    Returns:
        ParseResult with parsed metadata and content or error
    """
    ...

parse_content_early 

parse_content_early(
    candidate: BlockCandidate,
) -> dict[str, Any] | None

Parse content section early, before final block extraction.

This method is called when the content section completes (block closes), allowing early validation. The result can be cached in the candidate for reuse during full block parsing.

Default implementation returns None (no early parsing). Override in subclasses to provide early content parsing.

Parameters:

Name Type Description Default
candidate BlockCandidate

The complete block candidate with content accumulated

 required

Returns:

Type Description
dict[str, Any] | None

Parsed content dict if successful, None if parsing not supported
dict[str, Any] | None

or failed
Source code in src/hother/streamblocks/syntaxes/base.py 
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
def parse_content_early(self, candidate: BlockCandidate) -> dict[str, Any] | None:
    """Parse content section early, before final block extraction.

    This method is called when the content section completes (block closes),
    allowing early validation. The result can be cached in the candidate
    for reuse during full block parsing.

    Default implementation returns None (no early parsing).
    Override in subclasses to provide early content parsing.

    Args:
        candidate: The complete block candidate with content accumulated

    Returns:
        Parsed content dict if successful, None if parsing not supported
        or failed
    """
    return None

parse_metadata_early 

parse_metadata_early(
    candidate: BlockCandidate,
) -> dict[str, Any] | None

Parse metadata section early, before content accumulation.

This method is called when the metadata section completes, allowing early validation and processing. The result can be cached in the candidate for reuse during full block parsing.

Default implementation returns None (no early parsing). Override in subclasses to provide early metadata parsing.

Parameters:

Name Type Description Default
candidate BlockCandidate

The current block candidate with metadata accumulated

 required

Returns:

Type Description
dict[str, Any] | None

Parsed metadata dict if successful, None if parsing not supported
dict[str, Any] | None

or failed
Source code in src/hother/streamblocks/syntaxes/base.py 
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
def parse_metadata_early(self, candidate: BlockCandidate) -> dict[str, Any] | None:
    """Parse metadata section early, before content accumulation.

    This method is called when the metadata section completes, allowing
    early validation and processing. The result can be cached in the
    candidate for reuse during full block parsing.

    Default implementation returns None (no early parsing).
    Override in subclasses to provide early metadata parsing.

    Args:
        candidate: The current block candidate with metadata accumulated

    Returns:
        Parsed metadata dict if successful, None if parsing not supported
        or failed
    """
    return None

serialize_block 

serialize_block(
    block: Block[BaseMetadata, BaseContent],
) -> str

Serialize a block instance back to this syntax's textual form.

Used by prompt generation to render examples in the exact format the model is expected to produce. The shipped syntaxes override this. Custom syntaxes that want example rendering in prompts should override it as well.

Parameters:

Name Type Description Default
block Block[BaseMetadata, BaseContent]

Block instance to serialize

 required

Returns:

Type Description
str

The block rendered in this syntax's text format

Raises:

Type Description
NotImplementedError

If the concrete syntax does not implement it
Source code in src/hother/streamblocks/syntaxes/base.py 
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
def serialize_block(self, block: Block[BaseMetadata, BaseContent]) -> str:
    """Serialize a block instance back to this syntax's textual form.

    Used by prompt generation to render examples in the exact format the
    model is expected to produce. The shipped syntaxes override this.
    Custom syntaxes that want example rendering in prompts should override
    it as well.

    Args:
        block: Block instance to serialize

    Returns:
        The block rendered in this syntax's text format

    Raises:
        NotImplementedError: If the concrete syntax does not implement it
    """
    msg = f"{type(self).__name__} does not implement serialize_block()"
    raise NotImplementedError(msg)

should_accumulate_metadata abstractmethod 

should_accumulate_metadata(
    candidate: BlockCandidate,
) -> bool

Check if syntax expects more metadata lines.

This method determines whether the processor should continue accumulating metadata lines or move to content accumulation.

Parameters:

Name Type Description Default
candidate BlockCandidate

The current block candidate

 required

Returns:

Type Description
bool

True if more metadata lines are expected, False otherwise
Source code in src/hother/streamblocks/syntaxes/base.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
@abstractmethod
def should_accumulate_metadata(self, candidate: BlockCandidate) -> bool:
    """Check if syntax expects more metadata lines.

    This method determines whether the processor should continue accumulating
    metadata lines or move to content accumulation.

    Args:
        candidate: The current block candidate

    Returns:
        True if more metadata lines are expected, False otherwise
    """
    ...

validate_block 

validate_block(
    _block: ExtractedBlock[BaseMetadata, BaseContent],
) -> bool

Additional validation after parsing.

Default implementation always returns True. Override this method to add custom validation logic specific to your syntax or block type.

Parameters:

Name Type Description Default
_block ExtractedBlock[BaseMetadata, BaseContent]

Extracted block to validate

 required

Returns:

Type Description
bool

True if the block is valid, False otherwise
Source code in src/hother/streamblocks/syntaxes/base.py 
247
248
249
250
251
252
253
254
255
256
257
258
259
def validate_block(self, _block: ExtractedBlock[BaseMetadata, BaseContent]) -> bool:
    """Additional validation after parsing.

    Default implementation always returns True. Override this method
    to add custom validation logic specific to your syntax or block type.

    Args:
        _block: Extracted block to validate

    Returns:
        True if the block is valid, False otherwise
    """
    return True

YAMLFrontmatterMixin

Mixin providing YAML frontmatter parsing utilities.

This mixin reduces code duplication in syntaxes that use YAML for metadata. It provides two parsing methods: - _parse_yaml_metadata: Silent failure, returns None on error - _parse_yaml_metadata_strict: Returns exception for error handling

Example

class MySyntax(BaseSyntax, YAMLFrontmatterMixin): ... def extract_block_type(self, candidate): ... metadata = self._parse_yaml_metadata(candidate.metadata_lines) ... return metadata.get("block_type") if metadata else None

Source code in src/hother/streamblocks/syntaxes/base.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
class YAMLFrontmatterMixin:
    """Mixin providing YAML frontmatter parsing utilities.

    This mixin reduces code duplication in syntaxes that use YAML for metadata.
    It provides two parsing methods:
    - _parse_yaml_metadata: Silent failure, returns None on error
    - _parse_yaml_metadata_strict: Returns exception for error handling

    Example:
        >>> class MySyntax(BaseSyntax, YAMLFrontmatterMixin):
        ...     def extract_block_type(self, candidate):
        ...         metadata = self._parse_yaml_metadata(candidate.metadata_lines)
        ...         return metadata.get("block_type") if metadata else None
    """

    def _parse_yaml_metadata(self, metadata_lines: list[str]) -> dict[str, Any] | None:
        """Parse YAML from metadata lines. Returns None on error.

        Args:
            metadata_lines: Lines containing YAML content

        Returns:
            Parsed YAML as dict, or None if parsing fails or lines are empty
        """
        if not metadata_lines:
            return None
        yaml_content = "\n".join(metadata_lines)
        try:
            return yaml.safe_load(yaml_content) or {}
        except yaml.YAMLError as e:
            # Log parse failure for debugging
            _logger.debug(
                "YAML parse failed: %s (lines=%d, preview=%s)",
                str(e),
                len(metadata_lines),
                yaml_content[:100] if yaml_content else "",
            )
            return None

    def _parse_yaml_metadata_strict(self, metadata_lines: list[str]) -> tuple[dict[str, Any], Exception | None]:
        """Parse YAML, returning both result and exception.

        Use this method when you need to report the error in a ParseResult.

        Args:
            metadata_lines: Lines containing YAML content

        Returns:
            Tuple of (parsed dict, exception or None)
        """
        if not metadata_lines:
            return {}, None
        yaml_content = "\n".join(metadata_lines)
        try:
            return yaml.safe_load(yaml_content) or {}, None
        except yaml.YAMLError as e:
            return {}, e

hother.streamblocks.syntaxes.delimiter

Delimiter-based syntax implementations.

ContentParser

Bases: Protocol

Protocol for content classes with a parse method.

Source code in src/hother/streamblocks/syntaxes/delimiter.py 
21
22
23
24
25
26
27
28
@runtime_checkable
class ContentParser(Protocol):
    """Protocol for content classes with a parse method."""

    @classmethod
    def parse(cls, raw_text: str) -> BaseContent:
        """Parse raw text into content."""
        ...

parse classmethod 

parse(raw_text: str) -> BaseContent

Parse raw text into content.

Source code in src/hother/streamblocks/syntaxes/delimiter.py 
25
26
27
28
@classmethod
def parse(cls, raw_text: str) -> BaseContent:
    """Parse raw text into content."""
    ...

DelimiterFrontmatterSyntax

Bases: BaseSyntax, YAMLFrontmatterMixin

This syntax uses simple delimiter markers with YAML frontmatter for metadata. The frontmatter section is delimited by --- markers and must be valid YAML.

Format
!!start

id: block_001 block_type: example custom_field: value

Content lines here !!end

The YAML frontmatter should include
  • id: Block identifier (required if using BaseMetadata)
  • block_type: Block type (required if using BaseMetadata)
  • Any additional custom fields defined in your metadata class

Examples:

>>> # Simple block with minimal metadata
>>> '''
... !!start
... ---
... id: msg001
... block_type: message
... ---
... Hello, world!
... !!end
... '''
>>>
>>> # Block with nested YAML metadata
>>> '''
... !!start
... ---
... id: task001
... block_type: task
... priority: high
... tags:
...   - urgent
...   - backend
... ---
... Implement user authentication
... !!end
... '''

Parameters:

Name Type Description Default
start_delimiter str

Opening delimiter string (default: "!!start")

 '!!start'
end_delimiter str

Closing delimiter string (default: "!!end")

 '!!end'
Source code in src/hother/streamblocks/syntaxes/delimiter.py 
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
class DelimiterFrontmatterSyntax(BaseSyntax, YAMLFrontmatterMixin):
    """Syntax: Delimiter markers with YAML frontmatter.

    This syntax uses simple delimiter markers with YAML frontmatter for metadata.
    The frontmatter section is delimited by --- markers and must be valid YAML.

    Format:
        !!start
        ---
        id: block_001
        block_type: example
        custom_field: value
        ---
        Content lines here
        !!end

    The YAML frontmatter should include:
        - id: Block identifier (required if using BaseMetadata)
        - block_type: Block type (required if using BaseMetadata)
        - Any additional custom fields defined in your metadata class

    Examples:
        >>> # Simple block with minimal metadata
        >>> '''
        ... !!start
        ... ---
        ... id: msg001
        ... block_type: message
        ... ---
        ... Hello, world!
        ... !!end
        ... '''
        >>>
        >>> # Block with nested YAML metadata
        >>> '''
        ... !!start
        ... ---
        ... id: task001
        ... block_type: task
        ... priority: high
        ... tags:
        ...   - urgent
        ...   - backend
        ... ---
        ... Implement user authentication
        ... !!end
        ... '''

    Args:
        start_delimiter: Opening delimiter string (default: "!!start")
        end_delimiter: Closing delimiter string (default: "!!end")
    """

    def __init__(
        self,
        start_delimiter: str = "!!start",
        end_delimiter: str = "!!end",
    ) -> None:
        """Initialize delimiter frontmatter syntax.

        Args:
            start_delimiter: Starting delimiter
            end_delimiter: Ending delimiter
        """
        self.start_delimiter = start_delimiter
        self.end_delimiter = end_delimiter
        self._frontmatter_pattern = re.compile(r"^---\s*$")

    def detect_line(self, line: str, candidate: BlockCandidate | None = None) -> DetectionResult:
        """Detect delimiter markers and frontmatter boundaries."""
        if candidate is None:
            # Looking for opening
            if line.strip() == self.start_delimiter:
                return DetectionResult(is_opening=True)
        # Inside a block
        elif candidate.current_section == SectionType.HEADER:
            # Should be frontmatter start
            if self._frontmatter_pattern.match(line):
                candidate.transition_to_metadata()
                return DetectionResult(is_metadata_boundary=True)
            # Skip empty lines in header - frontmatter might follow
            if line.strip() == "":
                return DetectionResult()
            # Move directly to content if no frontmatter
            candidate.transition_to_content()
            candidate.content_lines.append(line)
        elif candidate.current_section == SectionType.METADATA:
            if self._frontmatter_pattern.match(line):
                candidate.transition_to_content()
                return DetectionResult(is_metadata_boundary=True)
            candidate.metadata_lines.append(line)
        elif candidate.current_section == SectionType.CONTENT:
            if line.strip() == self.end_delimiter:
                return DetectionResult(is_closing=True)
            candidate.content_lines.append(line)

        return DetectionResult()

    def should_accumulate_metadata(self, candidate: BlockCandidate) -> bool:
        """Check if we're still in metadata section."""
        return candidate.current_section in {SectionType.HEADER, SectionType.METADATA}

    def extract_block_type(self, candidate: BlockCandidate) -> str | None:
        """Extract block_type from YAML frontmatter."""
        metadata_dict = self._parse_yaml_metadata(candidate.metadata_lines)
        if metadata_dict and "block_type" in metadata_dict:
            return str(metadata_dict["block_type"])
        return None

    def parse_block(
        self, candidate: BlockCandidate, block_class: type[Any] | None = None
    ) -> ParseResult[BaseMetadata, BaseContent]:
        """Parse the complete block using the specified block class."""

        # Extract metadata and content classes from block_class
        if block_class is None:
            # Default to base classes
            metadata_class = BaseMetadata
            content_class = BaseContent
        else:
            # Extract from block class using type parameters
            metadata_class, content_class = extract_block_types(block_class)

        # Parse metadata from accumulated metadata lines
        metadata_dict, yaml_error = self._parse_yaml_metadata_strict(candidate.metadata_lines)
        if yaml_error:
            return ParseResult(success=False, error=f"YAML parse error: {yaml_error}", exception=yaml_error)

        # Parse metadata using helper
        metadata = self._safe_parse_metadata(metadata_class, metadata_dict)
        if isinstance(metadata, ParseResult):
            return metadata  # Return error

        # Parse content using helper
        content_text = "\n".join(candidate.content_lines)
        content = self._safe_parse_content(content_class, content_text)
        if isinstance(content, ParseResult):
            return content  # Return error

        return ParseResult(success=True, metadata=metadata, content=content)

    def validate_block(self, _block: ExtractedBlock[BaseMetadata, BaseContent]) -> bool:
        """Additional validation after parsing."""
        return True

    def parse_metadata_early(self, candidate: BlockCandidate) -> dict[str, Any] | None:
        """Parse YAML metadata section early.

        Returns parsed YAML frontmatter as a dict.
        """
        return self._parse_yaml_metadata(candidate.metadata_lines)

    def parse_content_early(self, candidate: BlockCandidate) -> dict[str, Any] | None:
        """Parse content section early.

        Returns raw content dict with the content text.
        """
        raw_content = "\n".join(candidate.content_lines)
        return {"raw_content": raw_content}

    def serialize_block(self, block: Block[BaseMetadata, BaseContent]) -> str:
        """Serialize a block to delimiter frontmatter format.

        Produces the round-trip form of this syntax::

            !!start
            ---
            <yaml metadata>
            ---
            <raw content>
            !!end

        Args:
            block: Block instance to serialize

        Returns:
            The block rendered in delimiter frontmatter format
        """
        metadata_yaml = yaml.dump(block.metadata.model_dump(), default_flow_style=False, sort_keys=False)
        return f"{self.start_delimiter}\n---\n{metadata_yaml}---\n{block.content.raw_content}\n{self.end_delimiter}"

    def describe_format(self) -> str:
        """Describe the delimiter frontmatter format for prompts."""
        return (
            "Delimiter Frontmatter Syntax\n\n"
            "Format:\n"
            f"{self.start_delimiter}\n"
            "---\n"
            "key: value\n"
            "---\n"
            "content lines\n"
            f"{self.end_delimiter}\n\n"
            "Components:\n"
            f"- Opening marker: {self.start_delimiter}\n"
            "- Metadata section: YAML frontmatter between --- delimiters\n"
            "- Content: any text after the metadata section\n"
            f"- Closing marker: {self.end_delimiter}"
        )

end_delimiter instance-attribute 

end_delimiter = end_delimiter

start_delimiter instance-attribute 

start_delimiter = start_delimiter

describe_format 

describe_format() -> str

Describe the delimiter frontmatter format for prompts.

Source code in src/hother/streamblocks/syntaxes/delimiter.py 
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
def describe_format(self) -> str:
    """Describe the delimiter frontmatter format for prompts."""
    return (
        "Delimiter Frontmatter Syntax\n\n"
        "Format:\n"
        f"{self.start_delimiter}\n"
        "---\n"
        "key: value\n"
        "---\n"
        "content lines\n"
        f"{self.end_delimiter}\n\n"
        "Components:\n"
        f"- Opening marker: {self.start_delimiter}\n"
        "- Metadata section: YAML frontmatter between --- delimiters\n"
        "- Content: any text after the metadata section\n"
        f"- Closing marker: {self.end_delimiter}"
    )

detect_line 

detect_line(
    line: str, candidate: BlockCandidate | None = None
) -> DetectionResult

Detect delimiter markers and frontmatter boundaries.

Source code in src/hother/streamblocks/syntaxes/delimiter.py 
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
def detect_line(self, line: str, candidate: BlockCandidate | None = None) -> DetectionResult:
    """Detect delimiter markers and frontmatter boundaries."""
    if candidate is None:
        # Looking for opening
        if line.strip() == self.start_delimiter:
            return DetectionResult(is_opening=True)
    # Inside a block
    elif candidate.current_section == SectionType.HEADER:
        # Should be frontmatter start
        if self._frontmatter_pattern.match(line):
            candidate.transition_to_metadata()
            return DetectionResult(is_metadata_boundary=True)
        # Skip empty lines in header - frontmatter might follow
        if line.strip() == "":
            return DetectionResult()
        # Move directly to content if no frontmatter
        candidate.transition_to_content()
        candidate.content_lines.append(line)
    elif candidate.current_section == SectionType.METADATA:
        if self._frontmatter_pattern.match(line):
            candidate.transition_to_content()
            return DetectionResult(is_metadata_boundary=True)
        candidate.metadata_lines.append(line)
    elif candidate.current_section == SectionType.CONTENT:
        if line.strip() == self.end_delimiter:
            return DetectionResult(is_closing=True)
        candidate.content_lines.append(line)

    return DetectionResult()

extract_block_type 

extract_block_type(candidate: BlockCandidate) -> str | None

Extract block_type from YAML frontmatter.

Source code in src/hother/streamblocks/syntaxes/delimiter.py 
349
350
351
352
353
354
def extract_block_type(self, candidate: BlockCandidate) -> str | None:
    """Extract block_type from YAML frontmatter."""
    metadata_dict = self._parse_yaml_metadata(candidate.metadata_lines)
    if metadata_dict and "block_type" in metadata_dict:
        return str(metadata_dict["block_type"])
    return None

parse_block 

parse_block(
    candidate: BlockCandidate,
    block_class: type[Any] | None = None,
) -> ParseResult[BaseMetadata, BaseContent]

Parse the complete block using the specified block class.

Source code in src/hother/streamblocks/syntaxes/delimiter.py 
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
def parse_block(
    self, candidate: BlockCandidate, block_class: type[Any] | None = None
) -> ParseResult[BaseMetadata, BaseContent]:
    """Parse the complete block using the specified block class."""

    # Extract metadata and content classes from block_class
    if block_class is None:
        # Default to base classes
        metadata_class = BaseMetadata
        content_class = BaseContent
    else:
        # Extract from block class using type parameters
        metadata_class, content_class = extract_block_types(block_class)

    # Parse metadata from accumulated metadata lines
    metadata_dict, yaml_error = self._parse_yaml_metadata_strict(candidate.metadata_lines)
    if yaml_error:
        return ParseResult(success=False, error=f"YAML parse error: {yaml_error}", exception=yaml_error)

    # Parse metadata using helper
    metadata = self._safe_parse_metadata(metadata_class, metadata_dict)
    if isinstance(metadata, ParseResult):
        return metadata  # Return error

    # Parse content using helper
    content_text = "\n".join(candidate.content_lines)
    content = self._safe_parse_content(content_class, content_text)
    if isinstance(content, ParseResult):
        return content  # Return error

    return ParseResult(success=True, metadata=metadata, content=content)

parse_content_early 

parse_content_early(
    candidate: BlockCandidate,
) -> dict[str, Any] | None

Parse content section early.

Returns raw content dict with the content text.

Source code in src/hother/streamblocks/syntaxes/delimiter.py 
399
400
401
402
403
404
405
def parse_content_early(self, candidate: BlockCandidate) -> dict[str, Any] | None:
    """Parse content section early.

    Returns raw content dict with the content text.
    """
    raw_content = "\n".join(candidate.content_lines)
    return {"raw_content": raw_content}

parse_metadata_early 

parse_metadata_early(
    candidate: BlockCandidate,
) -> dict[str, Any] | None

Parse YAML metadata section early.

Returns parsed YAML frontmatter as a dict.

Source code in src/hother/streamblocks/syntaxes/delimiter.py 
392
393
394
395
396
397
def parse_metadata_early(self, candidate: BlockCandidate) -> dict[str, Any] | None:
    """Parse YAML metadata section early.

    Returns parsed YAML frontmatter as a dict.
    """
    return self._parse_yaml_metadata(candidate.metadata_lines)

serialize_block 

serialize_block(
    block: Block[BaseMetadata, BaseContent],
) -> str

Serialize a block to delimiter frontmatter format.

Produces the round-trip form of this syntax::

!!start
---
<yaml metadata>
---
<raw content>
!!end

Parameters:

Name Type Description Default
block Block[BaseMetadata, BaseContent]

Block instance to serialize

 required

Returns:

Type Description
str

The block rendered in delimiter frontmatter format
Source code in src/hother/streamblocks/syntaxes/delimiter.py 
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
def serialize_block(self, block: Block[BaseMetadata, BaseContent]) -> str:
    """Serialize a block to delimiter frontmatter format.

    Produces the round-trip form of this syntax::

        !!start
        ---
        <yaml metadata>
        ---
        <raw content>
        !!end

    Args:
        block: Block instance to serialize

    Returns:
        The block rendered in delimiter frontmatter format
    """
    metadata_yaml = yaml.dump(block.metadata.model_dump(), default_flow_style=False, sort_keys=False)
    return f"{self.start_delimiter}\n---\n{metadata_yaml}---\n{block.content.raw_content}\n{self.end_delimiter}"

should_accumulate_metadata 

should_accumulate_metadata(
    candidate: BlockCandidate,
) -> bool

Check if we're still in metadata section.

Source code in src/hother/streamblocks/syntaxes/delimiter.py 
345
346
347
def should_accumulate_metadata(self, candidate: BlockCandidate) -> bool:
    """Check if we're still in metadata section."""
    return candidate.current_section in {SectionType.HEADER, SectionType.METADATA}

validate_block 

validate_block(
    _block: ExtractedBlock[BaseMetadata, BaseContent],
) -> bool

Additional validation after parsing.

Source code in src/hother/streamblocks/syntaxes/delimiter.py 
388
389
390
def validate_block(self, _block: ExtractedBlock[BaseMetadata, BaseContent]) -> bool:
    """Additional validation after parsing."""
    return True

DelimiterPreambleSyntax

Bases: BaseSyntax

This syntax uses delimiter markers with inline metadata in the opening line. Metadata is extracted from the delimiter preamble, and all lines between opening and closing delimiters become the content.

Format

!!:[:param1:param2:...] Content lines here !!end

The opening delimiter must include
  • Block ID (alphanumeric, required)
  • Block type (alphanumeric, required)
  • Additional parameters (optional, colon-separated)

Additional parameters are stored as param_0, param_1, etc. in metadata.

Examples:

>>> # Simple block with just ID and type
>>> '''
... !!patch001:patch
... Fix the login bug
... !!end
... '''
>>>
>>> # Block with parameters
>>> '''
... !!file123:operation:create:urgent
... Create new config file
... !!end
... '''
>>> # Metadata will be: {
>>> #     "id": "file123",
>>> #     "block_type": "operation",
>>> #     "param_0": "create",
>>> #     "param_1": "urgent"
>>> # }

Parameters:

Name Type Description Default
delimiter str

Opening delimiter string (default: "!!")

 '!!'
Source code in src/hother/streamblocks/syntaxes/delimiter.py 
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
class DelimiterPreambleSyntax(BaseSyntax):
    """Syntax: !! delimiter with inline metadata.

    This syntax uses delimiter markers with inline metadata in the opening line.
    Metadata is extracted from the delimiter preamble, and all lines between
    opening and closing delimiters become the content.

    Format:
        !!<id>:<type>[:param1:param2:...]
        Content lines here
        !!end

    The opening delimiter must include:
        - Block ID (alphanumeric, required)
        - Block type (alphanumeric, required)
        - Additional parameters (optional, colon-separated)

    Additional parameters are stored as param_0, param_1, etc. in metadata.

    Examples:
        >>> # Simple block with just ID and type
        >>> '''
        ... !!patch001:patch
        ... Fix the login bug
        ... !!end
        ... '''
        >>>
        >>> # Block with parameters
        >>> '''
        ... !!file123:operation:create:urgent
        ... Create new config file
        ... !!end
        ... '''
        >>> # Metadata will be: {
        >>> #     "id": "file123",
        >>> #     "block_type": "operation",
        >>> #     "param_0": "create",
        >>> #     "param_1": "urgent"
        >>> # }

    Args:
        delimiter: Opening delimiter string (default: "!!")
    """

    def __init__(
        self,
        delimiter: str = "!!",
    ) -> None:
        """Initialize delimiter preamble syntax.

        Args:
            delimiter: Delimiter string to use
        """
        self.delimiter = delimiter
        self._opening_pattern = re.compile(rf"^{re.escape(delimiter)}(\w+):(\w+)(:.+)?$")
        self._closing_pattern = re.compile(rf"^{re.escape(delimiter)}end$")

    def detect_line(self, line: str, candidate: BlockCandidate | None = None) -> DetectionResult:
        """Detect delimiter-based markers."""
        if candidate is None:
            # Looking for opening
            match = self._opening_pattern.match(line)
            if match:
                block_id, block_type, params = match.groups()
                metadata_dict: dict[str, object] = {
                    "id": block_id,
                    "block_type": block_type,
                }

                if params:
                    param_parts = params[1:].split(":")
                    for i, part in enumerate(param_parts):
                        metadata_dict[f"param_{i}"] = part

                return DetectionResult(
                    is_opening=True,
                    metadata=metadata_dict,  # Inline metadata
                )
        # Check for closing
        elif self._closing_pattern.match(line):
            return DetectionResult(is_closing=True)

        return DetectionResult()

    def should_accumulate_metadata(self, candidate: BlockCandidate) -> bool:
        """No separate metadata section for this syntax."""
        return False

    def extract_block_type(self, candidate: BlockCandidate) -> str | None:
        """Extract block_type from opening line."""
        if not candidate.lines:
            return None

        # Parse the opening line to get block_type
        detection = self.detect_line(candidate.lines[0], None)
        if detection.metadata and "block_type" in detection.metadata:
            return str(detection.metadata["block_type"])

        return None

    def parse_block(
        self, candidate: BlockCandidate, block_class: type[Any] | None = None
    ) -> ParseResult[BaseMetadata, BaseContent]:
        """Parse the complete block using the specified block class."""

        # Extract metadata and content classes from block_class
        if block_class is None:
            # Default to base classes
            metadata_class = BaseMetadata
            content_class = BaseContent
        else:
            # Extract from block class using type parameters
            metadata_class, content_class = extract_block_types(block_class)

        # Metadata was already extracted during detection
        detection = self.detect_line(candidate.lines[0], None)

        if not detection.metadata:
            return ParseResult(success=False, error="Missing metadata in preamble")

        # Convert all metadata values to strings for type safety
        # Note: id and block_type are always set by detect_line()
        typed_metadata = {k: str(v) for k, v in detection.metadata.items()}

        # Parse metadata using helper
        metadata = self._safe_parse_metadata(metadata_class, typed_metadata)
        if isinstance(metadata, ParseResult):
            return metadata  # Return error

        # Parse content (skip first and last lines)
        content_text = "\n".join(candidate.lines[1:-1])

        # Parse content using helper
        content = self._safe_parse_content(content_class, content_text)
        if isinstance(content, ParseResult):
            return content  # Return error

        return ParseResult(success=True, metadata=metadata, content=content)

    def validate_block(self, _block: ExtractedBlock[BaseMetadata, BaseContent]) -> bool:
        """Additional validation after parsing."""
        return True

    def parse_metadata_early(self, candidate: BlockCandidate) -> dict[str, Any] | None:
        """Parse metadata from inline preamble.

        For this syntax, metadata is extracted from the opening line
        (e.g., !!id:type:param1:param2).
        """
        if not candidate.lines:
            return None

        detection = self.detect_line(candidate.lines[0], None)
        if detection.metadata:
            # Convert all values to strings for consistency
            return {k: str(v) for k, v in detection.metadata.items()}
        return None

    def parse_content_early(self, candidate: BlockCandidate) -> dict[str, Any] | None:
        """Parse content section early.

        Returns raw content dict with the content text.
        """
        if len(candidate.lines) < _MIN_BLOCK_LINES:
            return None

        # Content is all lines except first (header) and last (closing)
        content_lines = candidate.lines[1:-1] if len(candidate.lines) > _MIN_BLOCK_LINES else candidate.lines[1:]
        raw_content = "\n".join(content_lines)
        return {"raw_content": raw_content}

    def serialize_block(self, block: Block[BaseMetadata, BaseContent]) -> str:
        """Serialize a block to delimiter preamble format.

        Produces the round-trip form of this syntax::

            !!<id>:<type>[:<param_0>:<param_1>...]
            <raw content>
            !!end

        Args:
            block: Block instance to serialize

        Returns:
            The block rendered in delimiter preamble format
        """
        metadata = block.metadata
        opening = f"{self.delimiter}{metadata.id}:{metadata.block_type}"

        # Re-emit positional params (param_0, param_1, ...) if present
        params: list[str] = []
        index = 0
        while hasattr(metadata, f"param_{index}"):
            params.append(str(getattr(metadata, f"param_{index}")))
            index += 1
        if params:
            opening += ":" + ":".join(params)

        return f"{opening}\n{block.content.raw_content}\n{self.delimiter}end"

    def describe_format(self) -> str:
        """Describe the delimiter preamble format for prompts."""
        return (
            "Delimiter Preamble Syntax\n\n"
            "Format:\n"
            f"{self.delimiter}<id>:<type>[:<param1>:<param2>...]\n"
            "content lines\n"
            f"{self.delimiter}end\n\n"
            "Components:\n"
            f"- Opening line: {self.delimiter}<id>:<type> where id is unique and type identifies the block\n"
            "- Optional parameters: additional colon-separated values after type\n"
            "- Content: any text between the opening and closing markers\n"
            f"- Closing line: {self.delimiter}end"
        )

delimiter instance-attribute 

delimiter = delimiter

describe_format 

describe_format() -> str

Describe the delimiter preamble format for prompts.

Source code in src/hother/streamblocks/syntaxes/delimiter.py 
231
232
233
234
235
236
237
238
239
240
241
242
243
244
def describe_format(self) -> str:
    """Describe the delimiter preamble format for prompts."""
    return (
        "Delimiter Preamble Syntax\n\n"
        "Format:\n"
        f"{self.delimiter}<id>:<type>[:<param1>:<param2>...]\n"
        "content lines\n"
        f"{self.delimiter}end\n\n"
        "Components:\n"
        f"- Opening line: {self.delimiter}<id>:<type> where id is unique and type identifies the block\n"
        "- Optional parameters: additional colon-separated values after type\n"
        "- Content: any text between the opening and closing markers\n"
        f"- Closing line: {self.delimiter}end"
    )

detect_line 

detect_line(
    line: str, candidate: BlockCandidate | None = None
) -> DetectionResult

Detect delimiter-based markers.

Source code in src/hother/streamblocks/syntaxes/delimiter.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def detect_line(self, line: str, candidate: BlockCandidate | None = None) -> DetectionResult:
    """Detect delimiter-based markers."""
    if candidate is None:
        # Looking for opening
        match = self._opening_pattern.match(line)
        if match:
            block_id, block_type, params = match.groups()
            metadata_dict: dict[str, object] = {
                "id": block_id,
                "block_type": block_type,
            }

            if params:
                param_parts = params[1:].split(":")
                for i, part in enumerate(param_parts):
                    metadata_dict[f"param_{i}"] = part

            return DetectionResult(
                is_opening=True,
                metadata=metadata_dict,  # Inline metadata
            )
    # Check for closing
    elif self._closing_pattern.match(line):
        return DetectionResult(is_closing=True)

    return DetectionResult()

extract_block_type 

extract_block_type(candidate: BlockCandidate) -> str | None

Extract block_type from opening line.

Source code in src/hother/streamblocks/syntaxes/delimiter.py 
119
120
121
122
123
124
125
126
127
128
129
def extract_block_type(self, candidate: BlockCandidate) -> str | None:
    """Extract block_type from opening line."""
    if not candidate.lines:
        return None

    # Parse the opening line to get block_type
    detection = self.detect_line(candidate.lines[0], None)
    if detection.metadata and "block_type" in detection.metadata:
        return str(detection.metadata["block_type"])

    return None

parse_block 

parse_block(
    candidate: BlockCandidate,
    block_class: type[Any] | None = None,
) -> ParseResult[BaseMetadata, BaseContent]

Parse the complete block using the specified block class.

Source code in src/hother/streamblocks/syntaxes/delimiter.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def parse_block(
    self, candidate: BlockCandidate, block_class: type[Any] | None = None
) -> ParseResult[BaseMetadata, BaseContent]:
    """Parse the complete block using the specified block class."""

    # Extract metadata and content classes from block_class
    if block_class is None:
        # Default to base classes
        metadata_class = BaseMetadata
        content_class = BaseContent
    else:
        # Extract from block class using type parameters
        metadata_class, content_class = extract_block_types(block_class)

    # Metadata was already extracted during detection
    detection = self.detect_line(candidate.lines[0], None)

    if not detection.metadata:
        return ParseResult(success=False, error="Missing metadata in preamble")

    # Convert all metadata values to strings for type safety
    # Note: id and block_type are always set by detect_line()
    typed_metadata = {k: str(v) for k, v in detection.metadata.items()}

    # Parse metadata using helper
    metadata = self._safe_parse_metadata(metadata_class, typed_metadata)
    if isinstance(metadata, ParseResult):
        return metadata  # Return error

    # Parse content (skip first and last lines)
    content_text = "\n".join(candidate.lines[1:-1])

    # Parse content using helper
    content = self._safe_parse_content(content_class, content_text)
    if isinstance(content, ParseResult):
        return content  # Return error

    return ParseResult(success=True, metadata=metadata, content=content)

parse_content_early 

parse_content_early(
    candidate: BlockCandidate,
) -> dict[str, Any] | None

Parse content section early.

Returns raw content dict with the content text.

Source code in src/hother/streamblocks/syntaxes/delimiter.py 
189
190
191
192
193
194
195
196
197
198
199
200
def parse_content_early(self, candidate: BlockCandidate) -> dict[str, Any] | None:
    """Parse content section early.

    Returns raw content dict with the content text.
    """
    if len(candidate.lines) < _MIN_BLOCK_LINES:
        return None

    # Content is all lines except first (header) and last (closing)
    content_lines = candidate.lines[1:-1] if len(candidate.lines) > _MIN_BLOCK_LINES else candidate.lines[1:]
    raw_content = "\n".join(content_lines)
    return {"raw_content": raw_content}

parse_metadata_early 

parse_metadata_early(
    candidate: BlockCandidate,
) -> dict[str, Any] | None

Parse metadata from inline preamble.

For this syntax, metadata is extracted from the opening line (e.g., !!id:type:param1:param2).

Source code in src/hother/streamblocks/syntaxes/delimiter.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def parse_metadata_early(self, candidate: BlockCandidate) -> dict[str, Any] | None:
    """Parse metadata from inline preamble.

    For this syntax, metadata is extracted from the opening line
    (e.g., !!id:type:param1:param2).
    """
    if not candidate.lines:
        return None

    detection = self.detect_line(candidate.lines[0], None)
    if detection.metadata:
        # Convert all values to strings for consistency
        return {k: str(v) for k, v in detection.metadata.items()}
    return None

serialize_block 

serialize_block(
    block: Block[BaseMetadata, BaseContent],
) -> str

Serialize a block to delimiter preamble format.

Produces the round-trip form of this syntax::

!!<id>:<type>[:<param_0>:<param_1>...]
<raw content>
!!end

Parameters:

Name Type Description Default
block Block[BaseMetadata, BaseContent]

Block instance to serialize

 required

Returns:

Type Description
str

The block rendered in delimiter preamble format
Source code in src/hother/streamblocks/syntaxes/delimiter.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
def serialize_block(self, block: Block[BaseMetadata, BaseContent]) -> str:
    """Serialize a block to delimiter preamble format.

    Produces the round-trip form of this syntax::

        !!<id>:<type>[:<param_0>:<param_1>...]
        <raw content>
        !!end

    Args:
        block: Block instance to serialize

    Returns:
        The block rendered in delimiter preamble format
    """
    metadata = block.metadata
    opening = f"{self.delimiter}{metadata.id}:{metadata.block_type}"

    # Re-emit positional params (param_0, param_1, ...) if present
    params: list[str] = []
    index = 0
    while hasattr(metadata, f"param_{index}"):
        params.append(str(getattr(metadata, f"param_{index}")))
        index += 1
    if params:
        opening += ":" + ":".join(params)

    return f"{opening}\n{block.content.raw_content}\n{self.delimiter}end"

should_accumulate_metadata 

should_accumulate_metadata(
    candidate: BlockCandidate,
) -> bool

No separate metadata section for this syntax.

Source code in src/hother/streamblocks/syntaxes/delimiter.py 
115
116
117
def should_accumulate_metadata(self, candidate: BlockCandidate) -> bool:
    """No separate metadata section for this syntax."""
    return False

validate_block 

validate_block(
    _block: ExtractedBlock[BaseMetadata, BaseContent],
) -> bool

Additional validation after parsing.

Source code in src/hother/streamblocks/syntaxes/delimiter.py 
170
171
172
def validate_block(self, _block: ExtractedBlock[BaseMetadata, BaseContent]) -> bool:
    """Additional validation after parsing."""
    return True

hother.streamblocks.syntaxes.markdown

Markdown-based syntax implementations.

MarkdownFrontmatterSyntax

Bases: BaseSyntax, YAMLFrontmatterMixin

This syntax uses Markdown-style fenced code blocks with optional YAML frontmatter for metadata. The info_string after the opening fence can be used as a fallback block_type when no frontmatter is present.

Format
```[info_string]

id: block_001 block_type: example custom_field: value

Content lines here ```

The info_string is optional. When provided, it's used as the block_type if no YAML frontmatter is present. The YAML frontmatter is also optional - if omitted, all content becomes the block content.

Examples:

>>> # Block with frontmatter
>>> '''
... ```python
... ---
... id: code001
... block_type: code
... language: python
... ---
... def hello():
...     print("Hello, world!")
... ```
... '''
>>>
>>> # Block without frontmatter (info_string becomes block_type)
>>> '''
... ```patch
... diff --git a/file.py b/file.py
... - old line
... + new line
... ```
... '''
>>> # block_type will be "patch" from info_string
>>>
>>> # Block with nested YAML
>>> '''
... ```task
... ---
... id: task001
... block_type: task
... assignees:
...   - alice
...   - bob
... ---
... Implement user authentication
... ```
... '''

Parameters:

Name Type Description Default
fence str

Fence string (default: "```")

 '```'
info_string str | None

Optional info string used as fallback block_type

 None
Source code in src/hother/streamblocks/syntaxes/markdown.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
class MarkdownFrontmatterSyntax(BaseSyntax, YAMLFrontmatterMixin):
    """Syntax: Markdown fenced code blocks with YAML frontmatter.

    This syntax uses Markdown-style fenced code blocks with optional YAML frontmatter
    for metadata. The info_string after the opening fence can be used as a fallback
    block_type when no frontmatter is present.

    Format:
        ```[info_string]
        ---
        id: block_001
        block_type: example
        custom_field: value
        ---
        Content lines here
        ```

    The info_string is optional. When provided, it's used as the block_type if
    no YAML frontmatter is present. The YAML frontmatter is also optional - if
    omitted, all content becomes the block content.

    Examples:
        >>> # Block with frontmatter
        >>> '''
        ... ```python
        ... ---
        ... id: code001
        ... block_type: code
        ... language: python
        ... ---
        ... def hello():
        ...     print("Hello, world!")
        ... ```
        ... '''
        >>>
        >>> # Block without frontmatter (info_string becomes block_type)
        >>> '''
        ... ```patch
        ... diff --git a/file.py b/file.py
        ... - old line
        ... + new line
        ... ```
        ... '''
        >>> # block_type will be "patch" from info_string
        >>>
        >>> # Block with nested YAML
        >>> '''
        ... ```task
        ... ---
        ... id: task001
        ... block_type: task
        ... assignees:
        ...   - alice
        ...   - bob
        ... ---
        ... Implement user authentication
        ... ```
        ... '''

    Args:
        fence: Fence string (default: "```")
        info_string: Optional info string used as fallback block_type
    """

    def __init__(
        self,
        fence: str = "```",
        info_string: str | None = None,
    ) -> None:
        """Initialize markdown frontmatter syntax.

        Args:
            fence: Fence string (e.g., "```")
            info_string: Optional info string after fence
        """
        self.fence = fence
        self.info_string = info_string
        self._fence_pattern = self._build_fence_pattern()
        self._frontmatter_pattern = re.compile(r"^---\s*$")

    def _build_fence_pattern(self) -> re.Pattern[str]:
        """Build pattern for fence detection."""
        pattern_str = rf"^{re.escape(self.fence)}"
        if self.info_string:
            pattern_str += re.escape(self.info_string)
        return re.compile(pattern_str)

    def detect_line(self, line: str, candidate: BlockCandidate | None = None) -> DetectionResult:
        """Detect markdown fence markers and frontmatter boundaries."""
        if candidate is None:
            # Looking for opening fence
            if self._fence_pattern.match(line):
                return DetectionResult(is_opening=True)
        # Inside a block
        elif candidate.current_section == SectionType.HEADER:
            # Check if this is frontmatter start
            if self._frontmatter_pattern.match(line):
                candidate.transition_to_metadata()
                return DetectionResult(is_metadata_boundary=True)
            # Skip empty lines in header - frontmatter might follow
            if line.strip() == "":
                return DetectionResult()
            # Non-empty, non-frontmatter line - move to content
            candidate.transition_to_content()
            candidate.content_lines.append(line)
        elif candidate.current_section == SectionType.METADATA:
            # Check for metadata end
            if self._frontmatter_pattern.match(line):
                candidate.transition_to_content()
                return DetectionResult(is_metadata_boundary=True)
            candidate.metadata_lines.append(line)
        elif candidate.current_section == SectionType.CONTENT:
            # Check for closing fence
            if line.strip() == self.fence:
                return DetectionResult(is_closing=True)
            candidate.content_lines.append(line)

        return DetectionResult()

    def should_accumulate_metadata(self, candidate: BlockCandidate) -> bool:
        """Check if we're still in metadata section."""
        return candidate.current_section in {SectionType.HEADER, SectionType.METADATA}

    def extract_block_type(self, candidate: BlockCandidate) -> str | None:
        """Extract block_type from YAML frontmatter."""
        if not candidate.metadata_lines:
            # Try to infer from info string
            return self.info_string

        # Parse YAML to extract block_type
        metadata_dict = self._parse_yaml_metadata(candidate.metadata_lines)
        if metadata_dict and "block_type" in metadata_dict:
            return str(metadata_dict["block_type"])
        # No block_type found in metadata or parse failed, return info_string
        return self.info_string

    def _parse_metadata_instance(
        self,
        metadata_class: type[BaseMetadata],
        metadata_dict: dict[str, Any],
    ) -> ParseResult[BaseMetadata, BaseContent] | BaseMetadata:
        """Parse and validate metadata dict into metadata instance."""
        return self._safe_parse_metadata(metadata_class, metadata_dict)

    def _parse_content_instance(
        self,
        content_class: type[BaseContent],
        candidate: BlockCandidate,
    ) -> ParseResult[BaseMetadata, BaseContent] | BaseContent:
        """Parse content lines into content instance."""
        content_text = "\n".join(candidate.content_lines)
        return self._safe_parse_content(content_class, content_text)

    def parse_block(
        self, candidate: BlockCandidate, block_class: type[Any] | None = None
    ) -> ParseResult[BaseMetadata, BaseContent]:
        """Parse the complete block using the specified block class."""

        # Extract metadata and content classes
        if block_class is None:
            metadata_class = BaseMetadata
            content_class = BaseContent
        else:
            metadata_class, content_class = extract_block_types(block_class)

        # Parse YAML metadata
        metadata_dict, yaml_error = self._parse_yaml_metadata_strict(candidate.metadata_lines)
        if yaml_error:
            return ParseResult(success=False, error=f"YAML parse error: {yaml_error}", exception=yaml_error)

        # Parse metadata instance
        metadata_result = self._parse_metadata_instance(metadata_class, metadata_dict)
        if isinstance(metadata_result, ParseResult):
            return metadata_result

        # Parse content instance
        content_result = self._parse_content_instance(content_class, candidate)
        if isinstance(content_result, ParseResult):
            return content_result

        return ParseResult(success=True, metadata=metadata_result, content=content_result)

    def validate_block(self, _block: ExtractedBlock[BaseMetadata, BaseContent]) -> bool:
        """Additional validation after parsing."""
        return True

    def serialize_block(self, block: Block[BaseMetadata, BaseContent]) -> str:
        """Serialize a block to markdown frontmatter format.

        Produces the round-trip form of this syntax::

            ```[info_string]
            ---
            <yaml metadata>
            ---
            <raw content>
            ```

        Args:
            block: Block instance to serialize

        Returns:
            The block rendered in markdown frontmatter format
        """
        metadata_yaml = yaml.dump(block.metadata.model_dump(), default_flow_style=False, sort_keys=False)
        opening = f"{self.fence}{self.info_string}" if self.info_string else self.fence
        return f"{opening}\n---\n{metadata_yaml}---\n{block.content.raw_content}\n{self.fence}"

    def describe_format(self) -> str:
        """Describe the markdown frontmatter format for prompts."""
        info_str_desc = f"{self.info_string}" if self.info_string else "[info_string]"
        return (
            "Markdown Frontmatter Syntax\n\n"
            "Format:\n"
            f"{self.fence}{info_str_desc}\n"
            "---\n"
            "key: value\n"
            "---\n"
            "content lines\n"
            f"{self.fence}\n\n"
            "Components:\n"
            f"- Opening fence: {self.fence} with optional info string\n"
            "- Metadata section: YAML frontmatter between --- delimiters\n"
            "- Content: any text after the metadata section\n"
            f"- Closing fence: {self.fence}"
        )

fence instance-attribute 

fence = fence

info_string instance-attribute 

info_string = info_string

describe_format 

describe_format() -> str

Describe the markdown frontmatter format for prompts.

Source code in src/hother/streamblocks/syntaxes/markdown.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
def describe_format(self) -> str:
    """Describe the markdown frontmatter format for prompts."""
    info_str_desc = f"{self.info_string}" if self.info_string else "[info_string]"
    return (
        "Markdown Frontmatter Syntax\n\n"
        "Format:\n"
        f"{self.fence}{info_str_desc}\n"
        "---\n"
        "key: value\n"
        "---\n"
        "content lines\n"
        f"{self.fence}\n\n"
        "Components:\n"
        f"- Opening fence: {self.fence} with optional info string\n"
        "- Metadata section: YAML frontmatter between --- delimiters\n"
        "- Content: any text after the metadata section\n"
        f"- Closing fence: {self.fence}"
    )

detect_line 

detect_line(
    line: str, candidate: BlockCandidate | None = None
) -> DetectionResult

Detect markdown fence markers and frontmatter boundaries.

Source code in src/hother/streamblocks/syntaxes/markdown.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def detect_line(self, line: str, candidate: BlockCandidate | None = None) -> DetectionResult:
    """Detect markdown fence markers and frontmatter boundaries."""
    if candidate is None:
        # Looking for opening fence
        if self._fence_pattern.match(line):
            return DetectionResult(is_opening=True)
    # Inside a block
    elif candidate.current_section == SectionType.HEADER:
        # Check if this is frontmatter start
        if self._frontmatter_pattern.match(line):
            candidate.transition_to_metadata()
            return DetectionResult(is_metadata_boundary=True)
        # Skip empty lines in header - frontmatter might follow
        if line.strip() == "":
            return DetectionResult()
        # Non-empty, non-frontmatter line - move to content
        candidate.transition_to_content()
        candidate.content_lines.append(line)
    elif candidate.current_section == SectionType.METADATA:
        # Check for metadata end
        if self._frontmatter_pattern.match(line):
            candidate.transition_to_content()
            return DetectionResult(is_metadata_boundary=True)
        candidate.metadata_lines.append(line)
    elif candidate.current_section == SectionType.CONTENT:
        # Check for closing fence
        if line.strip() == self.fence:
            return DetectionResult(is_closing=True)
        candidate.content_lines.append(line)

    return DetectionResult()

extract_block_type 

extract_block_type(candidate: BlockCandidate) -> str | None

Extract block_type from YAML frontmatter.

Source code in src/hother/streamblocks/syntaxes/markdown.py 
141
142
143
144
145
146
147
148
149
150
151
152
def extract_block_type(self, candidate: BlockCandidate) -> str | None:
    """Extract block_type from YAML frontmatter."""
    if not candidate.metadata_lines:
        # Try to infer from info string
        return self.info_string

    # Parse YAML to extract block_type
    metadata_dict = self._parse_yaml_metadata(candidate.metadata_lines)
    if metadata_dict and "block_type" in metadata_dict:
        return str(metadata_dict["block_type"])
    # No block_type found in metadata or parse failed, return info_string
    return self.info_string

parse_block 

parse_block(
    candidate: BlockCandidate,
    block_class: type[Any] | None = None,
) -> ParseResult[BaseMetadata, BaseContent]

Parse the complete block using the specified block class.

Source code in src/hother/streamblocks/syntaxes/markdown.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
def parse_block(
    self, candidate: BlockCandidate, block_class: type[Any] | None = None
) -> ParseResult[BaseMetadata, BaseContent]:
    """Parse the complete block using the specified block class."""

    # Extract metadata and content classes
    if block_class is None:
        metadata_class = BaseMetadata
        content_class = BaseContent
    else:
        metadata_class, content_class = extract_block_types(block_class)

    # Parse YAML metadata
    metadata_dict, yaml_error = self._parse_yaml_metadata_strict(candidate.metadata_lines)
    if yaml_error:
        return ParseResult(success=False, error=f"YAML parse error: {yaml_error}", exception=yaml_error)

    # Parse metadata instance
    metadata_result = self._parse_metadata_instance(metadata_class, metadata_dict)
    if isinstance(metadata_result, ParseResult):
        return metadata_result

    # Parse content instance
    content_result = self._parse_content_instance(content_class, candidate)
    if isinstance(content_result, ParseResult):
        return content_result

    return ParseResult(success=True, metadata=metadata_result, content=content_result)

parse_content_early 

parse_content_early(
    candidate: BlockCandidate,
) -> dict[str, Any] | None

Parse content section early, before final block extraction.

This method is called when the content section completes (block closes), allowing early validation. The result can be cached in the candidate for reuse during full block parsing.

Default implementation returns None (no early parsing). Override in subclasses to provide early content parsing.

Parameters:

Name Type Description Default
candidate BlockCandidate

The complete block candidate with content accumulated

 required

Returns:

Type Description
dict[str, Any] | None

Parsed content dict if successful, None if parsing not supported
dict[str, Any] | None

or failed
Source code in src/hother/streamblocks/syntaxes/base.py 
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
def parse_content_early(self, candidate: BlockCandidate) -> dict[str, Any] | None:
    """Parse content section early, before final block extraction.

    This method is called when the content section completes (block closes),
    allowing early validation. The result can be cached in the candidate
    for reuse during full block parsing.

    Default implementation returns None (no early parsing).
    Override in subclasses to provide early content parsing.

    Args:
        candidate: The complete block candidate with content accumulated

    Returns:
        Parsed content dict if successful, None if parsing not supported
        or failed
    """
    return None

parse_metadata_early 

parse_metadata_early(
    candidate: BlockCandidate,
) -> dict[str, Any] | None

Parse metadata section early, before content accumulation.

This method is called when the metadata section completes, allowing early validation and processing. The result can be cached in the candidate for reuse during full block parsing.

Default implementation returns None (no early parsing). Override in subclasses to provide early metadata parsing.

Parameters:

Name Type Description Default
candidate BlockCandidate

The current block candidate with metadata accumulated

 required

Returns:

Type Description
dict[str, Any] | None

Parsed metadata dict if successful, None if parsing not supported
dict[str, Any] | None

or failed
Source code in src/hother/streamblocks/syntaxes/base.py 
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
def parse_metadata_early(self, candidate: BlockCandidate) -> dict[str, Any] | None:
    """Parse metadata section early, before content accumulation.

    This method is called when the metadata section completes, allowing
    early validation and processing. The result can be cached in the
    candidate for reuse during full block parsing.

    Default implementation returns None (no early parsing).
    Override in subclasses to provide early metadata parsing.

    Args:
        candidate: The current block candidate with metadata accumulated

    Returns:
        Parsed metadata dict if successful, None if parsing not supported
        or failed
    """
    return None

serialize_block 

serialize_block(
    block: Block[BaseMetadata, BaseContent],
) -> str

Serialize a block to markdown frontmatter format.

Produces the round-trip form of this syntax::

```[info_string]
---
<yaml metadata>
---
<raw content>
```

Parameters:

Name Type Description Default
block Block[BaseMetadata, BaseContent]

Block instance to serialize

 required

Returns:

Type Description
str

The block rendered in markdown frontmatter format
Source code in src/hother/streamblocks/syntaxes/markdown.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
def serialize_block(self, block: Block[BaseMetadata, BaseContent]) -> str:
    """Serialize a block to markdown frontmatter format.

    Produces the round-trip form of this syntax::

        ```[info_string]
        ---
        <yaml metadata>
        ---
        <raw content>
        ```

    Args:
        block: Block instance to serialize

    Returns:
        The block rendered in markdown frontmatter format
    """
    metadata_yaml = yaml.dump(block.metadata.model_dump(), default_flow_style=False, sort_keys=False)
    opening = f"{self.fence}{self.info_string}" if self.info_string else self.fence
    return f"{opening}\n---\n{metadata_yaml}---\n{block.content.raw_content}\n{self.fence}"

should_accumulate_metadata 

should_accumulate_metadata(
    candidate: BlockCandidate,
) -> bool

Check if we're still in metadata section.

Source code in src/hother/streamblocks/syntaxes/markdown.py 
137
138
139
def should_accumulate_metadata(self, candidate: BlockCandidate) -> bool:
    """Check if we're still in metadata section."""
    return candidate.current_section in {SectionType.HEADER, SectionType.METADATA}

validate_block 

validate_block(
    _block: ExtractedBlock[BaseMetadata, BaseContent],
) -> bool

Additional validation after parsing.

Source code in src/hother/streamblocks/syntaxes/markdown.py 
200
201
202
def validate_block(self, _block: ExtractedBlock[BaseMetadata, BaseContent]) -> bool:
    """Additional validation after parsing."""
    return True

hother.streamblocks.syntaxes.factory

Factory function for creating syntax instances.

Syntax

Bases: StrEnum

Enum of built-in syntax types.

Source code in src/hother/streamblocks/syntaxes/factory.py 
17
18
19
20
21
22
class Syntax(StrEnum):
    """Enum of built-in syntax types."""

    DELIMITER_FRONTMATTER = auto()
    DELIMITER_PREAMBLE = auto()
    MARKDOWN_FRONTMATTER = auto()

DELIMITER_FRONTMATTER class-attribute instance-attribute 

DELIMITER_FRONTMATTER = auto()

DELIMITER_PREAMBLE class-attribute instance-attribute 

DELIMITER_PREAMBLE = auto()

MARKDOWN_FRONTMATTER class-attribute instance-attribute 

MARKDOWN_FRONTMATTER = auto()

get_syntax_instance 

get_syntax_instance(
    syntax: Syntax | BaseSyntax,
) -> BaseSyntax

Get a syntax instance from a Syntax enum or return custom instance.

This helper function allows users to specify built-in syntaxes using the Syntax enum or provide their own custom syntax implementations.

Parameters:

Name Type Description Default
syntax Syntax | BaseSyntax

Either a Syntax enum member or a custom BaseSyntax instance

 required

Returns:

Type Description
BaseSyntax

A syntax instance inheriting from BaseSyntax

Raises:

Type Description
SyntaxConfigError

If syntax is neither a Syntax enum nor a BaseSyntax instance
Example

Using built-in syntax

syntax = get_syntax_instance(Syntax.DELIMITER_PREAMBLE)

Using custom syntax

my_syntax = MySyntax() syntax = get_syntax_instance(my_syntax)

Source code in src/hother/streamblocks/syntaxes/factory.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def get_syntax_instance(
    syntax: Syntax | BaseSyntax,
) -> BaseSyntax:
    """Get a syntax instance from a Syntax enum or return custom instance.

    This helper function allows users to specify built-in syntaxes using
    the Syntax enum or provide their own custom syntax implementations.

    Args:
        syntax: Either a Syntax enum member or a custom BaseSyntax instance

    Returns:
        A syntax instance inheriting from BaseSyntax

    Raises:
        SyntaxConfigError: If syntax is neither a Syntax enum nor a BaseSyntax instance

    Example:
        >>> # Using built-in syntax
        >>> syntax = get_syntax_instance(Syntax.DELIMITER_PREAMBLE)
        >>>
        >>> # Using custom syntax
        >>> my_syntax = MySyntax()
        >>> syntax = get_syntax_instance(my_syntax)
    """
    if isinstance(syntax, Syntax):
        match syntax:
            case Syntax.DELIMITER_FRONTMATTER:
                return DelimiterFrontmatterSyntax()
            case Syntax.DELIMITER_PREAMBLE:
                return DelimiterPreambleSyntax()
            case Syntax.MARKDOWN_FRONTMATTER:
                return MarkdownFrontmatterSyntax()

    unchecked = cast("object", syntax)
    if isinstance(unchecked, BaseSyntax):
        return unchecked
    raise SyntaxConfigError(received_type=type(unchecked).__name__)

hother.streamblocks.syntaxes.models

Base syntax class and utilities for StreamBlocks syntax implementations.

Syntax

Bases: StrEnum

Enum of built-in syntax types.

Source code in src/hother/streamblocks/syntaxes/factory.py 
17
18
19
20
21
22
class Syntax(StrEnum):
    """Enum of built-in syntax types."""

    DELIMITER_FRONTMATTER = auto()
    DELIMITER_PREAMBLE = auto()
    MARKDOWN_FRONTMATTER = auto()

DELIMITER_FRONTMATTER class-attribute instance-attribute 

DELIMITER_FRONTMATTER = auto()

DELIMITER_PREAMBLE class-attribute instance-attribute 

DELIMITER_PREAMBLE = auto()

MARKDOWN_FRONTMATTER class-attribute instance-attribute 

MARKDOWN_FRONTMATTER = auto()

get_syntax_instance 

get_syntax_instance(
    syntax: Syntax | BaseSyntax,
) -> BaseSyntax

Get a syntax instance from a Syntax enum or return custom instance.

This helper function allows users to specify built-in syntaxes using the Syntax enum or provide their own custom syntax implementations.

Parameters:

Name Type Description Default
syntax Syntax | BaseSyntax

Either a Syntax enum member or a custom BaseSyntax instance

 required

Returns:

Type Description
BaseSyntax

A syntax instance inheriting from BaseSyntax

Raises:

Type Description
SyntaxConfigError

If syntax is neither a Syntax enum nor a BaseSyntax instance
Example

Using built-in syntax

syntax = get_syntax_instance(Syntax.DELIMITER_PREAMBLE)

Using custom syntax

my_syntax = MySyntax() syntax = get_syntax_instance(my_syntax)

Source code in src/hother/streamblocks/syntaxes/factory.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def get_syntax_instance(
    syntax: Syntax | BaseSyntax,
) -> BaseSyntax:
    """Get a syntax instance from a Syntax enum or return custom instance.

    This helper function allows users to specify built-in syntaxes using
    the Syntax enum or provide their own custom syntax implementations.

    Args:
        syntax: Either a Syntax enum member or a custom BaseSyntax instance

    Returns:
        A syntax instance inheriting from BaseSyntax

    Raises:
        SyntaxConfigError: If syntax is neither a Syntax enum nor a BaseSyntax instance

    Example:
        >>> # Using built-in syntax
        >>> syntax = get_syntax_instance(Syntax.DELIMITER_PREAMBLE)
        >>>
        >>> # Using custom syntax
        >>> my_syntax = MySyntax()
        >>> syntax = get_syntax_instance(my_syntax)
    """
    if isinstance(syntax, Syntax):
        match syntax:
            case Syntax.DELIMITER_FRONTMATTER:
                return DelimiterFrontmatterSyntax()
            case Syntax.DELIMITER_PREAMBLE:
                return DelimiterPreambleSyntax()
            case Syntax.MARKDOWN_FRONTMATTER:
                return MarkdownFrontmatterSyntax()

    unchecked = cast("object", syntax)
    if isinstance(unchecked, BaseSyntax):
        return unchecked
    raise SyntaxConfigError(received_type=type(unchecked).__name__)