Skip to content

Logging

Integrate StreamBlocks with stdlib logging, structlog, loguru, or a custom logger.

How loggers are injected

Both Registry and StreamBlockProcessor accept an optional logger argument:

registry = Registry(syntax=syntax, logger=logger)
processor = StreamBlockProcessor(registry, logger=logger)

Anything that implements the Logger protocol works: an object with debug, info, warning, error, and exception methods. StreamBlocks passes structured data as direct keyword arguments, the native pattern of structlog and loguru:

logger.info("block_extracted", block_type="files", block_id="abc123")

If you pass nothing, both components default to a wrapped logging.getLogger(__name__), so log output integrates with your existing stdlib configuration out of the box.

Stdlib logging

Stdlib loggers don't accept arbitrary kwargs, so wrap them in StdlibLoggerAdapter. The adapter appends structured fields to the message (msg | key=value …) and also stores them in the extra dict for handlers and filters:

# Configure stdlib logging
logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s [%(levelname)s] %(name)s - %(message)s",
)

# Wrap stdlib logger with adapter to enable direct kwargs and auto-display
stdlib_logger = logging.getLogger("my_app.streamblocks")
logger = StdlibLoggerAdapter(stdlib_logger)

syntax = DelimiterPreambleSyntax()
registry = Registry(syntax=syntax, logger=logger)
registry.register("files_operations", FileOperations)

from hother.streamblocks.core.processor import ProcessorConfig

config = ProcessorConfig(lines_buffer=5)
processor = StreamBlockProcessor(registry, config=config, logger=logger)

async for event in processor.process_stream(example_stream()):
    if isinstance(event, BlockEndEvent):
        block = event.get_block()
        if block is not None:
            print("✓ Extracted block:")
            print(block.model_dump_json(indent=2))

View source on GitHub

Structlog

Structlog loggers support direct kwargs natively, so no adapter is needed; configure structlog and pass the bound logger straight in. Install with the structlog extra (pip install streamblocks[structlog]):

# Configure structlog
structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.dev.ConsoleRenderer(),
    ],
    wrapper_class=structlog.make_filtering_bound_logger(logging.DEBUG),
    context_class=dict,
    logger_factory=structlog.PrintLoggerFactory(),
    cache_logger_on_first_use=False,
)

# Get a structlog logger
logger = structlog.get_logger("streamblocks")

syntax = DelimiterPreambleSyntax()
registry = Registry(syntax=syntax, logger=logger)
registry.register("files_operations", FileOperations)

from hother.streamblocks.core.processor import ProcessorConfig

config = ProcessorConfig(lines_buffer=5)
processor = StreamBlockProcessor(registry, config=config, logger=logger)

async for event in processor.process_stream(example_stream()):
    if isinstance(event, BlockEndEvent):
        block = event.get_block()
        if block is not None:
            print("✓ Extracted block:")
            print(block.model_dump_json(indent=2))

View source on GitHub

Loguru

Loguru also supports direct kwargs natively. Install with the loguru extra (pip install streamblocks[loguru]) and pass the global logger directly:

from loguru import logger

registry = Registry(syntax=syntax, logger=logger)
processor = StreamBlockProcessor(registry, logger=logger)

Custom logger

Any object with the five methods satisfies the protocol; no base class required. Accept **kwargs to receive the structured fields:

class CustomLogger:
    """Custom logger that implements the Logger protocol.

    This shows that any object with the required methods can be used as a logger.
    Demonstrates handling direct kwargs (the StreamBlocks pattern).
    """

    def __init__(self, prefix: str = "CUSTOM") -> None:
        """Initialize custom logger with prefix."""
        self.prefix = prefix

    def debug(self, msg: str, *args: Any, **kwargs: Any) -> None:
        """Log debug message with kwargs as structured data."""
        print(f"[{self.prefix} DEBUG] {msg} {kwargs}")

    def info(self, msg: str, *args: Any, **kwargs: Any) -> None:
        """Log info message with kwargs as structured data."""
        print(f"[{self.prefix} INFO] {msg} {kwargs}")

    def warning(self, msg: str, *args: Any, **kwargs: Any) -> None:
        """Log warning message with kwargs as structured data."""
        print(f"[{self.prefix} WARNING] {msg} {kwargs}")

    def error(self, msg: str, *args: Any, **kwargs: Any) -> None:
        """Log error message with kwargs as structured data."""
        print(f"[{self.prefix} ERROR] {msg} {kwargs}")

    def exception(self, msg: str, *args: Any, **kwargs: Any) -> None:
        """Log exception message with kwargs as structured data."""
        print(f"[{self.prefix} EXCEPTION] {msg} {kwargs}")

Then inject it like any other logger:

logger = CustomLogger(prefix="MY_APP")

syntax = DelimiterPreambleSyntax()
registry = Registry(syntax=syntax, logger=logger)
registry.register("files_operations", FileOperations)

from hother.streamblocks.core.processor import ProcessorConfig

config = ProcessorConfig(lines_buffer=5)
processor = StreamBlockProcessor(registry, config=config, logger=logger)

async for event in processor.process_stream(example_stream()):
    if isinstance(event, BlockEndEvent):
        block = event.get_block()
        if block is not None:
            print("✓ Extracted block:")
            print(block.model_dump_json(indent=2))

View source on GitHub

What gets logged

The registry logs block-type registration and block parsing/validation steps at DEBUG. The processor logs stream start at DEBUG and input adapter auto-detection at INFO. Run any example with level DEBUG to see the full trace, then raise the level in production.

Next steps