Skip to content

redis_service.operations.json_module

json_module

codex_platform.redis_service.operations.json_module

RedisJSON module operations (JSON.SET / JSON.GET / ...).

Requires the RedisJSON server module to be loaded on the Redis instance. Raises RuntimeError on first use if the module is not available in redis-py.

Classes

JsonModuleOperations

Redis JSON operations using the server-side RedisJSON module (JSON.* commands).

Requires the RedisJSON module on the server and redis-py with JSON support. Supports path-based access, atomic updates, array append, and more.

Example::

svc = JsonModuleOperations(client)
await svc.set("user:42", "$", {"name": "Alice", "age": 30})
name = await svc.get("user:42", "$.name")  # [["Alice"]]
Source code in src/codex_platform/redis_service/operations/json_module.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
class JsonModuleOperations:
    """Redis JSON operations using the server-side RedisJSON module (JSON.* commands).

    Requires the RedisJSON module on the server and redis-py with JSON support.
    Supports path-based access, atomic updates, array append, and more.

    Example::

        svc = JsonModuleOperations(client)
        await svc.set("user:42", "$", {"name": "Alice", "age": 30})
        name = await svc.get("user:42", "$.name")  # [["Alice"]]
    """

    def __init__(self, client: Redis) -> None:
        self.client = client

    def _require_json(self) -> None:
        if not _JSON_AVAILABLE:
            raise RuntimeError("RedisJSON module not available in redis-py. Install: pip install redis[hiredis]")

    @catch_redis_errors
    async def set(self, key: str, path: str, obj: Any, nx: bool = False, xx: bool = False) -> bool:
        """Set a JSON value at the given path (JSON.SET).

        Args:
            key: Redis key string.
            path: JSONPath expression, e.g. ``$`` or ``$.field``.
            obj: Any JSON-serializable object.
            nx: Set only if the key does not exist.
            xx: Set only if the key already exists.

        Returns:
            ``True`` if the value was successfully set.

        Raises:
            RuntimeError: If the RedisJSON module is not available.
            RedisConnectionError: Redis connection failure.
            RedisServiceError: Redis operation failure.
        """
        self._require_json()
        result = await self.client.json().set(key, path, obj, nx=nx, xx=xx)  # type: ignore[no-untyped-call]
        log.debug("JsonModuleOps | set key='%s' path='%s'", key, path)
        return bool(result)

    @catch_redis_errors
    async def get(self, key: str, path: str = "$") -> Any:
        """Retrieve a JSON value at the given path (JSON.GET).

        Args:
            key: Redis key string.
            path: JSONPath expression. Defaults to ``$`` (the whole document).

        Returns:
            Parsed value, or ``None`` if the key does not exist.

        Raises:
            RuntimeError: If the RedisJSON module is not available.
            RedisConnectionError: Redis connection failure.
            RedisServiceError: Redis operation failure.
        """
        self._require_json()
        return await self.client.json().get(key, path)  # type: ignore[no-untyped-call]

    @catch_redis_errors
    async def arrappend(self, key: str, path: str, *args: Any) -> int:
        """Append elements to a JSON array at the given path (JSON.ARRAPPEND).

        Args:
            key: Redis key string.
            path: JSONPath expression pointing to an array.
            *args: Values to append to the array.

        Returns:
            New length of the array after appending.

        Raises:
            RuntimeError: If the RedisJSON module is not available.
            RedisConnectionError: Redis connection failure.
            RedisServiceError: Redis operation failure.
        """
        self._require_json()
        count = await self.client.json().arrappend(key, path, *args)  # type: ignore[no-untyped-call]
        return int(count) if count else 0

    @catch_redis_errors
    async def delete(self, key: str, path: str = "$") -> int:
        """Delete a JSON value at the given path (JSON.DEL).

        Args:
            key: Redis key string.
            path: JSONPath expression. Defaults to ``$`` (the whole document).

        Returns:
            Number of elements deleted.

        Raises:
            RuntimeError: If the RedisJSON module is not available.
            RedisConnectionError: Redis connection failure.
            RedisServiceError: Redis operation failure.
        """
        self._require_json()
        result = await self.client.json().delete(key, path)  # type: ignore[no-untyped-call]
        return int(result) if result else 0
Functions
set(key, path, obj, nx=False, xx=False) async

Set a JSON value at the given path (JSON.SET).

Parameters:

Name Type Description Default
key str

Redis key string.

 required
path str

JSONPath expression, e.g. $ or $.field.

 required
obj Any

Any JSON-serializable object.

 required
nx bool

Set only if the key does not exist.

 False
xx bool

Set only if the key already exists.

 False

Returns:

