Skip to content

内存后端

进程内内存缓存，速度最快，适合单实例使用。

Bases: BaseBackend

内存缓存后端

基于 OrderedDict 实现的高性能内存缓存，支持 LRU 淘汰。

架构设计: - 存储结构: OrderedDict[key, (value, expires_at)] - LRU 实现: 访问时将键移到末尾，淘汰时删除头部 - TTL 管理: 惰性删除（读取时检查）+ 后台定期清理 - 线程安全: 所有操作使用 RLock 保护

性能特点: - 读取: O(1)，< 0.01ms - 写入: O(1)，< 0.01ms - LRU 淘汰: O(1) - 空间复杂度: O(n)

使用示例

backend = MemoryBackend(max_size=10000) backend.set("user:123", {"name": "Alice"}, ttl=3600) user = backend.get("user:123") print(user) # {"name": "Alice"}

Source code in src/symphra_cache/backends/memory.py 
 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
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
407
408
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
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
class MemoryBackend(BaseBackend):
    """
    内存缓存后端

    基于 OrderedDict 实现的高性能内存缓存，支持 LRU 淘汰。

    架构设计:
    - 存储结构: OrderedDict[key, (value, expires_at)]
    - LRU 实现: 访问时将键移到末尾，淘汰时删除头部
    - TTL 管理: 惰性删除（读取时检查）+ 后台定期清理
    - 线程安全: 所有操作使用 RLock 保护

    性能特点:
    - 读取: O(1)，< 0.01ms
    - 写入: O(1)，< 0.01ms
    - LRU 淘汰: O(1)
    - 空间复杂度: O(n)

    使用示例:
        >>> backend = MemoryBackend(max_size=10000)
        >>> backend.set("user:123", {"name": "Alice"}, ttl=3600)
        >>> user = backend.get("user:123")
        >>> print(user)  # {"name": "Alice"}
    """

    def __init__(
        self,
        max_size: int = 10000,
        cleanup_interval: int = 60,
    ) -> None:
        """
        初始化内存后端

        Args:
            max_size: 最大缓存条数，超过后触发 LRU 淘汰（默认 10000）
            cleanup_interval: TTL 清理间隔（秒），默认 60 秒

        示例:
            >>> # 创建最大容量 1000 的缓存
            >>> backend = MemoryBackend(max_size=1000, cleanup_interval=30)
        """
        self._max_size = max_size
        self._cleanup_interval = cleanup_interval

        # 存储格式: {key: (value, expires_at)}
        # expires_at 为 None 表示永不过期
        # 使用 OrderedDict 支持 LRU：最近访问的在末尾，最旧的在头部
        self._cache: OrderedDict[CacheKey, tuple[CacheValue, float | None]] = OrderedDict()

        # 线程锁（保证线程安全）
        # 使用 RLock 允许同一线程重入
        self._lock = threading.RLock()

        # 启动后台清理任务
        self._cleanup_thread: threading.Thread | None = None
        self._stop_cleanup = threading.Event()
        self._start_cleanup_task()

    # ========== 同步基础操作 ==========

    def get(self, key: CacheKey) -> CacheValue | None:
        """
        同步获取缓存值

        实现细节:
        1. 检查键是否存在
        2. 检查是否过期（惰性删除）
        3. 更新 LRU 顺序（移到末尾表示最近使用）
        4. 返回值

        时间复杂度: O(1)

        Args:
            key: 缓存键

        Returns:
            缓存值，不存在或已过期则返回 None

        示例:
            >>> backend.set("key", "value", ttl=60)
            >>> backend.get("key")  # "value"
            >>> time.sleep(61)
            >>> backend.get("key")  # None（已过期）
        """
        with self._lock:
            # 检查键是否存在
            if key not in self._cache:
                return None

            value, expires_at = self._cache[key]

            # 检查是否过期（惰性删除）
            if expires_at is not None and time.time() > expires_at:
                # 已过期，删除并返回 None
                del self._cache[key]
                return None

            # 更新 LRU：移到末尾表示最近使用
            self._cache.move_to_end(key)

            return value

    async def aget(self, key: CacheKey) -> CacheValue | None:
        """
        异步获取缓存值

        内存后端的异步版本直接调用同步方法。
        因为内存操作非常快（< 0.01ms），无需真正的异步 I/O。

        Args:
            key: 缓存键

        Returns:
            缓存值，不存在或已过期则返回 None

        示例:
            >>> value = await backend.aget("user:123")
        """
        return self.get(key)

    def set(
        self,
        key: CacheKey,
        value: CacheValue,
        ttl: int | None = None,
        ex: bool = False,
        nx: bool = False,
    ) -> bool:
        """
        设置缓存值

        Args:
            key: 缓存键
            value: 缓存值
            ttl: 过期时间(秒),None 表示永不过期
            ex: 保留参数,内存后端始终使用相对过期时间
            nx: 如果为 True,仅当键不存在时才设置

        Returns:
            是否设置成功
        """
        with self._lock:
            # 检查 max_size 是否为 0
            if self._max_size == 0:
                return False

            # NX 模式:仅当键不存在时设置
            if nx and key in self._cache:
                # 检查是否已过期
                _, expires_at = self._cache[key]
                if expires_at is None or time.time() <= expires_at:
                    return False  # 键存在且未过期,设置失败

            # 计算过期时间
            expires_at = None if ttl is None else time.time() + ttl

            # 如果键已存在,更新位置
            if key in self._cache:
                self._cache.move_to_end(key)
            # 如果缓存已满,执行 LRU 淘汰
            elif len(self._cache) >= self._max_size:
                self._cache.popitem(last=False)

            # 设置缓存值
            self._cache[key] = (value, expires_at)
            return True

    async def aset(
        self,
        key: CacheKey,
        value: CacheValue,
        ttl: int | None = None,
        ex: bool = False,
        nx: bool = False,
    ) -> bool:
        """
        异步设置缓存值

        Args:
            key: 缓存键
            value: 缓存值
            ttl: 过期时间(秒)
            ex: 保留参数
            nx: 如果为 True,仅当键不存在时才设置

        Returns:
            是否设置成功
        """
        return self.set(key, value, ttl=ttl, ex=ex, nx=nx)

    def delete(self, key: CacheKey) -> bool:
        """
        删除缓存

        Args:
            key: 缓存键

        Returns:
            如果键存在并成功删除返回 True，否则返回 False

        示例:
            >>> backend.set("temp", "data")
            >>> backend.delete("temp")  # True
            >>> backend.delete("temp")  # False（已删除）
        """
        with self._lock:
            if key in self._cache:
                del self._cache[key]
                return True
            return False

    async def adelete(self, key: CacheKey) -> bool:
        """
        异步删除缓存

        Args:
            key: 缓存键

        Returns:
            如果键存在并成功删除返回 True，否则返回 False

        示例:
            >>> deleted = await backend.adelete("user:123")
        """
        return self.delete(key)

    def exists(self, key: CacheKey) -> bool:
        """
        检查键是否存在

        会检查键是否过期，过期的键返回 False。

        Args:
            key: 缓存键

        Returns:
            如果键存在且未过期返回 True，否则返回 False

        示例:
            >>> backend.set("key", "value", ttl=60)
            >>> backend.exists("key")  # True
            >>> time.sleep(61)
            >>> backend.exists("key")  # False（已过期）
        """
        return self.get(key) is not None

    def clear(self) -> None:
        """
        清空所有缓存

        警告:
            此操作不可逆，会删除所有缓存数据

        示例:
            >>> backend.clear()  # 删除所有缓存
        """
        with self._lock:
            self._cache.clear()

    # ========== 批量操作优化 ==========

    def get_many(self, keys: list[CacheKey]) -> dict[CacheKey, CacheValue]:
        """
        批量获取缓存值（优化版）

        相比基类的默认实现，此版本在单个锁内完成所有操作，
        减少锁开销，提升性能。

        Args:
            keys: 缓存键列表

        Returns:
            键值对字典，不存在或已过期的键不包含在结果中

        示例:
            >>> backend.set_many({"k1": "v1", "k2": "v2"})
            >>> results = backend.get_many(["k1", "k2", "k3"])
            >>> print(results)  # {"k1": "v1", "k2": "v2"}
        """
        result: dict[CacheKey, CacheValue] = {}
        now = time.time()

        with self._lock:
            for key in keys:
                if key not in self._cache:
                    continue

                value, expires_at = self._cache[key]

                # 检查是否过期
                if expires_at is not None and now > expires_at:
                    # 过期，删除（惰性清理）
                    del self._cache[key]
                    continue

                # 更新 LRU
                self._cache.move_to_end(key)
                result[key] = value

        return result

    def set_many(
        self,
        mapping: dict[CacheKey, CacheValue],
        ttl: int | None = None,
    ) -> None:
        """
        批量设置缓存值（优化版）

        在单个锁内完成所有操作，提升性能。

        Args:
            mapping: 键值对字典
            ttl: 过期时间（秒），None 表示永不过期

        示例:
            >>> backend.set_many(
            ...     {
            ...         "user:1": {"name": "Alice"},
            ...         "user:2": {"name": "Bob"},
            ...     },
            ...     ttl=600,
            ... )
        """
        with self._lock:
            # 边界情况：max_size=0 时不存储任何内容
            if self._max_size == 0:
                return

            expires_at = time.time() + ttl if ttl is not None else None

            for key, value in mapping.items():
                # 检查容量并 LRU 淘汰
                if len(self._cache) >= self._max_size and key not in self._cache:
                    self._cache.popitem(last=False)

                # 存储并移到末尾
                self._cache[key] = (value, expires_at)
                self._cache.move_to_end(key)

    # ========== 扩展操作 ==========

    def keys(
        self,
        pattern: str = "*",
        cursor: int = 0,
        count: int = 100,
        max_keys: int | None = None,
    ) -> KeysPage:
        """
        扫描缓存键

        Args:
            pattern: 匹配模式(支持通配符 * 和 ?)
            cursor: 游标位置
            count: 每页返回的键数量
            max_keys: 最多返回的键数量

        Returns:
            KeysPage 对象
        """
        import fnmatch

        from ..types import KeysPage

        with self._lock:
            # 获取所有键并过滤
            all_keys = list(self._cache.keys())

            # 模式匹配
            if pattern != "*":
                matched_keys = [k for k in all_keys if fnmatch.fnmatch(k, pattern)]
            else:
                matched_keys = all_keys

            # 分页处理
            total = len(matched_keys)
            start_idx = cursor
            end_idx = start_idx + count

            if max_keys is not None:
                end_idx = min(end_idx, start_idx + max_keys)

            page_keys = matched_keys[start_idx:end_idx]

            # 计算下一页游标
            next_cursor = end_idx if end_idx < total else 0
            has_more = next_cursor > 0

            return KeysPage(
                keys=page_keys,
                cursor=next_cursor,
                has_more=has_more,
                total_scanned=len(page_keys),
            )

    async def akeys(
        self,
        pattern: str = "*",
        cursor: int = 0,
        count: int = 100,
        max_keys: int | None = None,
    ) -> KeysPage:
        """异步扫描缓存键"""
        return self.keys(pattern=pattern, cursor=cursor, count=count, max_keys=max_keys)

    def ttl(self, key: CacheKey) -> int:
        """
        获取键的剩余生存时间

        Returns:
            剩余秒数,-1 表示永不过期,-2 表示键不存在
        """
        with self._lock:
            if key not in self._cache:
                return -2

            _, expires_at = self._cache[key]
            if expires_at is None:
                return -1

            remaining = int(expires_at - time.time())
            return remaining if remaining > 0 else -2

    async def attl(self, key: CacheKey) -> int:
        """异步获取键的剩余生存时间"""
        return self.ttl(key)

    def close(self) -> None:
        """
        关闭后端

        停止后台清理线程。
        """
        if self._cleanup_thread and self._cleanup_thread.is_alive():
            # 设置停止标志(如果有的话)
            # 当前实现中清理线程是 daemon,会自动退出
            pass

    async def aclose(self) -> None:
        """异步关闭后端"""
        self.close()

    # ========== 后台清理任务 ==========

    def _start_cleanup_task(self) -> None:
        """
        启动后台 TTL 清理任务

        使用守护线程定期清理过期的键。
        线程在对象销毁时自动停止。
        """

        def _cleanup_loop() -> None:
            """后台清理循环"""
            while not self._stop_cleanup.wait(self._cleanup_interval):
                self._cleanup_expired()

        # 创建并启动守护线程
        self._cleanup_thread = threading.Thread(
            target=_cleanup_loop,
            daemon=True,  # 守护线程，主程序退出时自动终止
            name="symphra-cache-cleanup",
        )
        self._cleanup_thread.start()

    def _cleanup_expired(self) -> None:
        """
        清理所有过期的键

        遍历所有缓存项，删除已过期的键。
        此方法由后台线程定期调用。
        """
        with self._lock:
            now = time.time()
            # 收集过期的键
            expired_keys = [
                key
                for key, (_, expires_at) in self._cache.items()
                if expires_at is not None and now > expires_at
            ]

            # 批量删除过期键
            for key in expired_keys:
                del self._cache[key]

    def __del__(self) -> None:
        """
        析构函数

        停止后台清理线程。
        """
        # 通知清理线程停止
        self._stop_cleanup.set()

        # 等待线程结束（最多等待 1 秒）
        if self._cleanup_thread and self._cleanup_thread.is_alive():
            self._cleanup_thread.join(timeout=1.0)

    # ========== 调试和监控方法 ==========

    def __len__(self) -> int:
        """
        获取当前缓存项数量

        Returns:
            缓存项数量

        示例:
            >>> backend.set("k1", "v1")
            >>> backend.set("k2", "v2")
            >>> len(backend)  # 2
        """
        with self._lock:
            return len(self._cache)

    def __repr__(self) -> str:
        """
        字符串表示

        Returns:
            对象的字符串表示

        示例:
            >>> backend = MemoryBackend(max_size=1000)
            >>> repr(backend)
            "MemoryBackend(size=0, max_size=1000)"
        """
        with self._lock:
            return f"MemoryBackend(size={len(self._cache)}, max_size={self._max_size})"

