orchestrata_ultimate.verification.integrations.redis_manager

Reference › orchestrata_ultimate › verification › integrations › redis_manager

VerificationRedisManager persists verification state as JSON strings in Redis (Spec 035 US-001). It is used by the orchestrator's verification_gate. It takes an optional synchronous Redis client and a key prefix (default orchestrata_ultimate:verification); every key is stored as the prefix, a colon and the caller's key. save_state writes a dictionary with SET, load_state reads and decodes it, delete_state removes it and list_keys returns the keys matching a pattern with the prefix stripped. It degrades quietly: with no client, save_state and delete_state return false, load_state returns None and list_keys returns an empty list; Redis errors are caught, logged as warnings where logged, and reported the same way. The available property says whether a client was supplied. The settings that name the Redis location live in verification_config.

At a glance

Class diagram

Dependency diagram

API

Classes

VerificationRedisManager

Decorators: traces(spec='035', story='US-001', reqs=['REQ-035-US-001-010', 'REQ-035-US-001-018', 'REQ-035-US-002-009', 'REQ-035-US-… · Line 34

Manages verification state in Redis.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/verification/integrations/redis_manager.py

# mypy: ignore-errors
"""Redis state management for verification."""

# Spec: 035 - Verification Agent

from __future__ import annotations

import json
import logging
from typing import TYPE_CHECKING, Any, cast

from orchestrata_ultimate.core.base.traces import traces

if TYPE_CHECKING:
    from redis import Redis

logger = logging.getLogger(__name__)


@traces(
    spec="035",
    story="US-001",
    reqs=[
        "REQ-035-US-001-010",
        "REQ-035-US-001-018",
        "REQ-035-US-002-009",
        "REQ-035-US-002-015",
        "REQ-035-US-002-022",
        "REQ-035-US-003-006",
        "REQ-035-US-003-013",
        "REQ-035-US-003-022",
    ],
)
class VerificationRedisManager:
    """Manages verification state in Redis."""

    def __init__(
        self,
        redis: Redis | None = None,
        prefix: str = "orchestrata_ultimate:verification",
    ) -> None:
        self._redis = redis
        self._prefix = prefix

    @traces(
        spec="035",
        story="US-001",
        reqs=[
            "REQ-035-US-001-010",
            "REQ-035-US-002-009",
            "REQ-035-US-002-022",
            "REQ-035-US-003-013",
        ],
    )
    def save_state(self, key: str, data: dict[str, Any]) -> bool:
        """Save verification state to Redis."""
        if not self._redis:
            return False
        try:
            full_key = f"{self._prefix}:{key}"
            self._redis.set(full_key, json.dumps(data))
            return True
        except Exception as exc:
            logger.warning("Failed to save state: %s", exc)
            return False

    @traces(spec="035", story="US-002", reqs=["REQ-035-US-002-015"])
    def load_state(self, key: str) -> dict[str, Any] | None:
        """Load verification state from Redis."""
        if not self._redis:
            return None
        try:
            full_key = f"{self._prefix}:{key}"
            # get()/keys() are typed via redis-py's shared sync/async
            # command mixin (ResponseT); self._redis is always the
            # sync client, which always returns the plain value.
            data = cast("bytes | str | None", self._redis.get(full_key))
            if data:
                return json.loads(data)
        except Exception as exc:
            logger.warning("Failed to load state: %s", exc)
        return None

    def delete_state(self, key: str) -> bool:
        """Delete verification state."""
        if not self._redis:
            return False
        try:
            full_key = f"{self._prefix}:{key}"
            self._redis.delete(full_key)
            return True
        except Exception as exc:
            logger.warning("Failed to delete state: %s", exc)
            return False

    def list_keys(self, pattern: str = "*") -> list[str]:
        """List verification state keys."""
        if not self._redis:
            return []
        try:
            full_pattern = f"{self._prefix}:{pattern}"
            # See the cast() note in load_state() above.
            keys = cast("list[bytes | str]", self._redis.keys(full_pattern))
            prefix_len = len(self._prefix) + 1
            return [
                k.decode()[prefix_len:] if isinstance(k, bytes) else k[prefix_len:]
                for k in keys
            ]
        except Exception:
            return []

    @property
    @traces(spec="035", story="US-001", reqs=["REQ-035-US-001-018"])
    def available(self) -> bool:
        return self._redis is not None