mirror of
https://github.com/zhenxun-org/zhenxun_bot.git
synced 2026-09-28 16:20:56 +08:00
603 lines
21 KiB
Python
603 lines
21 KiB
Python
import re
|
|
from typing import Any, ClassVar, Generic, TypeVar, cast
|
|
|
|
from zhenxun.services.cache import Cache, CacheRoot, cache_config
|
|
from zhenxun.services.cache.config import COMPOSITE_KEY_SEPARATOR, CacheMode
|
|
from zhenxun.services.db_context import Model, with_db_timeout
|
|
from zhenxun.services.log import logger
|
|
|
|
T = TypeVar("T", bound=Model)
|
|
|
|
|
|
class DataAccess(Generic[T]):
|
|
"""数据访问兼容层，根据配置保留单点缓存读取和清理能力
|
|
|
|
边界说明：DataAccess 面向普通业务查询和低频管理链路。权限检查热路径
|
|
必须优先使用 RuntimeCache/AuthSnapshot，避免高并发消息处理时触发
|
|
DB/cache-aside 读放大。
|
|
|
|
新的高频运行态路径应优先使用 RuntimeCache 或 BoundedTTLCache。
|
|
这里不再把 filter/all/create/update_or_create 结果写入通用缓存，
|
|
create/update_or_create 只负责清理旧缓存，避免旧值残留。
|
|
|
|
使用示例:
|
|
```python
|
|
from zhenxun.services import DataAccess
|
|
from zhenxun.models.plugin_info import PluginInfo
|
|
|
|
# 创建数据访问对象
|
|
plugin_dao = DataAccess(PluginInfo)
|
|
|
|
# 获取单个数据
|
|
plugin = await plugin_dao.get(module="example_module")
|
|
|
|
# 获取所有数据
|
|
all_plugins = await plugin_dao.all()
|
|
|
|
# 筛选数据
|
|
enabled_plugins = await plugin_dao.filter(status=True)
|
|
|
|
# 创建数据
|
|
new_plugin = await plugin_dao.create(
|
|
module="new_module",
|
|
name="新插件",
|
|
status=True
|
|
)
|
|
```
|
|
"""
|
|
|
|
# 添加缓存统计信息
|
|
_cache_stats: ClassVar[dict] = {}
|
|
_ENABLE_CACHE_STATS: ClassVar[bool] = False
|
|
# 空结果标记
|
|
_NULL_RESULT = "__NULL_RESULT_PLACEHOLDER__"
|
|
# 默认空结果缓存时间（秒）- 设置为5分钟，避免频繁查询数据库
|
|
_NULL_RESULT_TTL = 300
|
|
|
|
@classmethod
|
|
def set_null_result_ttl(cls, seconds: int) -> None:
|
|
"""设置空结果缓存时间
|
|
|
|
参数:
|
|
seconds: 缓存时间（秒）
|
|
"""
|
|
if seconds < 0:
|
|
raise ValueError("缓存时间不能为负数")
|
|
cls._NULL_RESULT_TTL = seconds
|
|
logger.info(f"已设置DataAccess空结果缓存时间为 {seconds} 秒")
|
|
|
|
@classmethod
|
|
def get_null_result_ttl(cls) -> int:
|
|
"""获取空结果缓存时间
|
|
|
|
返回:
|
|
int: 缓存时间（秒）
|
|
"""
|
|
return cls._NULL_RESULT_TTL
|
|
|
|
def __init__(
|
|
self, model_cls: type[T], key_field: str = "id", cache_type: str | None = None
|
|
):
|
|
"""初始化数据访问对象
|
|
|
|
参数:
|
|
model_cls: 模型类
|
|
key_field: 主键字段
|
|
"""
|
|
self.model_cls = model_cls
|
|
self.key_field = getattr(model_cls, "cache_key_field", key_field)
|
|
self.cache_type = getattr(model_cls, "cache_type", cache_type)
|
|
|
|
self.cache = Cache(self.cache_type) if self.cache_type else None
|
|
|
|
# 初始化缓存统计
|
|
if self.cache_type and self.cache_type not in self._cache_stats:
|
|
self._cache_stats[self.cache_type] = {
|
|
"hits": 0, # 缓存命中次数
|
|
"misses": 0, # 缓存未命中次数
|
|
"null_hits": 0, # 空结果缓存命中次数
|
|
"sets": 0, # 缓存设置次数
|
|
"null_sets": 0, # 空结果缓存设置次数
|
|
"deletes": 0, # 缓存删除次数
|
|
}
|
|
|
|
@classmethod
|
|
def get_cache_stats(cls):
|
|
"""获取缓存统计信息"""
|
|
result = []
|
|
for cache_type, stats in cls._cache_stats.items():
|
|
hits = stats["hits"]
|
|
null_hits = stats.get("null_hits", 0)
|
|
misses = stats["misses"]
|
|
total = hits + null_hits + misses
|
|
hit_rate = ((hits + null_hits) / total * 100) if total > 0 else 0
|
|
result.append(
|
|
{
|
|
"cache_type": cache_type,
|
|
"hits": hits,
|
|
"null_hits": null_hits,
|
|
"misses": misses,
|
|
"sets": stats["sets"],
|
|
"null_sets": stats.get("null_sets", 0),
|
|
"deletes": stats["deletes"],
|
|
"hit_rate": f"{hit_rate:.2f}%",
|
|
}
|
|
)
|
|
return result
|
|
|
|
@classmethod
|
|
def reset_cache_stats(cls):
|
|
"""重置缓存统计信息"""
|
|
for stats in cls._cache_stats.values():
|
|
stats["hits"] = 0
|
|
stats["null_hits"] = 0
|
|
stats["misses"] = 0
|
|
stats["sets"] = 0
|
|
stats["null_sets"] = 0
|
|
stats["deletes"] = 0
|
|
|
|
@classmethod
|
|
def _bump_cache_stat(cls, cache_type: str | None, key: str) -> None:
|
|
if not cls._ENABLE_CACHE_STATS or not cache_type:
|
|
return
|
|
cls._cache_stats[cache_type][key] += 1
|
|
|
|
def _build_cache_key_from_kwargs(self, **kwargs) -> str | None:
|
|
"""从关键字参数构建缓存键
|
|
|
|
参数:
|
|
**kwargs: 关键字参数
|
|
|
|
返回:
|
|
str | None: 缓存键，如果无法构建则返回None
|
|
"""
|
|
# 含 ORM lookup 后缀(如 channel_id__isnull / x__gte)的查询无法可靠映射到
|
|
# 单条主键缓存键,退化为直查 DB,避免空串与 NULL 语义混淆导致错误命中(A5)。
|
|
if any("__" in key for key in kwargs):
|
|
return None
|
|
if isinstance(self.key_field, tuple):
|
|
# 多字段主键
|
|
key_parts = []
|
|
key_parts.extend(str(kwargs.get(field, "")) for field in self.key_field)
|
|
return COMPOSITE_KEY_SEPARATOR.join(key_parts) if key_parts else None
|
|
elif self.key_field in kwargs:
|
|
# 单字段主键
|
|
return str(kwargs[self.key_field])
|
|
return None
|
|
|
|
async def _get_with_cache(
|
|
self, db_query_func, allow_not_exist: bool = True, *args, **kwargs
|
|
) -> T | None:
|
|
"""带缓存的通用获取方法
|
|
|
|
参数:
|
|
db_query_func: 数据库查询函数
|
|
allow_not_exist: 是否允许数据不存在
|
|
*args: 查询参数
|
|
**kwargs: 查询参数
|
|
|
|
返回:
|
|
Optional[T]: 查询结果，如果不存在返回None
|
|
"""
|
|
# 如果没有缓存类型，直接从数据库获取
|
|
if not self.cache_type or cache_config.cache_mode == CacheMode.NONE:
|
|
logger.debug(f"{self.model_cls.__name__} 直接从数据库获取数据: {kwargs}")
|
|
return await with_db_timeout(
|
|
db_query_func(*args, **kwargs),
|
|
operation=f"{self.model_cls.__name__}.{db_query_func.__name__}",
|
|
source="DataAccess",
|
|
)
|
|
|
|
# 尝试从缓存获取
|
|
cache_key = None
|
|
try:
|
|
# 尝试构建缓存键
|
|
cache_key = self._build_cache_key_from_kwargs(**kwargs)
|
|
|
|
# 如果成功构建缓存键，尝试从缓存获取
|
|
if cache_key is not None:
|
|
data = await self.cache.get(cache_key) if self.cache else None
|
|
logger.debug(
|
|
f"{self.model_cls.__name__} key: {cache_key}"
|
|
f" 从缓存获取到的数据 {type(data)}: {data}"
|
|
)
|
|
|
|
if data == self._NULL_RESULT:
|
|
# 空结果缓存命中
|
|
self._bump_cache_stat(self.cache_type, "null_hits")
|
|
logger.debug(
|
|
f"{self.model_cls.__name__} 从缓存获取到空结果: {cache_key}"
|
|
)
|
|
if allow_not_exist:
|
|
logger.debug(
|
|
f"{self.model_cls.__name__} 从缓存获取"
|
|
f"到空结果: {cache_key}, 允许数据不存在，返回None"
|
|
)
|
|
return None
|
|
elif data:
|
|
# 缓存命中
|
|
self._bump_cache_stat(self.cache_type, "hits")
|
|
logger.debug(
|
|
f"{self.model_cls.__name__} 从缓存获取数据成功: {cache_key}"
|
|
)
|
|
return cast(T, data)
|
|
else:
|
|
# 缓存未命中
|
|
self._bump_cache_stat(self.cache_type, "misses")
|
|
logger.debug(f"{self.model_cls.__name__} 缓存未命中: {cache_key}")
|
|
except Exception as e:
|
|
logger.error(f"{self.model_cls.__name__} 从缓存获取数据失败: {kwargs}", e=e)
|
|
|
|
# 如果缓存中没有，从数据库获取
|
|
logger.debug(f"{self.model_cls.__name__} 从数据库获取数据: {kwargs}")
|
|
data = await db_query_func(*args, **kwargs)
|
|
|
|
# 如果获取到数据，存入缓存
|
|
if data:
|
|
try:
|
|
# 生成缓存键
|
|
cache_key = self._build_cache_key_for_item(data)
|
|
if cache_key is not None:
|
|
# 存入缓存
|
|
if self.cache:
|
|
await self.cache.set(cache_key, data)
|
|
self._bump_cache_stat(self.cache_type, "sets")
|
|
logger.debug(
|
|
f"{self.model_cls.__name__} 数据已存入缓存: {cache_key}"
|
|
)
|
|
except Exception as e:
|
|
logger.error(
|
|
f"{self.model_cls.__name__} 存入缓存失败，参数: {kwargs}", e=e
|
|
)
|
|
elif cache_key is not None:
|
|
# 如果没有获取到数据，缓存空结果
|
|
try:
|
|
# 存入空结果缓存，使用较短的过期时间
|
|
await self.cache.set(
|
|
cache_key, self._NULL_RESULT, expire=self._NULL_RESULT_TTL
|
|
) if self.cache else None
|
|
self._bump_cache_stat(self.cache_type, "null_sets")
|
|
logger.debug(
|
|
f"{self.model_cls.__name__} 空结果已存入缓存: {cache_key},"
|
|
f" TTL={self._NULL_RESULT_TTL}秒"
|
|
)
|
|
except Exception as e:
|
|
logger.error(
|
|
f"{self.model_cls.__name__} 存入空结果缓存失败，参数: {kwargs}", e=e
|
|
)
|
|
|
|
return data
|
|
|
|
async def get_or_none(
|
|
self, allow_not_exist: bool = True, *args, **kwargs
|
|
) -> T | None:
|
|
"""获取单条数据
|
|
|
|
参数:
|
|
allow_not_exist: 是否允许数据不存在
|
|
*args: 查询参数
|
|
**kwargs: 查询参数
|
|
|
|
返回:
|
|
Optional[T]: 查询结果，如果不存在返回None
|
|
"""
|
|
return await self._get_with_cache(
|
|
self.model_cls.get_or_none, allow_not_exist, *args, **kwargs
|
|
)
|
|
|
|
async def safe_get_or_none(
|
|
self, allow_not_exist: bool = True, *args, **kwargs
|
|
) -> T | None:
|
|
"""安全的获取单条数据
|
|
|
|
参数:
|
|
allow_not_exist: 是否允许数据不存在
|
|
*args: 查询参数
|
|
**kwargs: 查询参数
|
|
|
|
返回:
|
|
Optional[T]: 查询结果，如果不存在返回None
|
|
"""
|
|
return await self._get_with_cache(
|
|
self.model_cls.safe_get_or_none, allow_not_exist, *args, **kwargs
|
|
)
|
|
|
|
async def get_by_func_or_none(
|
|
self, func, allow_not_exist: bool = True, *args, **kwargs
|
|
) -> T | None:
|
|
"""根据函数获取数据
|
|
|
|
参数:
|
|
func: 函数
|
|
allow_not_exist: 是否允许数据不存在
|
|
*args: 查询参数
|
|
**kwargs: 查询参数
|
|
"""
|
|
return await self._get_with_cache(func, allow_not_exist, *args, **kwargs)
|
|
|
|
async def clear_cache(self, **kwargs) -> bool:
|
|
"""只清除缓存，不影响数据库数据
|
|
|
|
参数:
|
|
**kwargs: 查询参数，必须包含主键字段
|
|
|
|
返回:
|
|
bool: 是否成功清除缓存
|
|
"""
|
|
# 如果没有缓存类型，直接返回True
|
|
if not self.cache_type or cache_config.cache_mode == CacheMode.NONE:
|
|
return True
|
|
|
|
try:
|
|
# 构建缓存键
|
|
cache_key = self._build_cache_key_from_kwargs(**kwargs)
|
|
if cache_key is None:
|
|
if isinstance(self.key_field, tuple):
|
|
# 如果是复合键，检查缺少哪些字段
|
|
missing_fields = [
|
|
field for field in self.key_field if field not in kwargs
|
|
]
|
|
logger.error(
|
|
f"清除{self.model_cls.__name__}缓存失败: "
|
|
f"缺少主键字段 {', '.join(missing_fields)}"
|
|
)
|
|
else:
|
|
logger.error(
|
|
f"清除{self.model_cls.__name__}缓存失败: "
|
|
f"缺少主键字段 {self.key_field}"
|
|
)
|
|
return False
|
|
|
|
# 删除缓存
|
|
if self.cache:
|
|
await self.cache.delete(cache_key)
|
|
self._bump_cache_stat(self.cache_type, "deletes")
|
|
logger.debug(f"已清除{self.model_cls.__name__}缓存: {cache_key}")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"清除{self.model_cls.__name__}缓存失败", e=e)
|
|
return False
|
|
|
|
def _build_composite_key(self, data: T) -> str | None:
|
|
"""构建复合缓存键
|
|
|
|
参数:
|
|
data: 数据对象
|
|
|
|
返回:
|
|
str | None: 构建的缓存键，如果无法构建则返回None
|
|
"""
|
|
# 如果是元组，表示多个字段组成键
|
|
if isinstance(self.key_field, tuple):
|
|
# 构建键参数列表
|
|
key_parts = []
|
|
for field in self.key_field:
|
|
value = getattr(data, field, "")
|
|
key_parts.append(value if value is not None else "")
|
|
|
|
# 如果没有有效参数，返回None
|
|
return COMPOSITE_KEY_SEPARATOR.join(key_parts) if key_parts else None
|
|
elif hasattr(data, self.key_field):
|
|
value = getattr(data, self.key_field, None)
|
|
return str(value) if value is not None else None
|
|
|
|
return None
|
|
|
|
def _build_cache_key_for_item(self, item: T) -> str | None:
|
|
"""为数据项构建缓存键
|
|
|
|
参数:
|
|
item: 数据项
|
|
|
|
返回:
|
|
str | None: 缓存键，如果无法生成则返回None
|
|
"""
|
|
# 如果没有缓存类型，返回None
|
|
if not self.cache_type:
|
|
return None
|
|
|
|
# 获取缓存类型的配置信息
|
|
cache_model = CacheRoot.get_model(self.cache_type)
|
|
|
|
if not cache_model.key_format:
|
|
# 常规处理，使用主键作为缓存键
|
|
return self._build_composite_key(item)
|
|
# 构建键参数字典
|
|
key_parts = []
|
|
# 从格式字符串中提取所需的字段名
|
|
field_names = re.findall(r"{([^}]+)}", cache_model.key_format)
|
|
|
|
# 收集所有字段值
|
|
for field in field_names:
|
|
value = getattr(item, field, "")
|
|
key_parts.append(value if value is not None else "")
|
|
|
|
return COMPOSITE_KEY_SEPARATOR.join(key_parts)
|
|
|
|
async def _invalidate_item_cache(self, item: T, action: str) -> None:
|
|
if not self.cache_type or cache_config.cache_mode == CacheMode.NONE:
|
|
return
|
|
|
|
try:
|
|
cache_key = self._build_cache_key_for_item(item)
|
|
if cache_key is None:
|
|
return
|
|
|
|
if self.cache:
|
|
await self.cache.delete(cache_key)
|
|
self._bump_cache_stat(self.cache_type, "deletes")
|
|
logger.debug(
|
|
f"{self.model_cls.__name__} {action}: 已失效兼容缓存: {cache_key}"
|
|
)
|
|
except Exception as e:
|
|
logger.error(
|
|
f"{self.model_cls.__name__} {action}: 更新兼容缓存失败",
|
|
e=e,
|
|
)
|
|
|
|
async def filter(self, *args, **kwargs) -> list[T]:
|
|
"""筛选数据
|
|
|
|
参数:
|
|
*args: 查询参数
|
|
**kwargs: 查询参数
|
|
|
|
返回:
|
|
List[T]: 查询结果列表
|
|
"""
|
|
# 从数据库获取数据
|
|
logger.debug(f"{self.model_cls.__name__} filter: 从数据库查询, 参数: {kwargs}")
|
|
data_list = await self.model_cls.filter(*args, **kwargs)
|
|
logger.debug(
|
|
f"{self.model_cls.__name__} filter: 查询结果数量: {len(data_list)}"
|
|
)
|
|
|
|
return data_list
|
|
|
|
async def all(self) -> list[T]:
|
|
"""获取所有数据
|
|
|
|
返回:
|
|
List[T]: 所有数据列表
|
|
"""
|
|
# 直接从数据库获取
|
|
logger.debug(f"{self.model_cls.__name__} all: 从数据库查询所有数据")
|
|
data_list = await self.model_cls.all()
|
|
logger.debug(f"{self.model_cls.__name__} all: 查询结果数量: {len(data_list)}")
|
|
|
|
return data_list
|
|
|
|
async def count(self, *args, **kwargs) -> int:
|
|
"""获取数据数量
|
|
|
|
参数:
|
|
*args: 查询参数
|
|
**kwargs: 查询参数
|
|
|
|
返回:
|
|
int: 数据数量
|
|
"""
|
|
# 直接从数据库获取数量
|
|
return await self.model_cls.filter(*args, **kwargs).count()
|
|
|
|
async def exists(self, *args, **kwargs) -> bool:
|
|
"""判断数据是否存在
|
|
|
|
参数:
|
|
*args: 查询参数
|
|
**kwargs: 查询参数
|
|
|
|
返回:
|
|
bool: 是否存在
|
|
"""
|
|
# 直接从数据库判断是否存在
|
|
return await self.model_cls.filter(*args, **kwargs).exists()
|
|
|
|
async def create(self, **kwargs) -> T:
|
|
"""创建数据
|
|
|
|
参数:
|
|
**kwargs: 创建参数
|
|
|
|
返回:
|
|
T: 创建的数据
|
|
"""
|
|
# 创建数据
|
|
logger.debug(f"{self.model_cls.__name__} create: 创建数据, 参数: {kwargs}")
|
|
data = await self.model_cls.create(**kwargs)
|
|
|
|
await self._invalidate_item_cache(data, "create")
|
|
|
|
return data
|
|
|
|
async def update_or_create(
|
|
self, defaults: dict[str, Any] | None = None, **kwargs
|
|
) -> tuple[T, bool]:
|
|
"""更新或创建数据
|
|
|
|
参数:
|
|
defaults: 默认值
|
|
**kwargs: 查询参数
|
|
|
|
返回:
|
|
tuple[T, bool]: (数据, 是否创建)
|
|
"""
|
|
# 更新或创建数据
|
|
data, created = await self.model_cls.update_or_create(
|
|
defaults=defaults, **kwargs
|
|
)
|
|
|
|
await self._invalidate_item_cache(data, "update_or_create")
|
|
|
|
return data, created
|
|
|
|
async def delete(self, *args, **kwargs) -> int:
|
|
"""删除数据
|
|
|
|
参数:
|
|
*args: 查询参数
|
|
**kwargs: 查询参数
|
|
|
|
返回:
|
|
int: 删除的数据数量
|
|
"""
|
|
logger.debug(f"{self.model_cls.__name__} delete: 删除数据, 参数: {kwargs}")
|
|
|
|
# 如果有缓存类型且有key_field参数，先尝试删除缓存
|
|
if self.cache_type and cache_config.cache_mode != CacheMode.NONE:
|
|
try:
|
|
# 尝试构建缓存键
|
|
cache_key = self._build_cache_key_from_kwargs(**kwargs)
|
|
|
|
if cache_key is not None:
|
|
# 如果成功构建缓存键，直接删除缓存
|
|
if self.cache:
|
|
await self.cache.delete(cache_key)
|
|
self._bump_cache_stat(self.cache_type, "deletes")
|
|
logger.debug(
|
|
f"{self.model_cls.__name__} delete: 已删除缓存: {cache_key}"
|
|
)
|
|
else:
|
|
# 否则需要先查询出要删除的数据，然后删除对应的缓存
|
|
items = await self.model_cls.filter(*args, **kwargs)
|
|
logger.debug(
|
|
f"{self.model_cls.__name__} delete:"
|
|
f" 查询到 {len(items)} 条要删除的数据"
|
|
)
|
|
for item in items:
|
|
item_cache_key = self._build_cache_key_for_item(item)
|
|
if item_cache_key is not None:
|
|
if self.cache:
|
|
await self.cache.delete(item_cache_key)
|
|
self._bump_cache_stat(self.cache_type, "deletes")
|
|
if items:
|
|
logger.debug(
|
|
f"{self.model_cls.__name__} delete:"
|
|
f" 已删除 {len(items)} 条数据的缓存"
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"{self.model_cls.__name__} delete: 删除缓存失败", e=e)
|
|
|
|
# 删除数据
|
|
result = await self.model_cls.filter(*args, **kwargs).delete()
|
|
logger.debug(
|
|
f"{self.model_cls.__name__} delete: 已从数据库删除 {result} 条数据"
|
|
)
|
|
return result
|
|
|
|
def _generate_cache_key(self, data: T) -> str:
|
|
"""根据数据对象生成缓存键
|
|
|
|
参数:
|
|
data: 数据对象
|
|
|
|
返回:
|
|
str: 缓存键
|
|
"""
|
|
# 使用新方法构建复合键
|
|
if composite_key := self._build_composite_key(data):
|
|
return composite_key
|
|
|
|
# 如果无法生成复合键，生成一个唯一键
|
|
return f"object_{id(data)}"
|