Built-in Routines API Reference

This section provides API documentation for all built-in routines in Routilux.

Text Processing Routines

Text clipper routine.

Clips text to a maximum length while preserving important information.

class routilux.builtin_routines.text_processing.text_clipper.TextClipper

Bases: Routine

Routine for clipping text to a maximum length.

This routine clips text content to a specified maximum length while preserving important information. It handles special cases like tracebacks and provides informative truncation messages.

Features: - Preserves tracebacks completely (doesn’t clip them) - Clips text line by line to respect line boundaries - Provides informative truncation messages - Configurable maximum length

Examples

>>> clipper = TextClipper()
>>> clipper.set_config(max_length=1000)
>>> clipper.define_slot("input", handler=clipper.clip_text)
>>> clipper.define_event("output", ["clipped_text"])

__init__()

Initialize TextClipper routine.

Text renderer routine.

Renders objects (dicts, lists) into formatted text with XML-like tags.

class routilux.builtin_routines.text_processing.text_renderer.TextRenderer

Bases: Routine

Routine for rendering objects into formatted text.

This routine converts dictionaries and lists into formatted text with XML-like tags, making structured data more readable.

Features:

• Recursively renders nested dictionaries and lists

• Adds XML-like tags for structure

• Handles primitive types (str, int, float, bool)

• Configurable tag formatting

Examples

>>> renderer = TextRenderer()
>>> renderer.define_slot("input", handler=renderer.render)
>>> renderer.define_event("output", ["rendered_text"])
>>> # Render a dictionary
>>> data = {"name": "test", "value": 42}
>>> # Output: "<name>test</name>\n<value>42</value>"

__init__()

Initialize TextRenderer routine.

Result extractor routine.

Extracts and formats results from various output formats with extensible architecture.

class routilux.builtin_routines.text_processing.result_extractor.ExtractorProtocol(*args, **kwargs)

Bases: Protocol

Protocol for custom extractors.

extract(data, config)

Extract data from input.

Parameters:

• data (Any) – Input data to extract from.

• config (dict[str, Any]) – Configuration dictionary.

Returns:

Tuple of (extracted_result, format_type, metadata) if successful, None otherwise.

Return type:

tuple[Any, str, dict[str, Any]] | None

__init__(*args, **kwargs)

class routilux.builtin_routines.text_processing.result_extractor.ResultExtractor

Bases: Routine

Routine for extracting results from structured output.

This routine provides a flexible, extensible system for extracting structured data from various formats. It supports multiple extraction strategies and allows custom extractors to be registered.

Features:

• Multiple built-in extractors (JSON, YAML, XML, CSV, code blocks)

• Extensible architecture with custom extractor support

• Chain-based extraction with fallback mechanisms

• Intelligent type detection and conversion

• Comprehensive error handling and reporting

• Rich metadata about extraction process

Extraction Strategies:

• “auto”: Try all extractors in order until one succeeds

• “priority”: Try extractors in priority order

• “all”: Extract using all applicable extractors

• “first_match”: Return first successful extraction

Examples

>>> extractor = ResultExtractor()
>>> extractor.set_config(strategy="auto")
>>> extractor.input_slot.receive({"data": '```json\n{"key": "value"}\n```'})
>>> # Register custom extractor
>>> def my_extractor(data, config):
...     if isinstance(data, str) and data.startswith("CUSTOM:"):
...         return data[7:], "custom", {"method": "prefix"}
...     return None
>>> extractor.register_extractor("custom", my_extractor)

__init__()

Initialize ResultExtractor routine.

register_extractor(name, extractor)

Register a custom extractor function.

Parameters:

• name (str) – Extractor name (must be unique).

• extractor (Callable[[Any, dict[str, Any]], tuple[Any, str, dict[str, Any]] | None]) – Extractor function that takes (data, config) and returns (extracted_data, format_type, metadata) or None.

Examples

>>> def my_extractor(data, config):
...     if isinstance(data, str) and data.startswith("CUSTOM:"):
...         return data[7:], "custom", {"method": "prefix"}
...     return None
>>> extractor.register_extractor("custom_prefix", my_extractor)

Utility Routines

Time provider routine.

Provides current time in various formats.

class routilux.builtin_routines.utils.time_provider.TimeProvider

Bases: Routine

Routine for providing current time information.

This routine provides current time in various formats, useful for logging, timestamps, and time-based operations.

Features: - Multiple time formats (ISO, formatted string, timestamp) - Configurable format strings - Timezone support - Locale-aware formatting

Examples

>>> time_provider = TimeProvider()
>>> time_provider.set_config(format="iso", include_weekday=True)
>>> time_provider.define_slot("request", handler=time_provider.get_time)
>>> time_provider.define_event("output", ["time_string", "timestamp", "datetime"])

__init__()

Initialize TimeProvider routine.

Data flattener routine.

Flattens nested data structures into flat dictionaries.

class routilux.builtin_routines.utils.data_flattener.DataFlattener

Bases: Routine

Routine for flattening nested data structures.

This routine converts nested dictionaries, lists, and Serializable objects into flat dictionaries with dot-notation keys.

Features: - Recursive flattening of nested structures - Handles Serializable objects - Configurable separator for keys - Preserves list indices

Examples