Type Description
bool

True if the value was successfully set.

Raises:

Type Description
RuntimeError

If the RedisJSON module is not available.
RedisConnectionError

Redis connection failure.
RedisServiceError

Redis operation failure.
Source code in src/codex_platform/redis_service/operations/json_module.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
@catch_redis_errors
async def set(self, key: str, path: str, obj: Any, nx: bool = False, xx: bool = False) -> bool:
    """Set a JSON value at the given path (JSON.SET).

    Args:
        key: Redis key string.
        path: JSONPath expression, e.g. ``$`` or ``$.field``.
        obj: Any JSON-serializable object.
        nx: Set only if the key does not exist.
        xx: Set only if the key already exists.

    Returns:
        ``True`` if the value was successfully set.

    Raises:
        RuntimeError: If the RedisJSON module is not available.
        RedisConnectionError: Redis connection failure.
        RedisServiceError: Redis operation failure.
    """
    self._require_json()
    result = await self.client.json().set(key, path, obj, nx=nx, xx=xx)  # type: ignore[no-untyped-call]
    log.debug("JsonModuleOps | set key='%s' path='%s'", key, path)
    return bool(result)
get(key, path='$') async

Retrieve a JSON value at the given path (JSON.GET).

Parameters:

Name Type Description Default
key str

Redis key string.

 required
path str

JSONPath expression. Defaults to $ (the whole document).

 '$'

Returns:

Type Description
Any

Parsed value, or None if the key does not exist.

Raises:

Type Description
RuntimeError

If the RedisJSON module is not available.
RedisConnectionError

Redis connection failure.
RedisServiceError

Redis operation failure.
Source code in src/codex_platform/redis_service/operations/json_module.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
@catch_redis_errors
async def get(self, key: str, path: str = "$") -> Any:
    """Retrieve a JSON value at the given path (JSON.GET).

    Args:
        key: Redis key string.
        path: JSONPath expression. Defaults to ``$`` (the whole document).

    Returns:
        Parsed value, or ``None`` if the key does not exist.

    Raises:
        RuntimeError: If the RedisJSON module is not available.
        RedisConnectionError: Redis connection failure.
        RedisServiceError: Redis operation failure.
    """
    self._require_json()
    return await self.client.json().get(key, path)  # type: ignore[no-untyped-call]
arrappend(key, path, *args) async

Append elements to a JSON array at the given path (JSON.ARRAPPEND).

Parameters:

Name Type Description Default
key str

Redis key string.

 required
path str

JSONPath expression pointing to an array.

 required
*args Any

Values to append to the array.

 ()

Returns:

Type Description
int

New length of the array after appending.

Raises:

Type Description
RuntimeError

If the RedisJSON module is not available.
RedisConnectionError

Redis connection failure.
RedisServiceError

Redis operation failure.
Source code in src/codex_platform/redis_service/operations/json_module.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
@catch_redis_errors
async def arrappend(self, key: str, path: str, *args: Any) -> int:
    """Append elements to a JSON array at the given path (JSON.ARRAPPEND).

    Args:
        key: Redis key string.
        path: JSONPath expression pointing to an array.
        *args: Values to append to the array.

    Returns:
        New length of the array after appending.

    Raises:
        RuntimeError: If the RedisJSON module is not available.
        RedisConnectionError: Redis connection failure.
        RedisServiceError: Redis operation failure.
    """
    self._require_json()
    count = await self.client.json().arrappend(key, path, *args)  # type: ignore[no-untyped-call]
    return int(count) if count else 0
delete(key, path='$') async

Delete a JSON value at the given path (JSON.DEL).

Parameters:

Name Type Description Default
key str

Redis key string.

 required
path str

JSONPath expression. Defaults to $ (the whole document).

 '$'

Returns:

Type Description
int

Number of elements deleted.

Raises:

Type Description
RuntimeError

If the RedisJSON module is not available.
RedisConnectionError

Redis connection failure.
RedisServiceError

Redis operation failure.
Source code in src/codex_platform/redis_service/operations/json_module.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
@catch_redis_errors
async def delete(self, key: str, path: str = "$") -> int:
    """Delete a JSON value at the given path (JSON.DEL).

    Args:
        key: Redis key string.
        path: JSONPath expression. Defaults to ``$`` (the whole document).

    Returns:
        Number of elements deleted.

    Raises:
        RuntimeError: If the RedisJSON module is not available.
        RedisConnectionError: Redis connection failure.
        RedisServiceError: Redis operation failure.
    """
    self._require_json()
    result = await self.client.json().delete(key, path)  # type: ignore[no-untyped-call]
    return int(result) if result else 0

Functions