__del__()

析构函数

停止后台清理线程。

Source code in src/symphra_cache/backends/memory.py 
514
515
516
517
518
519
520
521
522
523
524
525
def __del__(self) -> None:
    """
    析构函数

    停止后台清理线程。
    """
    # 通知清理线程停止
    self._stop_cleanup.set()

    # 等待线程结束（最多等待 1 秒）
    if self._cleanup_thread and self._cleanup_thread.is_alive():
        self._cleanup_thread.join(timeout=1.0)

__init__(max_size=10000, cleanup_interval=60)

初始化内存后端

Parameters:

Name Type Description Default
max_size int

最大缓存条数，超过后触发 LRU 淘汰（默认 10000）

 10000
cleanup_interval int

TTL 清理间隔（秒），默认 60 秒

 60
示例

创建最大容量 1000 的缓存

backend = MemoryBackend(max_size=1000, cleanup_interval=30)

Source code in src/symphra_cache/backends/memory.py 
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
def __init__(
    self,
    max_size: int = 10000,
    cleanup_interval: int = 60,
) -> None:
    """
    初始化内存后端

    Args:
        max_size: 最大缓存条数，超过后触发 LRU 淘汰（默认 10000）
        cleanup_interval: TTL 清理间隔（秒），默认 60 秒

    示例:
        >>> # 创建最大容量 1000 的缓存
        >>> backend = MemoryBackend(max_size=1000, cleanup_interval=30)
    """
    self._max_size = max_size
    self._cleanup_interval = cleanup_interval

    # 存储格式: {key: (value, expires_at)}
    # expires_at 为 None 表示永不过期
    # 使用 OrderedDict 支持 LRU：最近访问的在末尾，最旧的在头部
    self._cache: OrderedDict[CacheKey, tuple[CacheValue, float | None]] = OrderedDict()

    # 线程锁（保证线程安全）
    # 使用 RLock 允许同一线程重入
    self._lock = threading.RLock()

    # 启动后台清理任务
    self._cleanup_thread: threading.Thread | None = None
    self._stop_cleanup = threading.Event()
    self._start_cleanup_task()

