orchestrata_ultimate.postman.routing_rules

Reference › orchestrata_ultimate › postman › routing_rules

The content-based routing rules of the broker, SPEC-002 story US-005 (REQ-002-US-005-012 to -020, REQ-002-FR3-009). RoutingRulesEngine implements IRoutingRules from interfaces and lets an operator define rules such as "messages from this sender to this topic also go to these agents". Matching is delegated to filter. No other module imports it.

add_rule stores a RoutingRule (from types) by its id and raises InvalidRoutingRuleError when the rule has no name or no destination. remove_rule, update_rule and list_rules manage the set. evaluate(envelope, payload) sorts the rules by priority number and then name, skips disabled ones, and for each rule whose filter matches adds its destinations to the result; a rule with stop_on_match ends the walk. The result is the combined destination list, which can be empty.

At a glance

Class diagram

Dependency diagram

API

Classes

RoutingRulesEngine

Bases: IRoutingRules · Decorators: traces(spec='002', story='US-005', reqs=['REQ-002-US-005-012', 'REQ-002-US-005-013', 'REQ-002-US-005-014', 'REQ-002-US-… · Line 60

Implementation of IRoutingRules.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/postman/routing_rules.py

# mypy: ignore-errors
"""
Routing Rules Engine.

Implements IRoutingRules interface for managing and evaluating routing rules.
"""

# Spec: 002 - Postman Message Broker

import asyncio
import logging
from typing import TYPE_CHECKING

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.postman.exceptions import InvalidRoutingRuleError
from orchestrata_ultimate.postman.interfaces import IMessageFilter, IRoutingRules
from orchestrata_ultimate.postman.types import (
    MessageEnvelope,
    MessagePayload,
    RoutingRule,
)

logger = logging.getLogger(__name__)

# Try to import our filter implementation, but provide a fallback. Resolved
# via TYPE_CHECKING so pyright always type-checks against the real
# postman.filter.MessageFilter (a single canonical type) instead of also
# seeing the runtime-only fallback stub below, whose narrower
# `compile_filter(...) -> None` doesn't structurally match
# IMessageFilter.compile_filter's `-> Callable` — that mismatch only matters
# if the fallback branch is ever actually reached (missing module), which
# never happens for pyright's own static analysis of this in-package import.
if TYPE_CHECKING:
    from orchestrata_ultimate.postman.filter import MessageFilter
else:
    try:
        from orchestrata_ultimate.postman.filter import MessageFilter
    except ImportError:
        # If filter module is not available, define a placeholder
        class MessageFilter(IMessageFilter):
            def matches(self, envelope, payload, criteria) -> bool:
                raise NotImplementedError("MessageFilter not available")

            @traces(spec="002", story="FR-003", reqs=["REQ-002-FR3-009"])
            def compile_filter(self, criteria) -> None:
                raise NotImplementedError("MessageFilter not available")


@traces(
    spec="002",
    story="US-005",
    reqs=[
        "REQ-002-US-005-012",
        "REQ-002-US-005-013",
        "REQ-002-US-005-014",
        "REQ-002-US-005-015",
        "REQ-002-US-005-018",
    ],
)
class RoutingRulesEngine(IRoutingRules):
    """Implementation of IRoutingRules."""

    def __init__(self, message_filter: IMessageFilter | None = None) -> None:
        """
        Initialize the routing rules engine.

        Args:
            message_filter: Optional IMessageFilter instance.
                If not provided, a default MessageFilter will be created.
        """
        self._rules: dict[str, RoutingRule] = {}
        self._filter = message_filter or MessageFilter()
        self._lock = asyncio.Lock()

    @traces(
        spec="002",
        story="US-005",
        reqs=["REQ-002-US-005-012", "REQ-002-US-005-017"],
    )
    async def add_rule(self, rule: RoutingRule) -> str:
        """
        Add a routing rule.

        Args:
            rule: Routing rule to add.

        Returns:
            The rule_id of the added rule.

        Raises:
            InvalidRoutingRuleError: If the rule is invalid.
        """
        async with self._lock:
            # Basic validation
            if not rule.name:
                raise InvalidRoutingRuleError("Rule must have a non‑empty name")
            if not rule.destinations:
                raise InvalidRoutingRuleError(
                    "Rule must specify at least one destination"
                )
            self._rules[rule.rule_id] = rule
            return rule.rule_id

    @traces(spec="002", story="US-005", reqs=["REQ-002-US-005-014"])
    async def remove_rule(self, rule_id: str) -> bool:
        """
        Remove a routing rule.

        Args:
            rule_id: ID of the rule to remove.

        Returns:
            True if the rule was found and removed, False otherwise.
        """
        async with self._lock:
            if rule_id in self._rules:
                del self._rules[rule_id]
                return True
            return False

    @traces(spec="002", story="US-005", reqs=["REQ-002-US-005-013"])
    async def update_rule(self, rule_id: str, rule: RoutingRule) -> bool:
        """
        Update an existing routing rule.

        Args:
            rule_id: ID of the rule to update.
            rule: Updated routing rule (its rule_id field may differ; it will be forced to rule_id).

        Returns:
            True if the rule was found and updated, False otherwise.
        """
        async with self._lock:
            if rule_id not in self._rules:
                return False
            # Ensure the stored rule uses the given rule_id
            rule.rule_id = rule_id
            self._rules[rule_id] = rule
            return True

    @traces(spec="002", story="US-005", reqs=["REQ-002-US-005-015"])
    async def list_rules(self) -> list[RoutingRule]:
        """
        List all routing rules.

        Returns:
            List of all stored routing rules.
        """
        async with self._lock:
            return list(self._rules.values())

    @traces(
        spec="002",
        story="US-005",
        reqs=[
            "REQ-002-FR3-009",
            "REQ-002-US-005-018",
            "REQ-002-US-005-019",
            "REQ-002-US-005-020",
        ],
    )
    async def evaluate(
        self,
        envelope: MessageEnvelope,
        payload: MessagePayload,
    ) -> list[str]:
        """
        Evaluate routing rules against a message.

        Args:
            envelope: Message envelope.
            payload: Message payload.

        Returns:
            List of destination agent IDs resulting from matching rules.
        """
        # Sort rules by priority (lower priority value = higher priority)
        sorted_rules = sorted(self._rules.values(), key=lambda r: (r.priority, r.name))
        destinations = []
        for rule in sorted_rules:
            if not rule.enabled:
                continue
            if self._filter.matches(envelope, payload, rule.filter_criteria):
                destinations.extend(rule.destinations)
                if rule.stop_on_match:
                    break
        return destinations