>>> flattener = DataFlattener()
>>> flattener.set_config(separator=".")
>>> flattener.define_slot("input", handler=flattener.flatten)
>>> flattener.define_event("output", ["flattened_data"])

>>> # Input: {"a": {"b": 1, "c": [2, 3]}}
>>> # Output: {"a.b": 1, "a.c.0": 2, "a.c.1": 3}

__init__()

Initialize DataFlattener routine.

Data Processing Routines

Data transformer routine.

Transforms data using configurable transformation functions.

class routilux.builtin_routines.data_processing.data_transformer.DataTransformer

Bases: Routine

Routine for transforming data using configurable functions.

This routine applies transformation functions to input data, useful for data cleaning, normalization, and format conversion.

Features: - Configurable transformation functions - Chain multiple transformations - Support for custom transformation logic - Error handling and validation

Examples

>>> transformer = DataTransformer()
>>> transformer.set_config(
...     transformations=["lowercase", "strip_whitespace"]
... )
>>> transformer.define_slot("input", handler=transformer.transform)
>>> transformer.define_event("output", ["transformed_data", "transformation_applied"])

__init__()

Initialize DataTransformer routine.

register_transformation(name, func)

Register a custom transformation function.

Parameters:

• name (str) – Transformation name.

• func (Callable) – Transformation function that takes data and returns transformed data.

Data validator routine.

Validates data against schemas or validation rules.

class routilux.builtin_routines.data_processing.data_validator.DataValidator

Bases: Routine

Routine for validating data against schemas or rules.

This routine validates input data against configurable validation rules, useful for data quality checks and schema validation.

Features: - Configurable validation rules - Support for custom validation functions - Detailed validation error messages - Optional strict mode (stop on first error)

Examples

>>> validator = DataValidator()
>>> validator.set_config(
...     rules={"name": lambda x: isinstance(x, str) and len(x) > 0}
... )
>>> validator.define_slot("input", handler=validator.validate)
>>> validator.define_event("output", ["is_valid", "errors", "validated_data"])

__init__()

Initialize DataValidator routine.

register_validator(name, func)

Register a custom validator function.

Parameters:

• name (str) – Validator name.

• func (Callable) – Validation function that takes value and returns bool or (bool, error_msg).

Control Flow Routines

Conditional router routine.

Routes data to different outputs based on conditions.

class routilux.builtin_routines.control_flow.conditional_router.ConditionalRouter

Bases: Routine

Routine for routing data based on conditions.

This routine evaluates conditions on input data and routes it to different output events based on which conditions are met.

Features:

• Multiple conditional routes

• Configurable condition functions, string expressions, or dictionaries

• Default route for unmatched cases

• Priority-based routing

• Access to routine’s config and stats in conditions

• Full serialization support

Condition Types:

• String expressions (recommended): Fully serializable, can access data, config, and stats variables

• Dictionary conditions: Field matching, fully serializable

• Function references: Module-level functions, serializable if in module. Can accept data, config, and stats as parameters

• Lambda functions: Can be used at runtime, may be converted to string expressions during serialization (if source code is available). Can access external variables via closure, but closure variables are lost during serialization

Examples

Using string expressions with config access (recommended):

>>> router = ConditionalRouter()
>>> router.set_config(
...     routes=[
...         ("high", "data.get('value', 0) > config.get('threshold', 0)"),
...         ("low", "data.get('value', 0) <= config.get('threshold', 0)"),
...     ],
...     threshold=10
... )
>>> router.input_slot.receive({"data": {"value": 15}})  # Routes to "high"

Using string expressions with stats access:

>>> router = ConditionalRouter()
>>> router.set_config(
...     routes=[
...         ("active", "stats.get('count', 0) < 10"),
...         ("full", "stats.get('count', 0) >= 10"),
...     ]
... )
>>> # Note: set_stat() is deprecated, use JobState for execution state
>>> router.input_slot.receive({"data": {}})  # Routes to "active"

Using dictionary conditions:

>>> router = ConditionalRouter()
>>> router.set_config(
...     routes=[
...         ("high", {"priority": "high"}),
...         ("low", {"priority": "low"}),
...     ]
... )
>>> router.input_slot.receive({"data": {"priority": "high"}})

Using lambda functions (runtime only, serialization may fail):

>>> threshold = 10
>>> router = ConditionalRouter()
>>> router.set_config(
...     routes=[
...         ("high", lambda data: data.get('value', 0) > threshold),
...     ]
... )
>>> # Lambda works at runtime but may not serialize properly

__init__()

Initialize ConditionalRouter routine.

serialize()

Serialize ConditionalRouter, handling lambda functions in routes.

Callable conditions are automatically serialized using the serialization module’s smart serialization with fallback to expression extraction.

Returns:

Serialized dictionary.

Return type:

dict[str, Any]

deserialize(data, registry=None)

Deserialize ConditionalRouter, restoring callable conditions from routes.

Callable conditions are automatically deserialized by the serialization module, including support for lambda expressions.

Parameters:

• data (dict[str, Any]) – Serialized dictionary.

• registry (Any | None) – Optional ObjectRegistry for deserializing callables.

add_route(route_name, condition)

Add a routing condition.

Parameters:

• route_name (str) – Name of the route (will be used as event name).

• condition (Callable | dict[str, Any] | str) – Condition function, dictionary, or string expression.