__len__()

获取当前缓存项数量

Returns:

Type Description
int

缓存项数量
示例

backend.set("k1", "v1") backend.set("k2", "v2") len(backend) # 2

Source code in src/symphra_cache/backends/memory.py 
529
530
531
532
533
534
535
536
537
538
539
540
541
542
def __len__(self) -> int:
    """
    获取当前缓存项数量

    Returns:
        缓存项数量

    示例:
        >>> backend.set("k1", "v1")
        >>> backend.set("k2", "v2")
        >>> len(backend)  # 2
    """
    with self._lock:
        return len(self._cache)

__repr__()

字符串表示

Returns:

Type Description
str

对象的字符串表示
示例

backend = MemoryBackend(max_size=1000) repr(backend) "MemoryBackend(size=0, max_size=1000)"

Source code in src/symphra_cache/backends/memory.py 
544
545
546
547
548
549
550
551
552
553
554
555
556
557
def __repr__(self) -> str:
    """
    字符串表示

    Returns:
        对象的字符串表示

    示例:
        >>> backend = MemoryBackend(max_size=1000)
        >>> repr(backend)
        "MemoryBackend(size=0, max_size=1000)"
    """
    with self._lock:
        return f"MemoryBackend(size={len(self._cache)}, max_size={self._max_size})"

