scripts.redis_exec

Reference › scripts › redis_exec

scripts/redis_exec.py executes one Redis command against the project's real Redis; it is the command-line seam that a plugin calls. Usage is redis_exec.py COMMAND [ARG ...]. main builds a synchronous client with create_sync_redis from orchestrata_ultimate.core.redis_factory, runs the command through execute_command and prints the result rendered by format_result, in the form the plugin's parsers expect. With no arguments it prints a usage line and exits 2. Any failure is printed as redis_exec failed: on standard error and exits 1; this is a fail-open seam, and callers treat a non-zero exit as "skip".

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/redis_exec.py

"""Execute one Redis command against the project's real Redis — CLI seam
for tooling that cannot import this repo's Python directly.

Built for the SPEC-021 OpenCode plugin (opencode/plugins/lib/redis-exec.ts):
its original transport was a bare ``redis-cli`` shell-out, which fails twice
over on this machine — redis-cli is not installed at all (Arch ships valkey,
uninstalled), and a bare invocation targets localhost while Redis is an
Incus container on the ``10.53.1.x`` fleet. Shelling to
``uv run python scripts/redis_exec.py`` instead reuses
:func:`orchestrata_ultimate.core.redis_factory.create_sync_redis` — the one
place that already resolves host/port/auth from ``.env`` — so the plugin
needs no Redis connection knowledge of its own.

Usage::

    uv run python scripts/redis_exec.py INCR some:counter
    uv run python scripts/redis_exec.py LPUSH some:list '{"json": "blob"}'
    uv run python scripts/redis_exec.py SMEMBERS some:set

Output: the command's result on stdout — scalars verbatim, ``None`` as an
empty string, collections one element per line (matching redis-cli's line
protocol closely enough for the plugin's ``.text()`` parsing). Exit 0 on
success, 2 on usage error, 1 on any Redis/connection failure (callers are
fail-open and treat nonzero as "capture skipped").
"""

from __future__ import annotations

import sys


def format_result(result: object) -> str:
    """Render a redis-py return value the way the plugin's parsers expect."""
    if result is None:
        return ""
    if isinstance(result, bytes):
        return result.decode("utf-8", errors="replace")
    if isinstance(result, (list, set, tuple)):
        return "\n".join(format_result(item) for item in result)
    if isinstance(result, bool):
        # redis-py returns True for OK-replies (SET etc.); render like
        # redis-cli's integer replies do for predictable parsing.
        return "1" if result else "0"
    return str(result)


def main(argv: list[str] | None = None) -> int:
    args = sys.argv[1:] if argv is None else argv
    if not args:
        print("usage: redis_exec.py COMMAND [ARG ...]", file=sys.stderr)
        return 2
    try:
        from orchestrata_ultimate.core.redis_factory import create_sync_redis

        client = create_sync_redis()
        result = client.execute_command(*args)
        print(format_result(result))
        return 0
    except Exception as exc:  # fail-open seam: callers treat nonzero as skip
        print(f"redis_exec failed: {exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    sys.exit(main())