orchestrata_ultimate.postman.topic_registry

Reference › orchestrata_ultimate › postman › topic_registry

The registry of topic definitions for publish and subscribe, SPEC-002 story FR-005 (REQ-002-FR5-001 to -013). TopicRegistry implements ITopicRegistry from interfaces and stores TopicDefinition objects (from types) in memory by topic id. Postman owns one and exposes it through a read-only property; subscriptions themselves are kept by the subscription_manager.

create_topic stores a definition, generating a UUID if the id is empty, and returns the id. get_topic and delete_topic read and remove by id. list_topics returns the ids, optionally filtered by a pattern on the topic name, and match_topics returns the ids whose name matches a wildcard pattern; both can raise InvalidTopicPatternError for a malformed pattern.

At a glance

Class diagram

Dependency diagram

API

Classes

TopicRegistry

Bases: ITopicRegistry · Decorators: traces(spec='002', story='FR-005', reqs=['REQ-002-FR5-001', 'REQ-002-FR5-003', 'REQ-002-FR5-006', 'REQ-002-FR5-007', 'R… · Line 37

Concrete implementation of ITopicRegistry.

Source

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

"""
Topic Registry implementation.

Implements ITopicRegistry interface for managing topic definitions.
"""

import fnmatch
import logging

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.postman.exceptions import (
    InvalidTopicPatternError,
    TopicNotFoundError,
)
from orchestrata_ultimate.postman.interfaces import ITopicRegistry
from orchestrata_ultimate.postman.types import TopicDefinition

# Spec: 002 - Postman Message Broker


logger = logging.getLogger(__name__)


@traces(
    spec="002",
    story="FR-005",
    reqs=[
        "REQ-002-FR5-001",
        "REQ-002-FR5-003",
        "REQ-002-FR5-006",
        "REQ-002-FR5-007",
        "REQ-002-FR5-008",
        "REQ-002-FR5-009",
        "REQ-002-FR5-013",
    ],
)
class TopicRegistry(ITopicRegistry):
    """Concrete implementation of ITopicRegistry."""

    @traces(spec="002", story="FR-005", reqs=["REQ-002-FR5-001"])
    def __init__(self) -> None:
        self._topics: dict[str, TopicDefinition] = {}

    @traces(spec="002", story="FR-005", reqs=["REQ-002-FR5-006"])
    async def create_topic(self, definition: TopicDefinition) -> str:
        """Create a new topic.

        Stores the definition. If definition.topic_id is empty, a new UUID
        will have already been generated by default_factory.

        Args:
            definition: TopicDefinition to store.

        Returns:
            The topic_id of the created topic.

        Raises:
            InvalidTopicPatternError: If the definition contains an invalid pattern.
        """
        # Ensure the definition has a topic_id (should already be there)
        if not definition.topic_id:
            # This should not happen because of default_factory, but we are defensive
            from uuid import uuid4

            definition.topic_id = str(uuid4())
        self._topics[definition.topic_id] = definition
        return definition.topic_id

    @traces(spec="002", story="FR-005", reqs=["REQ-002-FR5-007"])
    async def delete_topic(self, topic_id: str) -> bool:
        """Delete a topic by ID.

        Args:
            topic_id: ID of the topic to delete.

        Returns:
            True if the topic existed and was deleted, False otherwise.

        Raises:
            TopicNotFoundError: If the topic does not exist.
        """
        if topic_id not in self._topics:
            raise TopicNotFoundError(
                f"Topic with ID '{topic_id}' not found.",
                error_code="TOPIC_NOT_FOUND",
                context={"topic_id": topic_id},
            )
        del self._topics[topic_id]
        return True

    @traces(spec="002", story="FR-005", reqs=["REQ-002-FR5-009"])
    async def get_topic(self, topic_id: str) -> TopicDefinition | None:
        """Retrieve a topic definition by ID.

        Args:
            topic_id: ID of the topic.

        Returns:
            TopicDefinition if found, None otherwise.
        """
        return self._topics.get(topic_id)

    @traces(spec="002", story="FR-006", reqs=["REQ-002-US-002-003"])
    async def list_topics(self, pattern: str | None = None) -> list[str]:
        """List topic IDs, optionally filtered by a pattern on the topic name.

        Args:
            pattern: Optional wildcard pattern to match against topic names.
                If None, all topic IDs are returned.

        Returns:
            List of topic IDs that match the pattern (or all if pattern is None).

        Raises:
            InvalidTopicPatternError: If the pattern is malformed.
        """
        if pattern is None:
            return list(self._topics.keys())

        # Validate pattern (simple check for now)
        try:
            # Use fnmatch to test pattern on a dummy string
            fnmatch.fnmatch("dummy", pattern)
        except TypeError as e:
            raise InvalidTopicPatternError(
                f"Invalid pattern '{pattern}': {e}",
                error_code="INVALID_PATTERN",
                context={"pattern": pattern},
            ) from e

        matched = []
        for tid, topic in self._topics.items():
            if fnmatch.fnmatch(topic.name, pattern):
                matched.append(tid)
        return matched

    @traces(spec="002", story="FR-005", reqs=["REQ-002-FR5-013"])
    async def match_topics(self, pattern: str) -> list[str]:
        """Find topic IDs whose name matches the given wildcard/regex pattern.

        Args:
            pattern: A wildcard pattern to match against topic names.

        Returns:
            List of topic IDs where the topic name matches the pattern.

        Raises:
            InvalidTopicPatternError: If the pattern is malformed.
        """
        # Use the same logic as list_topics
        return await self.list_topics(pattern)