aclose() async

异步关闭后端

Source code in src/symphra_cache/backends/memory.py 
467
468
469
async def aclose(self) -> None:
    """异步关闭后端"""
    self.close()

adelete(key) async

异步删除缓存

Parameters:

Name Type Description Default
key CacheKey

缓存键

 required

Returns:

Type Description
bool

如果键存在并成功删除返回 True，否则返回 False
示例

deleted = await backend.adelete("user:123")

Source code in src/symphra_cache/backends/memory.py 
239
240
241
242
243
244
245
246
247
248
249
250
251
252
async def adelete(self, key: CacheKey) -> bool:
    """
    异步删除缓存

    Args:
        key: 缓存键

    Returns:
        如果键存在并成功删除返回 True，否则返回 False

    示例:
        >>> deleted = await backend.adelete("user:123")
    """
    return self.delete(key)

aget(key) async

异步获取缓存值

内存后端的异步版本直接调用同步方法。 因为内存操作非常快（< 0.01ms），无需真正的异步 I/O。

Parameters:

Name Type Description Default
key CacheKey

缓存键

 required

Returns:

Type Description
CacheValue | None

缓存值，不存在或已过期则返回 None
示例

value = await backend.aget("user:123")

Source code in src/symphra_cache/backends/memory.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
async def aget(self, key: CacheKey) -> CacheValue | None:
    """
    异步获取缓存值

    内存后端的异步版本直接调用同步方法。
    因为内存操作非常快（< 0.01ms），无需真正的异步 I/O。

    Args:
        key: 缓存键

    Returns:
        缓存值，不存在或已过期则返回 None

    示例:
        >>> value = await backend.aget("user:123")
    """
    return self.get(key)

