Skip to content

redis_service.operations.list_

list_

codex_platform.redis_service.operations.list_

Redis List operations.

Classes

ListOperations

Redis List operations (RPUSH / LPUSH / LPOP / LRANGE and more).

Accepts an already-constructed redis.asyncio.Redis client. All methods wrap Redis errors in typed exceptions.

Example::

ops = ListOperations(client)
await ops.push("queue:jobs", "job-1", "job-2")
job = await ops.pop("queue:jobs")
Source code in src/codex_platform/redis_service/operations/list_.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
class ListOperations:
    """Redis List operations (RPUSH / LPUSH / LPOP / LRANGE and more).

    Accepts an already-constructed ``redis.asyncio.Redis`` client.
    All methods wrap Redis errors in typed exceptions.

    Example::

        ops = ListOperations(client)
        await ops.push("queue:jobs", "job-1", "job-2")
        job = await ops.pop("queue:jobs")
    """

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

    @catch_redis_errors
    async def push(self, key: "str | BaseRedisKey", *values: str, **kwargs: Any) -> int:
        """Append one or more values to the end of a list (RPUSH).

        Args:
            key: Redis key or a ``BaseRedisKey`` instance.
            *values: String values to append.
            **kwargs: Extra parameters forwarded to ``resolve_key``.

        Returns:
            New length of the list after the operation.

        Raises:
            RedisConnectionError: Redis connection failure.
            RedisServiceError: Redis operation failure.
        """
        real_key = resolve_key(key, **kwargs)
        return int(await self.client.rpush(real_key, *values))

    @catch_redis_errors
    async def lpush(self, key: "str | BaseRedisKey", *values: str, **kwargs: Any) -> int:
        """Prepend one or more values to the beginning of a list (LPUSH).

        Args:
            key: Redis key or a ``BaseRedisKey`` instance.
            *values: String values to prepend.
            **kwargs: Extra parameters forwarded to ``resolve_key``.

        Returns:
            New length of the list after the operation.

        Raises:
            RedisConnectionError: Redis connection failure.
            RedisServiceError: Redis operation failure.
        """
        real_key = resolve_key(key, **kwargs)
        return int(await self.client.lpush(real_key, *values))

    @catch_redis_errors
    async def pop(self, key: "str | BaseRedisKey", **kwargs: Any) -> str | None:
        """Remove and return the first element of a list (LPOP).

        Args:
            key: Redis key or a ``BaseRedisKey`` instance.
            **kwargs: Extra parameters forwarded to ``resolve_key``.

        Returns:
            The first element, or ``None`` if the list is empty or does not exist.

        Raises:
            RedisConnectionError: Redis connection failure.
            RedisServiceError: Redis operation failure.
        """
        real_key = resolve_key(key, **kwargs)
        value = await self.client.lpop(real_key)
        return str(value) if value is not None else None

    @catch_redis_errors
    async def rpop(self, key: "str | BaseRedisKey", **kwargs: Any) -> str | None:
        """Remove and return the last element of a list (RPOP).

        Args:
            key: Redis key or a ``BaseRedisKey`` instance.
            **kwargs: Extra parameters forwarded to ``resolve_key``.

        Returns:
            The last element, or ``None`` if the list is empty or does not exist.

        Raises:
            RedisConnectionError: Redis connection failure.
            RedisServiceError: Redis operation failure.
        """
        real_key = resolve_key(key, **kwargs)
        value = await self.client.rpop(real_key)
        return str(value) if value is not None else None

    @catch_redis_errors
    async def brpop(self, *keys: str, timeout: int = 0) -> tuple[str, str] | None:
        """Blocking pop of the last element from the first non-empty list (BRPOP).

        Args:
            *keys: One or more Redis key strings to check in order.
            timeout: Maximum seconds to block. ``0`` blocks indefinitely.

        Returns:
            ``(key, value)`` tuple when an element is available,
            or ``None`` if the timeout expires.

        Raises:
            RedisConnectionError: Redis connection failure.
            RedisServiceError: Redis operation failure.
        """
        result = await self.client.brpop(list(keys), timeout=timeout)
        if result:
            return str(result[0]), str(result[1])
        return None

    @catch_redis_errors
    async def range(self, key: "str | BaseRedisKey", start: int = 0, end: int = -1, **kwargs: Any) -> list[str]:
        """Return a slice of list elements by index range (LRANGE).

        Args:
            key: Redis key or a ``BaseRedisKey`` instance.
            start: Start index (0-based). Negative values count from the end.
            end: End index inclusive. ``-1`` means the last element.
            **kwargs: Extra parameters forwarded to ``resolve_key``.

        Returns:
            List of elements in the requested range.

        Raises:
            RedisConnectionError: Redis connection failure.
            RedisServiceError: Redis operation failure.
        """
        real_key = resolve_key(key, **kwargs)
        return await self.client.lrange(real_key, start, end)

    @catch_redis_errors
    async def length(self, key: "str | BaseRedisKey", **kwargs: Any) -> int:
        """Return the number of elements in a list (LLEN).

        Args:
            key: Redis key or a ``BaseRedisKey`` instance.
            **kwargs: Extra parameters forwarded to ``resolve_key``.

        Returns:
            List length. ``0`` if the key does not exist.

        Raises:
            RedisConnectionError: Redis connection failure.
            RedisServiceError: Redis operation failure.
        """
        real_key = resolve_key(key, **kwargs)
        return int(await self.client.llen(real_key))

    @catch_redis_errors
    async def set_index(self, key: "str | BaseRedisKey", index: int, value: str, **kwargs: Any) -> None:
        """Set the element at a specific list index (LSET).

        Args:
            key: Redis key or a ``BaseRedisKey`` instance.
            index: Zero-based index. Negative values count from the end.
            value: New value to set at the given index.
            **kwargs: Extra parameters forwarded to ``resolve_key``.

        Raises:
            RedisConnectionError: Redis connection failure.
            RedisServiceError: Redis operation failure.
        """
        real_key = resolve_key(key, **kwargs)
        await self.client.lset(real_key, index, value)

    @catch_redis_errors
    async def trim(self, key: "str | BaseRedisKey", start: int, end: int, **kwargs: Any) -> None:
        """Trim a list to the specified index range, discarding all other elements (LTRIM).

        Args:
            key: Redis key or a ``BaseRedisKey`` instance.
            start: Start index (inclusive).
            end: End index (inclusive).
            **kwargs: Extra parameters forwarded to ``resolve_key``.

        Raises:
            RedisConnectionError: Redis connection failure.
            RedisServiceError: Redis operation failure.
        """
        real_key = resolve_key(key, **kwargs)
        await self.client.ltrim(real_key, start, end)

    @catch_redis_errors
    async def move(self, src: str, dst: str) -> str | None:
        """Atomically move the last element of ``src`` to the head of ``dst`` (RPOPLPUSH).

        Args:
            src: Source list key.
            dst: Destination list key.

        Returns:
            The moved element, or ``None`` if ``src`` is empty.

        Raises:
            RedisConnectionError: Redis connection failure.
            RedisServiceError: Redis operation failure.
        """
        value = await self.client.rpoplpush(src, dst)
        return str(value) if value is not None else None
