Syntaxes Examples
Examples covering the built-in block syntax formats and how to define your own. See the Syntaxes concept page for background. All run offline with no API keys.
Markdown Frontmatter
Extracts blocks written as Markdown code fences with YAML frontmatter using MarkdownFrontmatterSyntax, a natural fit for LLMs already fluent in Markdown.
src/hother/streamblocks_examples/02_syntaxes/01_markdown_frontmatter.py
View source on GitHub
import asyncio
from textwrap import dedent
from typing import TYPE_CHECKING
from hother.streamblocks import MarkdownFrontmatterSyntax, Registry, StreamBlockProcessor
from hother.streamblocks.core.processor import ProcessorConfig
from hother.streamblocks.core.types import (
BlockContentDeltaEvent,
BlockEndEvent,
BlockErrorEvent,
BlockHeaderDeltaEvent,
BlockMetadataDeltaEvent,
TextContentEvent,
)
from hother.streamblocks_examples.blocks.agent.patch import Patch
from hother.streamblocks_examples.helpers.simulator import simulated_stream
if TYPE_CHECKING:
from hother.streamblocks.core.models import ExtractedBlock
from hother.streamblocks.core.types import BaseContent, BaseMetadata
async def main() -> None:
"""Main example function."""
# Create markdown frontmatter syntax for patch blocks
# Each Registry holds exactly one syntax.
# To handle multiple info strings (patch/yaml/diff), you would need separate processors
# or a custom syntax that handles multiple patterns internally.
syntax = MarkdownFrontmatterSyntax(
fence="```",
info_string="patch", # Will match ```patch blocks
)
# Create type-specific registry and register block
registry = Registry(syntax=syntax)
registry.register("patch", Patch)
# Create processor with config
config = ProcessorConfig(lines_buffer=10)
processor = StreamBlockProcessor(registry, config=config)
# Example text with markdown frontmatter blocks
text = dedent("""
Here's a document with some patches using markdown-style blocks with YAML frontmatter.
```patch
---
id: security-fix
block_type: patch
file: auth.py
start_line: 45
---
def authenticate(user, password):
- if password == "admin": # pragma: allowlist secret
+ if check_password_hash(user.password_hash, password):
return True
return False
```
Now let's add another patch for the config file:
```patch
---
id: config-update
block_type: patch
file: config.yaml
start_line: 10
---
database:
host: localhost
- port: 3306
+ port: 5432
- driver: mysql
+ driver: postgresql
```
And here's a final patch with more metadata:
```patch
---
id: feature-add
block_type: patch
file: features.py
start_line: 100
author: dev-team
priority: high
---
+def new_feature():
+ \"\"\"Implement awesome new feature.\"\"\"
+ return \"awesome\"
+
class ExistingClass:
pass
```
That's all for the patches!
""")
# Process stream
print("Processing markdown frontmatter blocks...")
print("-" * 70)
blocks_extracted: list[ExtractedBlock[BaseMetadata, BaseContent]] = []
current_partial = None
async for event in processor.process_stream(simulated_stream(text)):
if isinstance(event, TextContentEvent):
# Raw text passed through
if event.content.strip():
print(f"[TEXT] {event.content.strip()}")
elif isinstance(event, (BlockHeaderDeltaEvent, BlockMetadataDeltaEvent, BlockContentDeltaEvent)):
# Track partial block updates
syntax = event.syntax
section = event.type.value.replace("BLOCK_", "").replace("_DELTA", "").lower()
if current_partial != syntax:
print(f"\n[DELTA] Started {syntax} block (section: {section})")
current_partial = syntax
elif isinstance(event, BlockEndEvent):
# Complete block extracted
block = event.get_block()
if block is None:
continue
blocks_extracted.append(block)
current_partial = None
print("\n[BLOCK] Extracted:")
print(block.model_dump_json(indent=2))
elif isinstance(event, BlockErrorEvent):
# Block rejected
reason = event.reason
syntax = event.syntax
print(f"\n[REJECT] {syntax} block rejected: {reason}")
print("-" * 70)
print(f"\nTotal blocks extracted: {len(blocks_extracted)}")
# Show all extracted blocks
print("\nExtracted blocks (full details):")
for i, block in enumerate(blocks_extracted, 1):
print(f"\n--- Block {i} ---")
print(block.model_dump_json(indent=2))
if __name__ == "__main__":
asyncio.run(main())
Delimiter Frontmatter
Uses DelimiterFrontmatterSyntax, the compact !!id:type ... !!end format with YAML frontmatter for metadata. This is the syntax used by most other examples.
src/hother/streamblocks_examples/02_syntaxes/02_delimiter_frontmatter.py
View source on GitHub
import asyncio
from textwrap import dedent
from pydantic import Field
from hother.streamblocks import DelimiterFrontmatterSyntax, Registry, StreamBlockProcessor
from hother.streamblocks.core.models import Block, ExtractedBlock
from hother.streamblocks.core.types import (
BaseContent,
BaseMetadata,
BlockContentDeltaEvent,
BlockEndEvent,
BlockErrorEvent,
BlockHeaderDeltaEvent,
BlockMetadataDeltaEvent,
TextContentEvent,
)
from hother.streamblocks_examples.helpers.simulator import simulated_stream
# Custom content models for this example
class TaskMetadata(BaseMetadata):
"""Metadata for task blocks."""
id: str
block_type: str
title: str = "Untitled Task"
priority: str = "medium"
assignee: str | None = None
due_date: str | None = None
tags: list[str] = Field(default_factory=list[str])
status: str = "todo"
class TaskContent(BaseContent):
"""Content for task blocks."""
description: str = ""
subtasks: list[str] = Field(default_factory=list[str])
@classmethod
def parse(cls, raw_text: str) -> "TaskContent":
"""Parse task content from raw text."""
lines = raw_text.strip().split("\n")
if not lines:
return cls(raw_content=raw_text, description="")
description = lines[0]
subtasks: list[str] = []
for line in lines[1:]:
stripped = line.strip()
if stripped.startswith(("- ", "* ")):
subtasks.append(stripped[2:])
return cls(raw_content=raw_text, description=description, subtasks=subtasks)
# Create the block type
TaskBlock = Block[TaskMetadata, TaskContent]
async def main() -> None:
"""Main example function."""
print("=== DelimiterFrontmatterSyntax Example ===\n")
# Create delimiter frontmatter syntax for tasks
# Using standard !!start/!!end delimiters
task_syntax = DelimiterFrontmatterSyntax(
start_delimiter="!!start",
end_delimiter="!!end",
)
# Create type-specific registry and register block
registry = Registry(syntax=task_syntax)
registry.register("task", TaskBlock)
# Add validators
def validate_task_priority(block: ExtractedBlock[TaskMetadata, TaskContent]) -> bool:
"""Ensure high priority tasks have assignees."""
return not (block.metadata.priority in ["high", "urgent"] and not block.metadata.assignee)
registry.add_validator("task", validate_task_priority)
# Create processor with config
from hother.streamblocks.core.processor import ProcessorConfig
config = ProcessorConfig(lines_buffer=10)
processor = StreamBlockProcessor(registry, config=config)
# Example text with delimiter frontmatter blocks
text = dedent("""
Let's manage some tasks using delimiter+frontmatter syntax.
!!start
---
id: task-001
block_type: task
title: Implement authentication
priority: high
assignee: alice
due_date: "2024-01-15"
tags:
- backend
- api
- urgent
status: in_progress
---
Implement user authentication API
- Create JWT token generation
- Add refresh token support
- Implement password reset flow
- Add 2FA support
!!end
Here's another task with simpler metadata:
!!start
---
id: task-002
block_type: task
title: Update documentation
assignee: bob
---
Update documentation
- API reference docs
- Installation guide
- Contributing guidelines
!!end
And a minimal task:
!!start
---
id: task-003
block_type: task
title: Fix payment bug
priority: urgent
---
Fix critical bug in payment processing
!!end
Some text between blocks.
!!start
---
id: task-004
block_type: task
title: Performance optimization
assignee: charlie
tags:
- performance
- backend
---
Optimize database queries
- Add proper indexes
- Implement query caching
- Review N+1 queries
!!end
That's all for now!
""")
# Process stream
print("Processing task blocks...\n")
blocks_extracted: list[ExtractedBlock[BaseMetadata, BaseContent]] = []
async for event in processor.process_stream(simulated_stream(text)):
if isinstance(event, TextContentEvent):
# Raw text passed through
if event.content.strip():
text = event.content.strip()
if len(text) > 60:
text = text[:57] + "..."
print(f"[TEXT] {text}")
elif isinstance(event, (BlockHeaderDeltaEvent, BlockMetadataDeltaEvent, BlockContentDeltaEvent)):
# Skip deltas for cleaner output
pass
elif isinstance(event, BlockEndEvent):
# Complete block extracted
block = event.get_block()
if block is None:
continue
blocks_extracted.append(block)
print("\n[TASK] Extracted:")
print(block.model_dump_json(indent=2))
elif isinstance(event, BlockErrorEvent):
# Block rejected
print(f"\n[REJECT] {event.reason}")
print("\n\nEXTRACTED BLOCKS SUMMARY:")
print(f"Total blocks: {len(blocks_extracted)}")
print("\nExtracted blocks (full details):")
for i, block in enumerate(blocks_extracted, 1):
print(f"\n--- Block {i} ---")
print(block.model_dump_json(indent=2))
print("\n✓ DelimiterFrontmatterSyntax processing complete!")
if __name__ == "__main__":
asyncio.run(main())
asyncio.run(main())
Parsing Decorators
Shows the @parse_as_yaml() and @parse_as_json() decorators that automatically parse block content into structured Pydantic models, including STRICT vs PERMISSIVE error-handling strategies and graceful recovery from malformed content.
src/hother/streamblocks_examples/02_syntaxes/03_parsing_decorators.py
View source on GitHub
#!/usr/bin/env python3
import asyncio
import logging
from collections.abc import AsyncIterator
from textwrap import dedent
from typing import Any
from pydantic import Field
from hother.streamblocks import (
DelimiterFrontmatterSyntax,
DelimiterPreambleSyntax,
MarkdownFrontmatterSyntax,
Registry,
StreamBlockProcessor,
)
from hother.streamblocks.core.models import Block
from hother.streamblocks.core.parsing import ParseStrategy, parse_as_json, parse_as_yaml
from hother.streamblocks.core.types import BaseContent, BaseMetadata, BlockEndEvent, BlockErrorEvent, TextContentEvent
# ============================================================================
# EXAMPLE 1: Basic YAML Parsing (Permissive Mode)
# ============================================================================
@parse_as_yaml(strategy=ParseStrategy.PERMISSIVE)
class ConfigContent(BaseContent):
"""Configuration content parsed from YAML.
With PERMISSIVE strategy, malformed YAML falls back to raw_content.
"""
app_name: str | None = None
version: str | None = None
debug: bool | None = None
port: int | None = None
features: dict[str, bool] = Field(default_factory=dict)
class ConfigMetadata(BaseMetadata):
"""Metadata for configuration blocks."""
block_type: str = "config"
environment: str | None = None
# Create the block type
ConfigBlock = Block[ConfigMetadata, ConfigContent]
async def example_1_basic_yaml_parsing() -> None:
"""Demonstrate basic YAML parsing with permissive error handling."""
print("\n" + "=" * 70)
print("EXAMPLE 1: Basic YAML Parsing (Permissive Mode)")
print("=" * 70)
# Create syntax and registry
syntax = DelimiterPreambleSyntax()
registry = Registry(syntax=syntax)
registry.register("config", ConfigBlock)
# Create processor
processor = StreamBlockProcessor(registry)
# Example stream with YAML content
async def config_stream() -> AsyncIterator[str]:
text = dedent("""
Application configuration:
!!config_prod:config
app_name: MyApp
version: 1.2.3
debug: false
port: 8080
features:
auth: true
logging: true
metrics: false
!!end
That's the production config. Now here's a malformed one:
!!config_bad:config
app_name: BrokenApp
version: [1, 2
debug: this is not valid YAML {{
!!end
Processing complete.
""")
for line in text.split("\n"):
yield line + "\n"
await asyncio.sleep(0.001)
# Process the stream
print("\nProcessing stream with YAML configs...")
async for event in processor.process_stream(config_stream()):
if isinstance(event, BlockEndEvent):
block = event.get_block()
if block is not None:
print("\n✅ Extracted Config Block:")
print(block.model_dump_json(indent=2))
elif isinstance(event, TextContentEvent):
if event.content.strip():
print(f"[TEXT] {event.content.strip()}")
# ============================================================================
# EXAMPLE 2: Strict JSON Parsing
# ============================================================================
@parse_as_json(strategy=ParseStrategy.STRICT)
class APIResponseContent(BaseContent):
"""API response parsed from JSON.
With STRICT strategy, malformed JSON raises an exception.
"""
status: int
message: str
data: dict[str, Any] = Field(default_factory=dict)
errors: list[str] = Field(default_factory=list)
class APIMetadata(BaseMetadata):
"""Metadata for API response blocks."""
block_type: str = "api_response"
endpoint: str | None = None
# Create the block type
APIBlock = Block[APIMetadata, APIResponseContent]
async def example_2_strict_json_parsing() -> None:
"""Demonstrate strict JSON parsing that raises on errors."""
print("\n" + "=" * 70)
print("EXAMPLE 2: Strict JSON Parsing")
print("=" * 70)
# Create syntax and registry
syntax = MarkdownFrontmatterSyntax(fence="```", info_string="json")
registry = Registry(syntax=syntax)
registry.register("api_response", APIBlock)
# Create processor
processor = StreamBlockProcessor(registry)
# Example stream with JSON API responses
async def api_stream() -> AsyncIterator[str]:
text = dedent("""
API responses from the server:
```json
---
id: resp_001
block_type: api_response
endpoint: /users/123
---
{
"status": 200,
"message": "User retrieved successfully",
"data": {
"user_id": 123,
"username": "alice",
"email": "alice@example.com"
}
}
```
```json
---
id: resp_002
block_type: api_response
endpoint: /posts
---
{
"status": 201,
"message": "Post created",
"data": {
"post_id": 456,
"title": "My New Post",
"created_at": "2024-12-15T10:30:00Z"
}
}
```
All responses processed.
""")
for line in text.split("\n"):
yield line + "\n"
await asyncio.sleep(0.001)
# Process the stream
print("\nProcessing API responses...")
async for event in processor.process_stream(api_stream()):
if isinstance(event, BlockEndEvent):
block = event.get_block()
if block is None:
continue
# Type narrowing for API blocks
if isinstance(block.metadata, APIMetadata) and isinstance(block.content, APIResponseContent):
metadata = block.metadata
content = block.content
print(f"\n📡 API Response from {metadata.endpoint}")
# Type-safe access to JSON data
status_emoji = "✅" if content.status < 300 else "❌"
print(f" {status_emoji} Status: {content.status}")
print(f" Message: {content.message}")
if content.data:
print(" Data:")
for key, value in content.data.items():
print(f" {key}: {value}")
if content.errors:
print(" Errors:")
for error in content.errors:
print(f" ❗ {error}")
elif isinstance(event, BlockErrorEvent):
print(f"\n❌ Block Rejected: {event.reason}")
elif isinstance(event, TextContentEvent):
if event.content.strip():
print(f"[TEXT] {event.content.strip()}")
# ============================================================================
# EXAMPLE 3: Non-Dict Value Handling
# ============================================================================
@parse_as_yaml(strategy=ParseStrategy.PERMISSIVE, handle_non_dict=True)
class ScalarWrapperContent(BaseContent):
"""Content that wraps scalar YAML values in {'value': ...}."""
value: str | int | float | bool | None = None
@parse_as_yaml(strategy=ParseStrategy.PERMISSIVE, handle_non_dict=False)
class ScalarNoWrapContent(BaseContent):
"""Content that doesn't wrap scalar values (will fail on scalars)."""
message: str | None = None
class ScalarMetadata(BaseMetadata):
"""Metadata for scalar value blocks."""
block_type: str = "scalar"
data_type: str | None = None
# Create block types
ScalarWrapperBlock = Block[ScalarMetadata, ScalarWrapperContent]
ScalarNoWrapBlock = Block[ScalarMetadata, ScalarNoWrapContent]
async def example_3_non_dict_handling() -> None:
"""Demonstrate handling of non-dict YAML values."""
print("\n" + "=" * 70)
print("EXAMPLE 3: Non-Dict Value Handling")
print("=" * 70)
# Example 3a: With handle_non_dict=True (wraps scalars)
print("\n3a) With handle_non_dict=True (wraps in 'value' field):")
print("-" * 70)
syntax = DelimiterPreambleSyntax()
registry = Registry(syntax=syntax)
registry.register("scalar", ScalarWrapperBlock)
processor = StreamBlockProcessor(registry)
async def scalar_stream() -> AsyncIterator[str]:
text = dedent("""
Scalar values:
!!scalar_str:scalar
Hello World
!!end
!!scalar_num:scalar
42
!!end
!!scalar_bool:scalar
true
!!end
Done.
""")
for line in text.split("\n"):
yield line + "\n"
await asyncio.sleep(0.001)
async for event in processor.process_stream(scalar_stream()):
if isinstance(event, BlockEndEvent):
block = event.get_block()
if block is not None:
print(" Extracted Block:")
print(block.model_dump_json(indent=2))
elif isinstance(event, TextContentEvent):
if event.content.strip():
print(f"[TEXT] {event.content.strip()}")
# Example 3b: With handle_non_dict=False (scalars fail)
print("\n3b) With handle_non_dict=False (scalars cause fallback):")
print("-" * 70)
registry2 = Registry(syntax=syntax)
registry2.register("scalar", ScalarNoWrapBlock)
processor2 = StreamBlockProcessor(registry2)
async def scalar_stream2() -> AsyncIterator[str]:
text = dedent("""
Testing with dict and scalar:
!!scalar_dict:scalar
message: This is a dict, it works
!!end
!!scalar_fail:scalar
This is a scalar, will fall back to raw_content
!!end
Done.
""")
for line in text.split("\n"):
yield line + "\n"
await asyncio.sleep(0.001)
async for event in processor2.process_stream(scalar_stream2()):
if isinstance(event, BlockEndEvent):
block = event.get_block()
if block is not None:
print(" Extracted Block:")
print(block.model_dump_json(indent=2))
elif isinstance(event, TextContentEvent):
if event.content.strip():
print(f"[TEXT] {event.content.strip()}")
# ============================================================================
# EXAMPLE 4: Real-World Mixed Stream
# ============================================================================
@parse_as_yaml(strategy=ParseStrategy.PERMISSIVE)
class DatabaseConfigContent(BaseContent):
"""Database configuration from YAML."""
host: str | None = None
port: int | None = None
database: str | None = None
pool_size: int | None = None
@parse_as_json(strategy=ParseStrategy.PERMISSIVE)
class MetricsContent(BaseContent):
"""Performance metrics from JSON."""
cpu_usage: float | None = None
memory_mb: int | None = None
requests_per_sec: int | None = None
error_rate: float | None = None
class MixedMetadata(BaseMetadata):
"""Metadata for mixed content blocks."""
timestamp: str | None = None
# Create block types
DBConfigBlock = Block[MixedMetadata, DatabaseConfigContent]
MetricsBlock = Block[MixedMetadata, MetricsContent]
async def example_4_mixed_stream() -> None:
"""Demonstrate processing multiple content types in one stream."""
print("\n" + "=" * 70)
print("EXAMPLE 4: Real-World Mixed Stream")
print("=" * 70)
# Create syntax and registry with multiple block types
syntax = DelimiterFrontmatterSyntax()
registry = Registry(syntax=syntax)
registry.register("db_config", DBConfigBlock)
registry.register("metrics", MetricsBlock)
# Create processor
processor = StreamBlockProcessor(registry)
# Stream with multiple block types
async def mixed_stream() -> AsyncIterator[str]:
text = dedent("""
System monitoring report:
!!start
---
id: db_001
block_type: db_config
timestamp: "2024-12-15T10:00:00Z"
---
host: db.example.com
port: 5432
database: production
pool_size: 20
!!end
Database configured. Now checking metrics:
!!start
---
id: metrics_001
block_type: metrics
timestamp: "2024-12-15T10:01:00Z"
---
{
"cpu_usage": 45.2,
"memory_mb": 2048,
"requests_per_sec": 1250,
"error_rate": 0.02
}
!!end
!!start
---
id: metrics_002
block_type: metrics
timestamp: "2024-12-15T10:02:00Z"
---
{
"cpu_usage": 52.8,
"memory_mb": 2156,
"requests_per_sec": 1420,
"error_rate": 0.01
}
!!end
Report complete.
""")
for line in text.split("\n"):
yield line + "\n"
await asyncio.sleep(0.001)
# Process with type-aware handling
print("\nProcessing mixed content stream...")
db_configs: list[Any] = []
metrics_samples: list[Any] = []
async for event in processor.process_stream(mixed_stream()):
if isinstance(event, BlockEndEvent):
block = event.get_block()
if block is None:
continue
if block.metadata.block_type == "db_config":
db_configs.append(block)
print("\n🗄️ Extracted Database Config:")
print(block.model_dump_json(indent=2))
elif block.metadata.block_type == "metrics":
metrics_samples.append(block)
print("\n📊 Extracted Metrics:")
print(block.model_dump_json(indent=2))
elif isinstance(event, TextContentEvent):
if event.content.strip():
print(f"[TEXT] {event.content.strip()}")
# Summary
print("\n📈 Summary:")
print(f" DB Configs: {len(db_configs)}")
print(f" Metric Samples: {len(metrics_samples)}")
if metrics_samples:
# Calculate average CPU with type checking
cpu_values: list[float] = []
for m in metrics_samples:
if isinstance(m.content, MetricsContent) and m.content.cpu_usage is not None:
cpu_values.append(m.content.cpu_usage)
if cpu_values:
avg_cpu = sum(cpu_values) / len(cpu_values)
print(f" Average CPU: {avg_cpu:.1f}%")
# ============================================================================
# EXAMPLE 5: Error Handling Comparison
# ============================================================================
@parse_as_json(strategy=ParseStrategy.PERMISSIVE)
class PermissiveJSONContent(BaseContent):
"""JSON content with permissive parsing."""
status: str | None = None
count: int | None = None
@parse_as_json(strategy=ParseStrategy.STRICT)
class StrictJSONContent(BaseContent):
"""JSON content with strict parsing."""
status: str | None = None
count: int | None = None
class ErrorTestMetadata(BaseMetadata):
"""Metadata for error testing blocks."""
expected: str | None = None
# Create block types
PermissiveBlock = Block[ErrorTestMetadata, PermissiveJSONContent]
StrictBlock = Block[ErrorTestMetadata, StrictJSONContent]
async def example_5_error_handling() -> None:
"""Compare PERMISSIVE vs STRICT error handling."""
print("\n" + "=" * 70)
print("EXAMPLE 5: Error Handling Comparison")
print("=" * 70)
# Test data with both valid and invalid JSON
test_stream_text = dedent("""
Testing error handling:
!!test_valid:error_test
{"status": "ok", "count": 5}
!!end
!!test_invalid:error_test
{this is: not valid JSON at all}
!!end
!!test_empty:error_test
!!end
Done.
""")
# 5a: PERMISSIVE strategy
print("\n5a) PERMISSIVE Strategy (graceful fallback):")
print("-" * 70)
syntax = DelimiterPreambleSyntax()
registry_permissive = Registry(syntax=syntax)
registry_permissive.register("error_test", PermissiveBlock)
processor_permissive = StreamBlockProcessor(registry_permissive)
async def permissive_stream() -> AsyncIterator[str]:
for line in test_stream_text.split("\n"):
yield line + "\n"
await asyncio.sleep(0.001)
async for event in processor_permissive.process_stream(permissive_stream()):
if isinstance(event, BlockEndEvent):
block = event.get_block()
if block is not None:
print(" Extracted Block:")
print(block.model_dump_json(indent=2))
elif isinstance(event, TextContentEvent):
if event.content.strip():
print(f"[TEXT] {event.content.strip()}")
# 5b: STRICT strategy
print("\n5b) STRICT Strategy (raises errors):")
print("-" * 70)
registry_strict = Registry(syntax=syntax)
registry_strict.register("error_test", StrictBlock)
processor_strict = StreamBlockProcessor(registry_strict)
async def strict_stream() -> AsyncIterator[str]:
for line in test_stream_text.split("\n"):
yield line + "\n"
await asyncio.sleep(0.001)
async for event in processor_strict.process_stream(strict_stream()):
if isinstance(event, BlockEndEvent):
block = event.get_block()
if block is not None:
print(" ✅ Extracted Block:")
print(block.model_dump_json(indent=2))
elif isinstance(event, BlockErrorEvent):
# STRICT mode causes parsing failures to reject the block
print(f" ❌ Block rejected: {event.reason}")
elif isinstance(event, TextContentEvent):
if event.content.strip():
print(f"[TEXT] {event.content.strip()}")
# ============================================================================
# Main
# ============================================================================
async def main() -> None:
"""Run all examples."""
# Suppress library logging to stderr (Example 5 demonstrates error handling)
logging.getLogger("hother.streamblocks").setLevel(logging.CRITICAL)
print("🎯 Parsing Decorators Examples")
print("Demonstrating @parse_as_yaml() and @parse_as_json() decorators")
await example_1_basic_yaml_parsing()
await example_2_strict_json_parsing()
await example_3_non_dict_handling()
await example_4_mixed_stream()
await example_5_error_handling()
print("\n" + "=" * 70)
print(" - Use @parse_as_yaml() for automatic YAML parsing")
print(" - Use @parse_as_json() for automatic JSON parsing")
print(" - PERMISSIVE: falls back to raw_content on errors")
print(" - STRICT: raises exceptions on parsing errors")
print(" - handle_non_dict: wraps scalar values in {'value': ...}")
print(" - Works with any syntax (delimiter, markdown, etc.)")
if __name__ == "__main__":
asyncio.run(main())
Custom Syntax
Builds a completely custom syntax from scratch, defining your own open/close detection and section parsing, for when none of the built-in formats match your protocol.
src/hother/streamblocks_examples/02_syntaxes/04_custom_syntax.py
View source on GitHub
#!/usr/bin/env python3
import asyncio
import re
from textwrap import dedent
from typing import Any
from hother.streamblocks import Registry, StreamBlockProcessor
from hother.streamblocks.core.models import BlockCandidate, extract_block_types
from hother.streamblocks.core.types import (
BaseContent,
BaseMetadata,
BlockEndEvent,
DetectionResult,
ParseResult,
)
from hother.streamblocks.syntaxes.base import BaseSyntax
from hother.streamblocks_examples.blocks.agent.files import FileOperations
class XMLBlockSyntax(BaseSyntax):
"""Custom XML-like syntax for blocks.
Format:
<!-- block:type id="..." key="value" -->
content here
<!-- /block -->
"""
def __init__(self) -> None:
"""Initialize XML block syntax."""
# Pattern: <!-- block:type attr="value" ... -->
self._opening_pattern = re.compile(r"^<!--\s*block:(\w+)\s+(.+?)\s*-->$")
# Pattern: <!-- /block -->
self._closing_pattern = re.compile(r"^<!--\s*/block\s*-->$")
# Pattern for attributes: key="value"
self._attr_pattern = re.compile(r'(\w+)="([^"]*)"')
def detect_line(self, line: str, candidate: BlockCandidate | None) -> DetectionResult:
"""Detect XML-style block markers."""
stripped = line.strip()
if candidate is None:
# Looking for opening tag
match = self._opening_pattern.match(stripped)
if match:
block_type = match.group(1)
attrs_str = match.group(2)
# Parse attributes
attrs = dict(self._attr_pattern.findall(attrs_str))
attrs["block_type"] = block_type
return DetectionResult(is_opening=True, metadata=attrs)
else:
# Inside a block - check for closing
if self._closing_pattern.match(stripped):
return DetectionResult(is_closing=True)
# Accumulate content
candidate.content_lines.append(line)
return DetectionResult()
def should_accumulate_metadata(self, candidate: BlockCandidate) -> bool:
"""No separate metadata section - all in opening tag."""
return False
def extract_block_type(self, candidate: BlockCandidate) -> str | None:
"""Extract block_type from cached metadata."""
if not candidate.lines:
return None
match = self._opening_pattern.match(candidate.lines[0].strip())
return match.group(1) if match else None
def parse_block(
self, candidate: BlockCandidate, block_class: type[Any] | None = None
) -> ParseResult[BaseMetadata, BaseContent]:
"""Parse the complete block."""
if block_class is None:
metadata_class = BaseMetadata
content_class = BaseContent
else:
metadata_class, content_class = extract_block_types(block_class)
# Parse opening line for metadata
if not candidate.lines:
return ParseResult(success=False, error="No lines in candidate")
match = self._opening_pattern.match(candidate.lines[0].strip())
if not match:
return ParseResult(success=False, error="Invalid opening tag")
block_type = match.group(1)
attrs_str = match.group(2)
metadata_dict: dict[str, Any] = dict(self._attr_pattern.findall(attrs_str))
metadata_dict["block_type"] = block_type
try:
metadata = metadata_class(**metadata_dict)
except Exception as e:
return ParseResult(success=False, error=f"Metadata error: {e}", exception=e)
# Content is accumulated lines (excluding opening/closing)
content_text = "\n".join(candidate.content_lines)
try:
content = content_class.parse(content_text)
except Exception as e:
return ParseResult(success=False, error=f"Content error: {e}", exception=e)
return ParseResult(success=True, metadata=metadata, content=content)
async def main() -> None:
"""Demonstrate custom XML-like syntax."""
syntax = XMLBlockSyntax()
registry = Registry(syntax=syntax)
registry.register("files_operations", FileOperations)
processor = StreamBlockProcessor(registry)
text = dedent("""
Some text before the block.
<!-- block:files_operations id="ops001" description="Create files" -->
src/main.py:C
src/utils.py:E
<!-- /block -->
Some text after the block.
""").strip()
from hother.streamblocks_examples.helpers.simulator import simple_text_stream
print("=== Custom XML-like Syntax ===")
async for event in processor.process_stream(simple_text_stream(text)):
if isinstance(event, BlockEndEvent):
block = event.get_block()
if block:
print("Extracted block:")
print(block.model_dump_json(indent=2))
if __name__ == "__main__":
asyncio.run(main())