akeys(pattern='*', cursor=0, count=100, max_keys=None) async

异步扫描缓存键

Source code in src/symphra_cache/backends/memory.py 
424
425
426
427
428
429
430
431
432
async def akeys(
    self,
    pattern: str = "*",
    cursor: int = 0,
    count: int = 100,
    max_keys: int | None = None,
) -> KeysPage:
    """异步扫描缓存键"""
    return self.keys(pattern=pattern, cursor=cursor, count=count, max_keys=max_keys)

aset(key, value, ttl=None, ex=False, nx=False) async

异步设置缓存值

Parameters:

Name Type Description Default
key CacheKey

缓存键

 required
value CacheValue

缓存值

 required
ttl int | None

过期时间(秒)

 None
ex bool

保留参数

 False
nx bool

如果为 True,仅当键不存在时才设置

 False

Returns:

Type Description
bool

是否设置成功
Source code in src/symphra_cache/backends/memory.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
async def aset(
    self,
    key: CacheKey,
    value: CacheValue,
    ttl: int | None = None,
    ex: bool = False,
    nx: bool = False,
) -> bool:
    """
    异步设置缓存值

    Args:
        key: 缓存键
        value: 缓存值
        ttl: 过期时间(秒)
        ex: 保留参数
        nx: 如果为 True,仅当键不存在时才设置

    Returns:
        是否设置成功
    """
    return self.set(key, value, ttl=ttl, ex=ex, nx=nx)