Functions
push(key, *values, **kwargs) async

Append one or more values to the end of a list (RPUSH).

Parameters:

Name Type Description Default
key str | BaseRedisKey

Redis key or a BaseRedisKey instance.

 required
*values str

String values to append.

 ()
**kwargs Any

Extra parameters forwarded to resolve_key.

 {}

Returns:

Type Description
int

New length of the list after the operation.

Raises:

Type Description
RedisConnectionError

Redis connection failure.
RedisServiceError

Redis operation failure.
Source code in src/codex_platform/redis_service/operations/list_.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
@catch_redis_errors
async def push(self, key: "str | BaseRedisKey", *values: str, **kwargs: Any) -> int:
    """Append one or more values to the end of a list (RPUSH).

    Args:
        key: Redis key or a ``BaseRedisKey`` instance.
        *values: String values to append.
        **kwargs: Extra parameters forwarded to ``resolve_key``.

    Returns:
        New length of the list after the operation.

    Raises:
        RedisConnectionError: Redis connection failure.
        RedisServiceError: Redis operation failure.
    """
    real_key = resolve_key(key, **kwargs)
    return int(await self.client.rpush(real_key, *values))
lpush(key, *values, **kwargs) async

Prepend one or more values to the beginning of a list (LPUSH).

Parameters:

Name Type Description Default
key str | BaseRedisKey

Redis key or a BaseRedisKey instance.

 required
*values str

String values to prepend.

 ()
**kwargs Any

Extra parameters forwarded to resolve_key.

 {}

Returns:

Type Description
int

New length of the list after the operation.

Raises:

Type Description
RedisConnectionError

Redis connection failure.
RedisServiceError

Redis operation failure.
Source code in src/codex_platform/redis_service/operations/list_.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
@catch_redis_errors
async def lpush(self, key: "str | BaseRedisKey", *values: str, **kwargs: Any) -> int:
    """Prepend one or more values to the beginning of a list (LPUSH).

    Args:
        key: Redis key or a ``BaseRedisKey`` instance.
        *values: String values to prepend.
        **kwargs: Extra parameters forwarded to ``resolve_key``.

    Returns:
        New length of the list after the operation.

    Raises:
        RedisConnectionError: Redis connection failure.
        RedisServiceError: Redis operation failure.
    """
    real_key = resolve_key(key, **kwargs)
    return int(await self.client.lpush(real_key, *values))
