Files
Aether/_deprecated_py_src/plugins/cache/memory.py
fawney19 1d9c77522a refactor: 移除 Python 后端源码,全面迁移至 Rust gateway 架构
- 删除全部 Python 源码 (src/) 及 Alembic 迁移脚本,归档至 _deprecated_py_src/
- 重构 Rust gateway ai_pipeline: 拆分 planner/finalize 模块,新增 contracts/adaptation 层
- 重组 handlers 模块为 admin/public/proxy/internal/shared 子模块结构
- 新增 executor 模块,引入 Rust 原生数据库迁移 (aether-data/migrations)
- 简化 CI/Docker 构建流程,移除 base image 二级构建,统一为单一 app image
- 移除 Python 相关基础设施文件 (entrypoint.sh, gunicorn_conf.py, Dockerfile.base)
2026-04-03 16:26:16 +08:00

210 lines
6.5 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
内存缓存插件
基于Python字典的简单内存缓存实现
"""
from __future__ import annotations
import asyncio
import time
from collections import OrderedDict
from typing import Any
from .base import CachePlugin
class MemoryCachePlugin(CachePlugin):
"""
内存缓存插件
使用OrderedDict实现LRU缓存
"""
def __init__(self, name: str = "memory", config: dict[str, Any] | None = None):
super().__init__(name, config)
self._cache: OrderedDict = OrderedDict()
self._expiry: dict[str, float] = {}
self._lock = asyncio.Lock()
self._hits = 0
self._misses = 0
self._evictions = 0
self._cleanup_task = None
self._cleanup_interval = 60 # 默认值
# 启动清理任务
if config is not None:
self._cleanup_interval = config.get("cleanup_interval", 60)
try:
self._start_cleanup_task()
except Exception:
pass # 事件循环尚未启动,将在首次 set 时延迟启动
def _start_cleanup_task(self) -> None:
"""启动后台清理任务"""
async def cleanup_loop() -> None:
while self.enabled:
await asyncio.sleep(self._cleanup_interval)
await self._cleanup_expired()
try:
loop = asyncio.get_running_loop()
self._cleanup_task = loop.create_task(cleanup_loop())
except RuntimeError:
# 没有运行的事件循环,稍后再启动
pass
async def _cleanup_expired(self) -> None:
"""清理过期的缓存项"""
now = time.time()
expired_keys = []
async with self._lock:
for key, expiry in self._expiry.items():
if expiry < now:
expired_keys.append(key)
for key in expired_keys:
self._cache.pop(key, None)
self._expiry.pop(key, None)
self._evictions += 1
def _check_size(self) -> None:
"""检查并维护缓存大小限制"""
if len(self._cache) >= self.max_size:
# 删除最老的项LRU
key = next(iter(self._cache))
self._cache.pop(key)
self._expiry.pop(key, None)
self._evictions += 1
async def get(self, key: str) -> Any | None:
"""获取缓存值"""
async with self._lock:
# 检查是否过期
if key in self._expiry:
if self._expiry[key] < time.time():
# 已过期,删除
self._cache.pop(key, None)
self._expiry.pop(key)
self._misses += 1
return None
# 获取值并更新访问顺序LRU
if key in self._cache:
value = self._cache.pop(key)
self._cache[key] = value # 移到末尾
self._hits += 1
return self.deserialize(value) if isinstance(value, str) else value
else:
self._misses += 1
return None
def _ensure_cleanup_task(self) -> None:
"""确保清理任务已启动(延迟启动,在事件循环可用后调用)"""
if self._cleanup_task is None or self._cleanup_task.done():
try:
self._start_cleanup_task()
except Exception:
pass
async def set(self, key: str, value: Any, ttl: int | None = None) -> bool:
"""设置缓存值"""
self._ensure_cleanup_task()
async with self._lock:
# 检查大小限制
if key not in self._cache:
self._check_size()
# 序列化值
if not isinstance(value, str):
value = self.serialize(value)
# 设置值
self._cache[key] = value
self._cache.move_to_end(key) # 移到末尾(最新)
# 设置过期时间
if ttl is None:
ttl = self.default_ttl
if ttl > 0:
self._expiry[key] = time.time() + ttl
return True
async def delete(self, key: str) -> bool:
"""删除缓存项"""
async with self._lock:
if key in self._cache:
self._cache.pop(key)
self._expiry.pop(key, None)
return True
return False
async def exists(self, key: str) -> bool:
"""检查缓存项是否存在"""
async with self._lock:
# 检查是否过期
if key in self._expiry:
if self._expiry[key] < time.time():
# 已过期,删除
self._cache.pop(key, None)
self._expiry.pop(key)
return False
return key in self._cache
async def clear(self) -> bool:
"""清空所有缓存"""
async with self._lock:
self._cache.clear()
self._expiry.clear()
return True
async def get_many(self, keys: list[str]) -> dict[str, Any]:
"""批量获取缓存值"""
result = {}
for key in keys:
value = await self.get(key)
if value is not None:
result[key] = value
return result
async def set_many(self, items: dict[str, Any], ttl: int | None = None) -> bool:
"""批量设置缓存值"""
success = True
for key, value in items.items():
if not await self.set(key, value, ttl):
success = False
return success
async def get_stats(self) -> dict[str, Any]:
"""获取缓存统计信息"""
total_requests = self._hits + self._misses
hit_rate = self._hits / total_requests if total_requests > 0 else 0
return {
"type": "memory",
"size": len(self._cache),
"max_size": self.max_size,
"hits": self._hits,
"misses": self._misses,
"hit_rate": hit_rate,
"evictions": self._evictions,
"cleanup_interval": self._cleanup_interval,
}
async def _do_shutdown(self) -> None:
"""清理资源"""
# 取消清理任务
if self._cleanup_task:
self._cleanup_task.cancel()
try:
await self._cleanup_task
except asyncio.CancelledError:
pass
def __del__(self) -> None:
"""清理资源"""
if hasattr(self, "_cleanup_task") and self._cleanup_task:
self._cleanup_task.cancel()