attl(key) async

异步获取键的剩余生存时间

Source code in src/symphra_cache/backends/memory.py 
452
453
454
async def attl(self, key: CacheKey) -> int:
    """异步获取键的剩余生存时间"""
    return self.ttl(key)

clear()

清空所有缓存

警告

此操作不可逆，会删除所有缓存数据

示例

backend.clear() # 删除所有缓存

Source code in src/symphra_cache/backends/memory.py 
274
275
276
277
278
279
280
281
282
283
284
285
def clear(self) -> None:
    """
    清空所有缓存

    警告:
        此操作不可逆，会删除所有缓存数据

    示例:
        >>> backend.clear()  # 删除所有缓存
    """
    with self._lock:
        self._cache.clear()

close()

关闭后端

停止后台清理线程。

Source code in src/symphra_cache/backends/memory.py 
456
457
458
459
460
461
462
463
464
465
def close(self) -> None:
    """
    关闭后端

    停止后台清理线程。
    """
    if self._cleanup_thread and self._cleanup_thread.is_alive():
        # 设置停止标志(如果有的话)
        # 当前实现中清理线程是 daemon,会自动退出
        pass

delete(key)

删除缓存

Parameters:

Name Type Description Default
key CacheKey

缓存键

 required

Returns:

Type Description
bool

如果键存在并成功删除返回 True，否则返回 False
示例

backend.set("temp", "data") backend.delete("temp") # True backend.delete("temp") # False（已删除）

Source code in src/symphra_cache/backends/memory.py 
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
def delete(self, key: CacheKey) -> bool:
    """
    删除缓存

    Args:
        key: 缓存键

    Returns:
        如果键存在并成功删除返回 True，否则返回 False

    示例:
        >>> backend.set("temp", "data")
        >>> backend.delete("temp")  # True
        >>> backend.delete("temp")  # False（已删除）
    """
    with self._lock:
        if key in self._cache:
            del self._cache[key]
            return True
        return False

exists(key)

检查键是否存在

会检查键是否过期，过期的键返回 False。

Parameters:

Name Type Description Default
key CacheKey

缓存键

 required

Returns:

Type Description
bool

如果键存在且未过期返回 True，否则返回 False
示例

backend.set("key", "value", ttl=60) backend.exists("key") # True time.sleep(61) backend.exists("key") # False（已过期）

Source code in src/symphra_cache/backends/memory.py 
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
def exists(self, key: CacheKey) -> bool:
    """
    检查键是否存在

    会检查键是否过期，过期的键返回 False。

    Args:
        key: 缓存键

    Returns:
        如果键存在且未过期返回 True，否则返回 False

    示例:
        >>> backend.set("key", "value", ttl=60)
        >>> backend.exists("key")  # True
        >>> time.sleep(61)
        >>> backend.exists("key")  # False（已过期）
    """
    return self.get(key) is not None

get(key)

同步获取缓存值

实现细节: 1. 检查键是否存在 2. 检查是否过期（惰性删除） 3. 更新 LRU 顺序（移到末尾表示最近使用） 4. 返回值

时间复杂度: O(1)

Parameters:

Name Type Description Default
key CacheKey

缓存键

 required

Returns:

Type Description
CacheValue | None

缓存值，不存在或已过期则返回 None
示例

backend.set("key", "value", ttl=60) backend.get("key") # "value" time.sleep(61) backend.get("key") # None（已过期）

Source code in src/symphra_cache/backends/memory.py 
 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