pop(key, **kwargs) async

Remove and return the first element of a list (LPOP).

Parameters:

Name Type Description Default
key str | BaseRedisKey

Redis key or a BaseRedisKey instance.

 required
**kwargs Any

Extra parameters forwarded to resolve_key.

 {}

Returns:

Type Description
str | None

The first element, or None if the list is empty or does not exist.

Raises:

Type Description
RedisConnectionError

Redis connection failure.
RedisServiceError

Redis operation failure.
Source code in src/codex_platform/redis_service/operations/list_.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
@catch_redis_errors
async def pop(self, key: "str | BaseRedisKey", **kwargs: Any) -> str | None:
    """Remove and return the first element of a list (LPOP).

    Args:
        key: Redis key or a ``BaseRedisKey`` instance.
        **kwargs: Extra parameters forwarded to ``resolve_key``.

    Returns:
        The first element, or ``None`` if the list is empty or does not exist.

    Raises:
        RedisConnectionError: Redis connection failure.
        RedisServiceError: Redis operation failure.
    """
    real_key = resolve_key(key, **kwargs)
    value = await self.client.lpop(real_key)
    return str(value) if value is not None else None
rpop(key, **kwargs) async

Remove and return the last element of a list (RPOP).

Parameters:

Name Type Description Default
key str | BaseRedisKey

Redis key or a BaseRedisKey instance.

 required
**kwargs Any

Extra parameters forwarded to resolve_key.

 {}

Returns:

Type Description
str | None

The last element, or None if the list is empty or does not exist.

Raises:

Type Description
RedisConnectionError

Redis connection failure.
RedisServiceError

Redis operation failure.
Source code in src/codex_platform/redis_service/operations/list_.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
@catch_redis_errors
async def rpop(self, key: "str | BaseRedisKey", **kwargs: Any) -> str | None:
    """Remove and return the last element of a list (RPOP).

    Args:
        key: Redis key or a ``BaseRedisKey`` instance.
        **kwargs: Extra parameters forwarded to ``resolve_key``.

    Returns:
        The last element, or ``None`` if the list is empty or does not exist.

    Raises:
        RedisConnectionError: Redis connection failure.
        RedisServiceError: Redis operation failure.
    """
    real_key = resolve_key(key, **kwargs)
    value = await self.client.rpop(real_key)
    return str(value) if value is not None else None
brpop(*keys, timeout=0) async

Blocking pop of the last element from the first non-empty list (BRPOP).

Parameters:

Name Type Description Default
*keys str

One or more Redis key strings to check in order.

 ()
timeout int

Maximum seconds to block. 0 blocks indefinitely.

 0

Returns:

Type Description
tuple[str, str] | None

(key, value) tuple when an element is available,
tuple[str, str] | None

or None if the timeout expires.

Raises:

Type Description
RedisConnectionError

Redis connection failure.
RedisServiceError

Redis operation failure.
Source code in src/codex_platform/redis_service/operations/list_.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
@catch_redis_errors
async def brpop(self, *keys: str, timeout: int = 0) -> tuple[str, str] | None:
    """Blocking pop of the last element from the first non-empty list (BRPOP).

    Args:
        *keys: One or more Redis key strings to check in order.
        timeout: Maximum seconds to block. ``0`` blocks indefinitely.

    Returns:
        ``(key, value)`` tuple when an element is available,
        or ``None`` if the timeout expires.

    Raises:
        RedisConnectionError: Redis connection failure.
        RedisServiceError: Redis operation failure.
    """
    result = await self.client.brpop(list(keys), timeout=timeout)
    if result:
        return str(result[0]), str(result[1])
    return None
range(key, start=0, end=-1, **kwargs) async

Return a slice of list elements by index range (LRANGE).

Parameters:

Name Type Description Default
key str | BaseRedisKey

Redis key or a BaseRedisKey instance.

 required
start int

Start index (0-based). Negative values count from the end.

 0
end int

End index inclusive. -1 means the last element.

 -1
**kwargs Any

Extra parameters forwarded to resolve_key.

 {}

Returns:

Type Description
list[str]

List of elements in the requested range.

Raises:

Type Description
RedisConnectionError

Redis connection failure.
RedisServiceError

Redis operation failure.
Source code in src/codex_platform/redis_service/operations/list_.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
@catch_redis_errors
async def range(self, key: "str | BaseRedisKey", start: int = 0, end: int = -1, **kwargs: Any) -> list[str]:
    """Return a slice of list elements by index range (LRANGE).

    Args:
        key: Redis key or a ``BaseRedisKey`` instance.
        start: Start index (0-based). Negative values count from the end.
        end: End index inclusive. ``-1`` means the last element.
        **kwargs: Extra parameters forwarded to ``resolve_key``.

    Returns:
        List of elements in the requested range.

    Raises:
        RedisConnectionError: Redis connection failure.
        RedisServiceError: Redis operation failure.
    """
    real_key = resolve_key(key, **kwargs)
    return await self.client.lrange(real_key, start, end)
