Skip to content

API Reference

Complete API documentation for redis-memory.

Classes

Memory

redis_memory.Memory

A synchronized key-value store that uses Redis as a shared memory backend. If Redis is unavailable, values are cached locally and queued for later syncing.

Environment Variables:
  • REDIS_HOST: Hostname of the Redis server (default: 'localhost')
  • REDIS_PORT: Port of the Redis server (default: 6379)
  • REDIS_PREFIX: Prefix to use for Redis keys (default: 'memory:')
Attributes:

_timeout : float Timeout for Redis operations in seconds (default: 0.5). _queue : list Queue of (key, value) tuples to be synced when Redis becomes available. _attributes : dict Local cache of attributes (always up-to-date with the last set values).

Examples:

os.environ['REDIS_HOST'] = 'localhost' os.environ['REDIS_PORT'] = '6379'

mem1 = Memory() mem1.foo = 42 mem2 = Memory() print(mem2.foo) 42

mem1.bar = {"a": 1} print(mem2.bar)

Source code in redis_memory/__init__.py 
 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
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
class Memory:
    """A synchronized key-value store that uses Redis as a shared memory
    backend. If Redis is unavailable, values are cached locally and
    queued for later syncing.

    Environment Variables:
    ----------------------
    - REDIS_HOST:   Hostname of the Redis server (default: 'localhost')
    - REDIS_PORT:   Port of the Redis server (default: 6379)
    - REDIS_PREFIX: Prefix to use for Redis keys (default: 'memory:')

    Attributes:
    -----------
    _timeout : float
        Timeout for Redis operations in seconds (default: 0.5).
    _queue : list
        Queue of (key, value) tuples to be synced when Redis becomes available.
    _attributes : dict
        Local cache of attributes (always up-to-date with the last set values).

    Examples:
    ---------
    >>> os.environ['REDIS_HOST'] = 'localhost'
    >>> os.environ['REDIS_PORT'] = '6379'

    >>> mem1 = Memory()
    >>> mem1.foo = 42
    >>> mem2 = Memory()
    >>> print(mem2.foo)
    42

    >>> mem1.bar = {"a": 1}
    >>> print(mem2.bar)
    {'a': 1}
    """

    def __init__(
        self,
        redis_hostname: str = "redis",
        redis_port: int = 6379,
        redis_prefix: str = "memory:",
    ):
        """Initialize the Memory instance and flush any queued
        updates."""
        self._host = os.environ.get("REDIS_HOST", redis_hostname)
        self._port = int(os.environ.get("REDIS_PORT", redis_port))
        self._prefix = redis_prefix
        self._timeout = 0.5  # Seconds

        self._queue = []
        self._attributes = {}
        self._last_modified = {}  # Track last modified timestamps

        self._stop_event = threading.Event()
        self._thread = None

        self._redis_available = False  # owned exclusively by the background thread
        self._is_connected_to_redis_at_least_once = False

        self.start_background_flush()

        self._load_from_redis()

    def __enter__(self):
        """Enter the runtime context related to this object."""
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Flush any remaining queue then stop the background thread."""
        if self._queue:
            while True:
                try:
                    self._flush_queue()
                    break
                except Exception:
                    if not self._is_connected_to_redis_at_least_once:
                        break  # Never connected; give up.
                    time.sleep(1)

        self.stop_background_flush()

    def _connect(self):
        """Establish a new Redis connection.

        Returns:
            redis.Redis or None: A Redis client if connection works;
            otherwise None.
        """
        client = redis.Redis(
            host=self._host,
            port=self._port,
            socket_connect_timeout=self._timeout,
            socket_timeout=self._timeout,
        )
        client.ping()
        return client

    def _key(self, name):
        """Generate Redis key with prefix."""
        return f"{self._prefix}{name}"

    def _flush_queue(self):
        """Attempt one flush of the queue to Redis.

        Connects once; raises on failure so the caller can apply
        backoff. For each queued item, skips writes where Redis already
        holds a newer timestamp.
        """
        client = self._connect()  # raises on failure
        self._is_connected_to_redis_at_least_once = True

        while self._queue:
            key, payload = self._queue.pop(0)
            value = payload.get("value")
            queued_timestamp = payload.get("last_modified")

            raw = client.get(self._key(key))
            redis_timestamp = 0
            if raw is not None:
                obj = json.loads(raw)
                redis_timestamp = obj.get("last_modified", 0)

            if redis_timestamp > queued_timestamp:
                continue

            if value is None:
                client.delete(self._key(key))
            else:
                client.set(self._key(key), json.dumps(payload))

    def _background_flush_loop(self):
        """Own the Redis connection state and flush the write queue.

        On each cycle this thread attempts to connect and flush any
        queued writes. Success sets _redis_available = True and resets
        the backoff to 1 s. Failure sets _redis_available = False and
        doubles the wait (capped at 30 s), so the main thread can fall
        back to local cache without blocking.
        """
        backoff = 1
        while not self._stop_event.is_set():
            try:
                self._flush_queue()  # raises on connection failure
                self._redis_available = True
                backoff = 1
            except Exception:
                self._redis_available = False
                backoff = min(backoff * 2, 30)
                self._stop_event.wait(backoff)
                continue
            self._stop_event.wait(1)

    def start_background_flush(self):
        """Start the background thread to flush the queue regularly."""
        if self._thread is None or not self._thread.is_alive():
            self._stop_event.clear()
            self._thread = threading.Thread(
                target=self._background_flush_loop, daemon=True
            )
            self._thread.start()

    def stop_background_flush(self):
        """Stop the background flushing thread."""
        self._stop_event.set()
        if self._thread is not None:
            self._thread.join()
            self._thread = None

    def _load_from_redis(self):
        """Load all keys with the current prefix from Redis into local
        cache."""
        try:
            client = self._connect()
            self._redis_available = True
            self._is_connected_to_redis_at_least_once = True
            pattern = f"{self._prefix}*"
            for key in client.scan_iter(match=pattern):
                name = key.decode().replace(self._prefix, "", 1)
                try:
                    raw = client.get(key)
                    obj = json.loads(raw)
                    if (
                        isinstance(obj, dict)
                        and "value" in obj
                        and "last_modified" in obj
                    ):
                        self._attributes[name] = obj["value"]
                        self._last_modified[name] = obj["last_modified"]
                    else:
                        self._attributes[name] = obj
                except Exception:
                    pass
        except Exception:
            self._redis_available = False

    def __setattr__(self, name, value):
        """Set an attribute.

        Store in Redis if available, otherwise queue it.

        Raises:
            ValueError: If the value is not serializable.
        """
        if name.startswith("_"):
            super().__setattr__(name, value)
            return

        try:
            _ = json.dumps(value)
        except json.JSONDecodeError:
            raise

        self._set(name, value)

    def _write_to_redis_or_queue(self, name: str, payload: dict):
        """Write to Redis if available; queue for later if not.

        On a connection failure the flag is cleared so subsequent calls
        skip the attempt until the background thread re-establishes the
        connection.
        """
        if self._redis_available:
            try:
                client = self._connect()
                client.set(self._key(name), json.dumps(payload))
                self._is_connected_to_redis_at_least_once = True
                return
            except Exception:
                self._redis_available = False
        self._queue.append((name, payload))

    def _set(self, name: str, value: Any):
        """Update local cache and write through to Redis (or queue if
        down)."""
        self._attributes[name] = wrap_sync(value, self, name)
        timestamp = time.time_ns()
        self._last_modified[name] = timestamp
        self._write_to_redis_or_queue(
            name, {"value": value, "last_modified": timestamp}
        )

    def __getattr__(self, name: str) -> Any:
        """Return the attribute, reading from Redis when available.

        Falls back to local cache when Redis is down so the main thread
        never blocks.

        Raises:
            AttributeError: If the attribute has not been set.
        """
        if name.startswith("_"):
            return super().__getattribute__(name)

        if self._redis_available:
            try:
                client = self._connect()
                raw = client.get(self._key(name))
                if raw is not None:
                    self._is_connected_to_redis_at_least_once = True
                    obj = json.loads(raw)
                    value = obj["value"]
                    self._attributes[name] = wrap_sync(value, self, name)
                    self._last_modified[name] = obj["last_modified"]
                    return self._attributes[name]
                # Redis is up but key not there yet (still in queue);
                # fall through to local cache.
            except Exception:
                self._redis_available = False

        if name in self._attributes:
            return self._attributes[name]
        raise AttributeError(f"'Memory' object has no attribute '{name}'")

    def sync(self, name: str):
        """Write the current local value of an attribute to Redis (or
        queue if down)."""
        if name not in self._attributes:
            raise AttributeError(f"'Memory' object has no attribute '{name}'")

        self._write_to_redis_or_queue(
            name,
            {
                "value": self._attributes[name],
                "last_modified": self._last_modified.get(name, 0),
            },
        )

    def __delattr__(self, name):
        """Remove an attribute from local cache and queue its deletion.

        Raises:
            AttributeError: If the attribute is not found.
        """
        if name.startswith("_"):
            super().__delattr__(name)
            return

        if name not in self._attributes:
            raise AttributeError(f"'Memory' object has no attribute '{name}'")

        del self._attributes[name]
        if name in self._last_modified:
            del self._last_modified[name]

        timestamp = time.time_ns()
        if self._redis_available:
            try:
                client = self._connect()
                client.delete(self._key(name))
                self._is_connected_to_redis_at_least_once = True
                return
            except Exception:
                self._redis_available = False
        self._queue.append((name, {"value": None, "last_modified": timestamp}))

__delattr__(name)

Remove an attribute from local cache and queue its deletion.

Raises:

Type Description
AttributeError

If the attribute is not found.
Source code in redis_memory/__init__.py 
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
def __delattr__(self, name):
    """Remove an attribute from local cache and queue its deletion.

    Raises:
        AttributeError: If the attribute is not found.
    """
    if name.startswith("_"):
        super().__delattr__(name)
        return

    if name not in self._attributes:
        raise AttributeError(f"'Memory' object has no attribute '{name}'")

    del self._attributes[name]
    if name in self._last_modified:
        del self._last_modified[name]

    timestamp = time.time_ns()
    if self._redis_available:
        try:
            client = self._connect()
            client.delete(self._key(name))
            self._is_connected_to_redis_at_least_once = True
            return
        except Exception:
            self._redis_available = False
    self._queue.append((name, {"value": None, "last_modified": timestamp}))

__enter__()

Enter the runtime context related to this object.

Source code in redis_memory/__init__.py 
156
157
158
def __enter__(self):
    """Enter the runtime context related to this object."""
    return self

__exit__(exc_type, exc_val, exc_tb)

Flush any remaining queue then stop the background thread.

Source code in redis_memory/__init__.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
def __exit__(self, exc_type, exc_val, exc_tb):
    """Flush any remaining queue then stop the background thread."""
    if self._queue:
        while True:
            try:
                self._flush_queue()
                break
            except Exception:
                if not self._is_connected_to_redis_at_least_once:
                    break  # Never connected; give up.
                time.sleep(1)

    self.stop_background_flush()

__getattr__(name)

Return the attribute, reading from Redis when available.

Falls back to local cache when Redis is down so the main thread never blocks.

Raises:

Type Description
AttributeError

If the attribute has not been set.
Source code in redis_memory/__init__.py 
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
def __getattr__(self, name: str) -> Any:
    """Return the attribute, reading from Redis when available.

    Falls back to local cache when Redis is down so the main thread
    never blocks.

    Raises:
        AttributeError: If the attribute has not been set.
    """
    if name.startswith("_"):
        return super().__getattribute__(name)

    if self._redis_available:
        try:
            client = self._connect()
            raw = client.get(self._key(name))
            if raw is not None:
                self._is_connected_to_redis_at_least_once = True
                obj = json.loads(raw)
                value = obj["value"]
                self._attributes[name] = wrap_sync(value, self, name)
                self._last_modified[name] = obj["last_modified"]
                return self._attributes[name]
            # Redis is up but key not there yet (still in queue);
            # fall through to local cache.
        except Exception:
            self._redis_available = False

    if name in self._attributes:
        return self._attributes[name]
    raise AttributeError(f"'Memory' object has no attribute '{name}'")

__init__(redis_hostname='redis', redis_port=6379, redis_prefix='memory:')

Initialize the Memory instance and flush any queued updates.

Source code in redis_memory/__init__.py 
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
def __init__(
    self,
    redis_hostname: str = "redis",
    redis_port: int = 6379,
    redis_prefix: str = "memory:",
):
    """Initialize the Memory instance and flush any queued
    updates."""
    self._host = os.environ.get("REDIS_HOST", redis_hostname)
    self._port = int(os.environ.get("REDIS_PORT", redis_port))
    self._prefix = redis_prefix
    self._timeout = 0.5  # Seconds

    self._queue = []
    self._attributes = {}
    self._last_modified = {}  # Track last modified timestamps

    self._stop_event = threading.Event()
    self._thread = None

    self._redis_available = False  # owned exclusively by the background thread
    self._is_connected_to_redis_at_least_once = False

    self.start_background_flush()

    self._load_from_redis()

__setattr__(name, value)

Set an attribute.

Store in Redis if available, otherwise queue it.

Raises:

Type Description
ValueError

If the value is not serializable.
Source code in redis_memory/__init__.py 
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
def __setattr__(self, name, value):
    """Set an attribute.

    Store in Redis if available, otherwise queue it.

    Raises:
        ValueError: If the value is not serializable.
    """
    if name.startswith("_"):
        super().__setattr__(name, value)
        return

    try:
        _ = json.dumps(value)
    except json.JSONDecodeError:
        raise

    self._set(name, value)

start_background_flush()

Start the background thread to flush the queue regularly.

Source code in redis_memory/__init__.py 
245
246
247
248
249
250
251
252
def start_background_flush(self):
    """Start the background thread to flush the queue regularly."""
    if self._thread is None or not self._thread.is_alive():
        self._stop_event.clear()
        self._thread = threading.Thread(
            target=self._background_flush_loop, daemon=True
        )
        self._thread.start()

stop_background_flush()

Stop the background flushing thread.

Source code in redis_memory/__init__.py 
254
255
256
257
258
259
def stop_background_flush(self):
    """Stop the background flushing thread."""
    self._stop_event.set()
    if self._thread is not None:
        self._thread.join()
        self._thread = None

sync(name)

Write the current local value of an attribute to Redis (or queue if down).

Source code in redis_memory/__init__.py 
366
367
368
369
370
371
372
373
374
375
376
377
378
def sync(self, name: str):
    """Write the current local value of an attribute to Redis (or
    queue if down)."""
    if name not in self._attributes:
        raise AttributeError(f"'Memory' object has no attribute '{name}'")

    self._write_to_redis_or_queue(
        name,
        {
            "value": self._attributes[name],
            "last_modified": self._last_modified.get(name, 0),
        },
    )

ConversationMemory

redis_memory.ConversationMemory

Bases: Memory

Memory subclass that namespaces keys by conversation ID.

Parameters:

Name Type Description Default
conversation_id str

Unique ID to isolate this conversation's memory.

 required
redis_hostname str

Redis host (default 'redis').

 'redis'
redis_port int

Redis port (default 6379).

 6379
redis_prefix str

Base prefix for keys (default 'memory:').

 'memory:'
Source code in redis_memory/__init__.py 
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
class ConversationMemory(Memory):
    """Memory subclass that namespaces keys by conversation ID.

    Args:
        conversation_id (str): Unique ID to isolate this conversation's memory.
        redis_hostname (str): Redis host (default 'redis').
        redis_port (int): Redis port (default 6379).
        redis_prefix (str): Base prefix for keys (default 'memory:').
    """

    def __init__(
        self,
        conversation_id: str,
        redis_hostname="redis",
        redis_port=6379,
        redis_prefix="memory:",
    ):
        self._conversation_id = conversation_id
        super().__init__(
            redis_hostname=redis_hostname,
            redis_port=redis_port,
            redis_prefix=redis_prefix,
        )

    def _key(self, name):
        # Override to include conversation_id in the Redis key
        return f"{self._prefix}{self._conversation_id}:{name}"

SyncedList

redis_memory.SyncedList

Bases: list

Source code in redis_memory/__init__.py 
12
13
14
15
16
17
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
class SyncedList(list):

    def __init__(self, iterable, parent, topmost_key: str):
        for i, item in enumerate(iterable):
            iterable[i] = wrap_sync(item, self, f"{topmost_key}[{i}]")
        super().__init__(iterable)
        self._parent = parent
        self._topmost_key = topmost_key

    def append(self, item):
        super().append(item)
        self._parent.sync(self._topmost_key)

    def extend(self, iterable):
        super().extend(iterable)
        self._parent.sync(self._topmost_key)

    def insert(self, index, item):
        super().insert(index, item)
        self._parent.sync(self._topmost_key)

    def sync(self, name: str):
        self._parent.sync(self._topmost_key)

    def aslist(self):
        """Return a plain Python list, recursively converting nested
        SyncedList/SyncedDict objects."""
        result = []
        for item in self:
            if isinstance(item, SyncedList):
                result.append(item.aslist())
            elif isinstance(item, SyncedDict):
                result.append(item.asdict())
            else:
                result.append(item)
        return result

aslist()

Return a plain Python list, recursively converting nested SyncedList/SyncedDict objects.

Source code in redis_memory/__init__.py 
36
37
38
39
40
41
42
43
44
45
46
47
def aslist(self):
    """Return a plain Python list, recursively converting nested
    SyncedList/SyncedDict objects."""
    result = []
    for item in self:
        if isinstance(item, SyncedList):
            result.append(item.aslist())
        elif isinstance(item, SyncedDict):
            result.append(item.asdict())
        else:
            result.append(item)
    return result

SyncedDict

redis_memory.SyncedDict

Bases: dict

Source code in redis_memory/__init__.py 
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
class SyncedDict(dict):

    def __init__(self, mapping, parent, topmost_key: str):
        for k, v in mapping.items():
            mapping[k] = wrap_sync(v, self, topmost_key)
        super().__init__(mapping)
        self._parent = parent
        self._topmost_key = topmost_key

    def __setitem__(self, k, v):
        super().__setitem__(k, v)
        self._parent.sync(self._topmost_key)

    def update(self, *args, **kwargs):
        super().update(*args, **kwargs)
        self._parent.sync(self._topmost_key)

    def sync(self, name: str):
        self._parent.sync(self._topmost_key)

    def asdict(self):
        """Return a plain Python dict, recursively converting nested
        SyncedList/SyncedDict objects."""
        result = {}
        for key, value in self.items():
            if isinstance(value, SyncedList):
                result[key] = value.aslist()
            elif isinstance(value, SyncedDict):
                result[key] = value.asdict()
            else:
                result[key] = value
        return result

asdict()

Return a plain Python dict, recursively converting nested SyncedList/SyncedDict objects.

Source code in redis_memory/__init__.py 
70
71
72
73
74
75
76
77
78
79
80
81
def asdict(self):
    """Return a plain Python dict, recursively converting nested
    SyncedList/SyncedDict objects."""
    result = {}
    for key, value in self.items():
        if isinstance(value, SyncedList):
            result[key] = value.aslist()
        elif isinstance(value, SyncedDict):
            result[key] = value.asdict()
        else:
            result[key] = value
    return result

Environment Variables

Variable Default Description
REDIS_HOST redis Redis server hostname
REDIS_PORT 6379 Redis server port
REDIS_PREFIX memory: Key prefix for namespacing

Exceptions

redis-memory uses standard Python exceptions and Redis exceptions:

  • AttributeError: Raised when accessing non-existent attributes
  • redis.ConnectionError: Raised when Redis connection fails
  • redis.TimeoutError: Raised when Redis operations timeout

Type Hints

redis-memory supports type hints:

from redis_memory import Memory
from typing import List, Dict, Any

mem: Memory = Memory()
mem.items: List[int] = [1, 2, 3]
mem.config: Dict[str, Any] = {"key": "value"}