def get(self, key: CacheKey) -> CacheValue | None:
    """
    同步获取缓存值

    实现细节:
    1. 检查键是否存在
    2. 检查是否过期（惰性删除）
    3. 更新 LRU 顺序（移到末尾表示最近使用）
    4. 返回值

    时间复杂度: O(1)

    Args:
        key: 缓存键

    Returns:
        缓存值，不存在或已过期则返回 None

    示例:
        >>> backend.set("key", "value", ttl=60)
        >>> backend.get("key")  # "value"
        >>> time.sleep(61)
        >>> backend.get("key")  # None（已过期）
    """
    with self._lock:
        # 检查键是否存在
        if key not in self._cache:
            return None

        value, expires_at = self._cache[key]

        # 检查是否过期（惰性删除）
        if expires_at is not None and time.time() > expires_at:
            # 已过期，删除并返回 None
            del self._cache[key]
            return None

        # 更新 LRU：移到末尾表示最近使用
        self._cache.move_to_end(key)

        return value

get_many(keys)

批量获取缓存值（优化版）

相比基类的默认实现，此版本在单个锁内完成所有操作， 减少锁开销，提升性能。

Parameters:

Name Type Description Default
keys list[CacheKey]

缓存键列表

 required

Returns:

Type Description
dict[CacheKey, CacheValue]

键值对字典，不存在或已过期的键不包含在结果中
示例

backend.set_many({"k1": "v1", "k2": "v2"}) results = backend.get_many(["k1", "k2", "k3"]) print(results) # {"k1": "v1", "k2": "v2"}

Source code in src/symphra_cache/backends/memory.py 
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
def get_many(self, keys: list[CacheKey]) -> dict[CacheKey, CacheValue]:
    """
    批量获取缓存值（优化版）

    相比基类的默认实现，此版本在单个锁内完成所有操作，
    减少锁开销，提升性能。

    Args:
        keys: 缓存键列表

    Returns:
        键值对字典，不存在或已过期的键不包含在结果中

    示例:
        >>> backend.set_many({"k1": "v1", "k2": "v2"})
        >>> results = backend.get_many(["k1", "k2", "k3"])
        >>> print(results)  # {"k1": "v1", "k2": "v2"}
    """
    result: dict[CacheKey, CacheValue] = {}
    now = time.time()

    with self._lock:
        for key in keys:
            if key not in self._cache:
                continue

            value, expires_at = self._cache[key]

            # 检查是否过期
            if expires_at is not None and now > expires_at:
                # 过期，删除（惰性清理）
                del self._cache[key]
                continue

            # 更新 LRU
            self._cache.move_to_end(key)
            result[key] = value

    return result

keys(pattern='*', cursor=0, count=100, max_keys=None)

扫描缓存键

Parameters:

Name Type Description Default
pattern str

匹配模式(支持通配符 * 和 ?)

 '*'
cursor int

游标位置

 0
count int

每页返回的键数量

 100
max_keys int | None

最多返回的键数量

 None

Returns:

Type Description
KeysPage

KeysPage 对象
Source code in src/symphra_cache/backends/memory.py 
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
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
def keys(
    self,
    pattern: str = "*",
    cursor: int = 0,
    count: int = 100,
    max_keys: int | None = None,
) -> KeysPage:
    """
    扫描缓存键

    Args:
        pattern: 匹配模式(支持通配符 * 和 ?)
        cursor: 游标位置
        count: 每页返回的键数量
        max_keys: 最多返回的键数量

    Returns:
        KeysPage 对象
    """
    import fnmatch

    from ..types import KeysPage

    with self._lock:
        # 获取所有键并过滤
        all_keys = list(self._cache.keys())

        # 模式匹配
        if pattern != "*":
            matched_keys = [k for k in all_keys if fnmatch.fnmatch(k, pattern)]
        else:
            matched_keys = all_keys

        # 分页处理
        total = len(matched_keys)
        start_idx = cursor
        end_idx = start_idx + count

        if max_keys is not None:
            end_idx = min(end_idx, start_idx + max_keys)

        page_keys = matched_keys[start_idx:end_idx]

        # 计算下一页游标
        next_cursor = end_idx if end_idx < total else 0
        has_more = next_cursor > 0

        return KeysPage(
            keys=page_keys,
            cursor=next_cursor,
            has_more=has_more,
            total_scanned=len(page_keys),
        )