length(key, **kwargs) async

Return the number of elements in a list (LLEN).

Parameters:

Name Type Description Default
key str | BaseRedisKey

Redis key or a BaseRedisKey instance.

 required
**kwargs Any

Extra parameters forwarded to resolve_key.

 {}

Returns:

Type Description
int

List length. 0 if the key does not exist.

Raises:

Type Description
RedisConnectionError

Redis connection failure.
RedisServiceError

Redis operation failure.
Source code in src/codex_platform/redis_service/operations/list_.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
@catch_redis_errors
async def length(self, key: "str | BaseRedisKey", **kwargs: Any) -> int:
    """Return the number of elements in a list (LLEN).

    Args:
        key: Redis key or a ``BaseRedisKey`` instance.
        **kwargs: Extra parameters forwarded to ``resolve_key``.

    Returns:
        List length. ``0`` if the key does not exist.

    Raises:
        RedisConnectionError: Redis connection failure.
        RedisServiceError: Redis operation failure.
    """
    real_key = resolve_key(key, **kwargs)
    return int(await self.client.llen(real_key))
set_index(key, index, value, **kwargs) async

Set the element at a specific list index (LSET).

Parameters:

Name Type Description Default
key str | BaseRedisKey

Redis key or a BaseRedisKey instance.

 required
index int

Zero-based index. Negative values count from the end.

 required
value str

New value to set at the given index.

 required
**kwargs Any

Extra parameters forwarded to resolve_key.

 {}

Raises:

Type Description
RedisConnectionError

Redis connection failure.
RedisServiceError

Redis operation failure.
Source code in src/codex_platform/redis_service/operations/list_.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
@catch_redis_errors
async def set_index(self, key: "str | BaseRedisKey", index: int, value: str, **kwargs: Any) -> None:
    """Set the element at a specific list index (LSET).

    Args:
        key: Redis key or a ``BaseRedisKey`` instance.
        index: Zero-based index. Negative values count from the end.
        value: New value to set at the given index.
        **kwargs: Extra parameters forwarded to ``resolve_key``.

    Raises:
        RedisConnectionError: Redis connection failure.
        RedisServiceError: Redis operation failure.
    """
    real_key = resolve_key(key, **kwargs)
    await self.client.lset(real_key, index, value)
trim(key, start, end, **kwargs) async

Trim a list to the specified index range, discarding all other elements (LTRIM).

Parameters:

Name Type Description Default
key str | BaseRedisKey

Redis key or a BaseRedisKey instance.

 required
start int

Start index (inclusive).

 required
end int

End index (inclusive).

 required
**kwargs Any

Extra parameters forwarded to resolve_key.

 {}

Raises:

Type Description
RedisConnectionError

Redis connection failure.
RedisServiceError

Redis operation failure.
Source code in src/codex_platform/redis_service/operations/list_.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
@catch_redis_errors
async def trim(self, key: "str | BaseRedisKey", start: int, end: int, **kwargs: Any) -> None:
    """Trim a list to the specified index range, discarding all other elements (LTRIM).

    Args:
        key: Redis key or a ``BaseRedisKey`` instance.
        start: Start index (inclusive).
        end: End index (inclusive).
        **kwargs: Extra parameters forwarded to ``resolve_key``.

    Raises:
        RedisConnectionError: Redis connection failure.
        RedisServiceError: Redis operation failure.
    """
    real_key = resolve_key(key, **kwargs)
    await self.client.ltrim(real_key, start, end)
move(src, dst) async

Atomically move the last element of src to the head of dst (RPOPLPUSH).

Parameters:

Name Type Description Default
src str

Source list key.

 required
dst str

Destination list key.

 required

Returns:

Type Description
str | None

The moved element, or None if src is empty.

Raises:

Type Description
RedisConnectionError

Redis connection failure.
RedisServiceError

Redis operation failure.
Source code in src/codex_platform/redis_service/operations/list_.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
@catch_redis_errors
async def move(self, src: str, dst: str) -> str | None:
    """Atomically move the last element of ``src`` to the head of ``dst`` (RPOPLPUSH).

    Args:
        src: Source list key.
        dst: Destination list key.

    Returns:
        The moved element, or ``None`` if ``src`` is empty.

    Raises:
        RedisConnectionError: Redis connection failure.
        RedisServiceError: Redis operation failure.
    """
    value = await self.client.rpoplpush(src, dst)
    return str(value) if value is not None else None

Functions