set(key, value, ttl=None, ex=False, nx=False)

设置缓存值

Parameters:

Name Type Description Default
key CacheKey

缓存键

 required
value CacheValue

缓存值

 required
ttl int | None

过期时间(秒),None 表示永不过期

 None
ex bool

保留参数,内存后端始终使用相对过期时间

 False
nx bool

如果为 True,仅当键不存在时才设置

 False

Returns:

Type Description
bool

是否设置成功
Source code in src/symphra_cache/backends/memory.py 
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
def set(
    self,
    key: CacheKey,
    value: CacheValue,
    ttl: int | None = None,
    ex: bool = False,
    nx: bool = False,
) -> bool:
    """
    设置缓存值

    Args:
        key: 缓存键
        value: 缓存值
        ttl: 过期时间(秒),None 表示永不过期
        ex: 保留参数,内存后端始终使用相对过期时间
        nx: 如果为 True,仅当键不存在时才设置

    Returns:
        是否设置成功
    """
    with self._lock:
        # 检查 max_size 是否为 0
        if self._max_size == 0:
            return False

        # NX 模式:仅当键不存在时设置
        if nx and key in self._cache:
            # 检查是否已过期
            _, expires_at = self._cache[key]
            if expires_at is None or time.time() <= expires_at:
                return False  # 键存在且未过期,设置失败

        # 计算过期时间
        expires_at = None if ttl is None else time.time() + ttl

        # 如果键已存在,更新位置
        if key in self._cache:
            self._cache.move_to_end(key)
        # 如果缓存已满,执行 LRU 淘汰
        elif len(self._cache) >= self._max_size:
            self._cache.popitem(last=False)

        # 设置缓存值
        self._cache[key] = (value, expires_at)
        return True

set_many(mapping, ttl=None)

批量设置缓存值（优化版）

在单个锁内完成所有操作，提升性能。

Parameters:

Name Type Description Default
mapping dict[CacheKey, CacheValue]

键值对字典

 required
ttl int | None

过期时间（秒），None 表示永不过期

 None
示例

backend.set_many( ... { ... "user:1": {"name": "Alice"}, ... "user:2": {"name": "Bob"}, ... }, ... ttl=600, ... )

Source code in src/symphra_cache/backends/memory.py 
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
def set_many(
    self,
    mapping: dict[CacheKey, CacheValue],
    ttl: int | None = None,
) -> None:
    """
    批量设置缓存值（优化版）

    在单个锁内完成所有操作，提升性能。

    Args:
        mapping: 键值对字典
        ttl: 过期时间（秒），None 表示永不过期

    示例:
        >>> backend.set_many(
        ...     {
        ...         "user:1": {"name": "Alice"},
        ...         "user:2": {"name": "Bob"},
        ...     },
        ...     ttl=600,
        ... )
    """
    with self._lock:
        # 边界情况：max_size=0 时不存储任何内容
        if self._max_size == 0:
            return

        expires_at = time.time() + ttl if ttl is not None else None

        for key, value in mapping.items():
            # 检查容量并 LRU 淘汰
            if len(self._cache) >= self._max_size and key not in self._cache:
                self._cache.popitem(last=False)

            # 存储并移到末尾
            self._cache[key] = (value, expires_at)
            self._cache.move_to_end(key)

ttl(key)

获取键的剩余生存时间

Returns:

Type Description
int

剩余秒数,-1 表示永不过期,-2 表示键不存在
Source code in src/symphra_cache/backends/memory.py 
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
def ttl(self, key: CacheKey) -> int:
    """
    获取键的剩余生存时间

    Returns:
        剩余秒数,-1 表示永不过期,-2 表示键不存在
    """
    with self._lock:
        if key not in self._cache:
            return -2

        _, expires_at = self._cache[key]
        if expires_at is None:
            return -1

        remaining = int(expires_at - time.time())
        return remaining if remaining > 0 else -2