407 lines
18 KiB
Python
407 lines
18 KiB
Python
from typing import Any, Literal
|
||
|
||
from cli.exit_codes import ARGUMENT_ERROR, DATABASE_ERROR, REDIS_ERROR
|
||
|
||
from .gateway import ConfigInfrastructureGateway
|
||
from .support import ConfigDomainSupport
|
||
|
||
|
||
class ConfigRuntimeService:
|
||
"""
|
||
参数配置运行时服务。
|
||
|
||
该服务作为参数配置运行时 facade，对外统一暴露参数配置列表、详情、
|
||
写入、缓存刷新与一致性诊断入口。
|
||
|
||
:param infrastructure_gateway: 参数配置基础设施网关
|
||
:param domain_support: 参数配置领域支持对象
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
infrastructure_gateway: ConfigInfrastructureGateway | None = None,
|
||
domain_support: ConfigDomainSupport | None = None,
|
||
) -> None:
|
||
"""
|
||
初始化参数配置运行时服务。
|
||
|
||
:param infrastructure_gateway: 参数配置基础设施网关
|
||
:param domain_support: 参数配置领域支持对象
|
||
:return: None
|
||
"""
|
||
self.infrastructure_gateway = infrastructure_gateway or ConfigInfrastructureGateway()
|
||
self.domain_support = domain_support or ConfigDomainSupport(self.infrastructure_gateway)
|
||
|
||
@staticmethod
|
||
def build_cli_config_model(config_vo_module: Any, config_record: Any) -> Any | None:
|
||
"""
|
||
将 ORM 配置记录显式转换为 CLI 可用的配置模型。
|
||
|
||
`ConfigModel` 当前未统一开启 `populate_by_name`，因此 CLI 侧在处理
|
||
ORM 蛇形字段对象时，需要主动映射为驼峰 alias 输入，避免详情查询结果
|
||
被静默序列化为空。
|
||
|
||
:param config_vo_module: 配置 VO 模块
|
||
:param config_record: ORM 配置记录
|
||
:return: CLI 可用配置模型
|
||
"""
|
||
if config_record is None:
|
||
return None
|
||
return config_vo_module.ConfigModel(
|
||
configId=getattr(config_record, 'config_id', None),
|
||
configName=getattr(config_record, 'config_name', None),
|
||
configKey=getattr(config_record, 'config_key', None),
|
||
configValue=getattr(config_record, 'config_value', None),
|
||
configType=getattr(config_record, 'config_type', None),
|
||
createBy=getattr(config_record, 'create_by', None),
|
||
createTime=getattr(config_record, 'create_time', None),
|
||
updateBy=getattr(config_record, 'update_by', None),
|
||
updateTime=getattr(config_record, 'update_time', None),
|
||
remark=getattr(config_record, 'remark', None),
|
||
)
|
||
|
||
async def load_config_from_database(
|
||
self,
|
||
config_key: str,
|
||
) -> tuple[Any | None, dict[str, Any] | None] | dict[str, Any]:
|
||
"""
|
||
从数据库加载参数配置。
|
||
|
||
:param config_key: 参数键名
|
||
:return: 成功时返回配置模型与序列化结果，失败时返回错误结果字典
|
||
"""
|
||
async_session_local = self.infrastructure_gateway.get_async_session_local()
|
||
config_dao = self.infrastructure_gateway.get_config_dao()
|
||
config_vo_module = self.infrastructure_gateway.get_config_vo_module()
|
||
try:
|
||
async with async_session_local() as session:
|
||
config_record = await config_dao.get_config_detail_by_info(
|
||
session,
|
||
config_vo_module.ConfigModel(configKey=config_key),
|
||
)
|
||
except Exception as exc:
|
||
return {'ok': False, 'message': '读取参数配置失败', 'error': str(exc), 'exit_code': DATABASE_ERROR}
|
||
|
||
database_model = self.build_cli_config_model(config_vo_module, config_record)
|
||
return database_model, self.domain_support.serialize_config_model(database_model)
|
||
|
||
async def load_config_from_cache(
|
||
self,
|
||
config_key: str,
|
||
) -> tuple[str | None, dict[str, Any] | None] | dict[str, Any]:
|
||
"""
|
||
从缓存加载参数配置。
|
||
|
||
:param config_key: 参数键名
|
||
:return: 成功时返回原始缓存值与序列化结果，失败时返回错误结果字典
|
||
"""
|
||
redis = None
|
||
redis_util = self.infrastructure_gateway.get_redis_util()
|
||
config_service = self.infrastructure_gateway.get_config_service()
|
||
redis_error = self.infrastructure_gateway.get_redis_error_class()
|
||
try:
|
||
redis = await redis_util.create_redis_pool(log_enabled=False)
|
||
cache_value = await config_service.query_config_list_from_cache_services(redis, config_key)
|
||
return cache_value, self.domain_support.serialize_cache_payload(config_key, cache_value)
|
||
except redis_error as exc:
|
||
return {'ok': False, 'message': '读取参数缓存失败', 'error': str(exc), 'exit_code': REDIS_ERROR}
|
||
finally:
|
||
if redis is not None:
|
||
await redis.close()
|
||
|
||
async def list_configs(
|
||
self,
|
||
config_name: str = '',
|
||
config_key: str = '',
|
||
config_type: Literal['Y', 'N'] | None = None,
|
||
begin_date: str = '',
|
||
end_date: str = '',
|
||
*,
|
||
paged: bool = False,
|
||
page_num: int = 1,
|
||
page_size: int = 20,
|
||
) -> dict[str, Any]:
|
||
"""
|
||
查询参数配置列表。
|
||
|
||
:param config_name: 参数名称过滤条件
|
||
:param config_key: 参数键名过滤条件
|
||
:param config_type: 参数类型过滤条件
|
||
:param begin_date: 开始日期
|
||
:param end_date: 结束日期
|
||
:param paged: 是否开启分页
|
||
:param page_num: 页码
|
||
:param page_size: 每页数量
|
||
:return: CLI 标准结果字典
|
||
"""
|
||
async_session_local = self.infrastructure_gateway.get_async_session_local()
|
||
config_service = self.infrastructure_gateway.get_config_service()
|
||
query_model = self.domain_support.build_config_query(
|
||
config_name,
|
||
config_key,
|
||
config_type,
|
||
begin_date,
|
||
end_date,
|
||
page_num,
|
||
page_size,
|
||
)
|
||
try:
|
||
async with async_session_local() as session:
|
||
result = await config_service.get_config_list_services(session, query_model, is_page=paged)
|
||
except Exception as exc:
|
||
return {'ok': False, 'message': '读取参数配置列表失败', 'error': str(exc), 'exit_code': DATABASE_ERROR}
|
||
|
||
filters = self.domain_support.build_list_filters(
|
||
config_name=config_name,
|
||
config_key=config_key,
|
||
config_type=config_type,
|
||
begin_date=begin_date,
|
||
end_date=end_date,
|
||
paged=paged,
|
||
page_num=page_num,
|
||
page_size=page_size,
|
||
)
|
||
if paged and isinstance(result, self.infrastructure_gateway.get_page_model()):
|
||
page_payload = result.model_dump(by_alias=True)
|
||
page_payload['rows'] = [
|
||
self.domain_support.sanitize_config_mapping(item) for item in page_payload.get('rows', [])
|
||
]
|
||
return {'ok': True, 'filters': filters, 'page': page_payload}
|
||
|
||
items = [self.domain_support.sanitize_config_mapping(item) for item in result]
|
||
return {'ok': True, 'filters': filters, 'count': len(items), 'items': items}
|
||
|
||
async def get_config(
|
||
self,
|
||
config_key: str,
|
||
*,
|
||
source: Literal['db', 'cache', 'both'] = 'both',
|
||
) -> dict[str, Any]:
|
||
"""
|
||
按参数键读取配置详情。
|
||
|
||
:param config_key: 参数键名
|
||
:param source: 读取来源
|
||
:return: CLI 标准结果字典
|
||
"""
|
||
database_model: Any | None = None
|
||
database_payload: dict[str, Any] | None = None
|
||
cache_value: str | None = None
|
||
cache_payload: dict[str, Any] | None = None
|
||
|
||
if source in {'db', 'both'}:
|
||
database_result = await self.load_config_from_database(config_key)
|
||
if isinstance(database_result, dict):
|
||
return database_result
|
||
database_model, database_payload = database_result
|
||
|
||
if source in {'cache', 'both'}:
|
||
cache_result = await self.load_config_from_cache(config_key)
|
||
if isinstance(cache_result, dict):
|
||
error_payload = {**cache_result, 'source': source}
|
||
if database_payload is not None:
|
||
error_payload['database'] = database_payload
|
||
return error_payload
|
||
cache_value, cache_payload = cache_result
|
||
|
||
if source == 'db' and database_payload is None:
|
||
return self.domain_support.build_missing_config_result(config_key, source)
|
||
if source == 'cache' and cache_payload is None:
|
||
return self.domain_support.build_missing_config_result(config_key, source)
|
||
if source == 'both' and database_payload is None and cache_payload is None:
|
||
return self.domain_support.build_missing_config_result(config_key, source)
|
||
|
||
payload = {'ok': True, 'key': config_key, 'source': source}
|
||
if source in {'db', 'both'}:
|
||
payload['database'] = database_payload
|
||
if source in {'cache', 'both'}:
|
||
payload['cache'] = cache_payload
|
||
if source == 'both':
|
||
payload['inSync'] = (database_model.config_value if database_model else None) == cache_value
|
||
return payload
|
||
|
||
async def set_config(
|
||
self,
|
||
config_key: str,
|
||
config_value: str,
|
||
*,
|
||
config_name: str | None = None,
|
||
config_type: Literal['Y', 'N'] | None = None,
|
||
remark: str | None = None,
|
||
dry_run: bool = False,
|
||
) -> dict[str, Any]:
|
||
"""
|
||
新增或更新参数配置，并同步缓存。
|
||
|
||
:param config_key: 参数键名
|
||
:param config_value: 参数键值
|
||
:param config_name: 参数名称
|
||
:param config_type: 参数类型
|
||
:param remark: 备注
|
||
:param dry_run: 是否仅执行演练
|
||
:return: CLI 标准结果字典
|
||
"""
|
||
async_session_local = self.infrastructure_gateway.get_async_session_local()
|
||
config_dao = self.infrastructure_gateway.get_config_dao()
|
||
config_vo_module = self.infrastructure_gateway.get_config_vo_module()
|
||
redis_util = self.infrastructure_gateway.get_redis_util()
|
||
redis_init_key_config = self.infrastructure_gateway.get_redis_init_key_config()
|
||
redis_error = self.infrastructure_gateway.get_redis_error_class()
|
||
existing_config = None
|
||
target_config = None
|
||
try:
|
||
async with async_session_local() as session:
|
||
config_record = await config_dao.get_config_detail_by_info(
|
||
session,
|
||
config_vo_module.ConfigModel(configKey=config_key),
|
||
)
|
||
existing_config = self.build_cli_config_model(config_vo_module, config_record)
|
||
if existing_config is None and not config_name:
|
||
return {
|
||
'ok': False,
|
||
'message': '新增参数配置时必须传入 --name',
|
||
'exit_code': ARGUMENT_ERROR,
|
||
}
|
||
|
||
target_config = self.domain_support.build_target_config_model(
|
||
config_key,
|
||
config_value,
|
||
config_name,
|
||
config_type,
|
||
remark,
|
||
existing_config,
|
||
)
|
||
if dry_run:
|
||
return {
|
||
'ok': True,
|
||
'message': '参数配置演练完成，未执行实际写入',
|
||
'dryRun': True,
|
||
'action': 'update' if existing_config else 'create',
|
||
'config': self.domain_support.serialize_config_model(target_config),
|
||
}
|
||
|
||
if existing_config:
|
||
await config_dao.edit_config_dao(session, target_config.model_dump(exclude_none=True))
|
||
else:
|
||
await config_dao.add_config_dao(session, target_config)
|
||
await session.commit()
|
||
except Exception as exc:
|
||
return {'ok': False, 'message': '写入参数配置失败', 'error': str(exc), 'exit_code': DATABASE_ERROR}
|
||
|
||
redis = None
|
||
try:
|
||
redis = await redis_util.create_redis_pool(log_enabled=False)
|
||
await redis.set(f'{redis_init_key_config.SYS_CONFIG.key}:{config_key}', config_value)
|
||
except redis_error as exc:
|
||
return {
|
||
'ok': False,
|
||
'message': '参数配置已写入数据库，但同步缓存失败',
|
||
'error': str(exc),
|
||
'config': self.domain_support.serialize_config_model(target_config),
|
||
'databaseCommitted': True,
|
||
'exit_code': REDIS_ERROR,
|
||
}
|
||
finally:
|
||
if redis is not None:
|
||
await redis.close()
|
||
|
||
return {
|
||
'ok': True,
|
||
'message': '参数配置已同步到数据库和缓存',
|
||
'action': 'update' if existing_config else 'create',
|
||
'config': self.domain_support.serialize_config_model(target_config),
|
||
}
|
||
|
||
async def sync_config_cache(self) -> dict[str, Any]:
|
||
"""
|
||
刷新参数配置缓存。
|
||
|
||
:return: CLI 标准结果字典
|
||
"""
|
||
redis = None
|
||
async_session_local = self.infrastructure_gateway.get_async_session_local()
|
||
config_service = self.infrastructure_gateway.get_config_service()
|
||
config_vo_module = self.infrastructure_gateway.get_config_vo_module()
|
||
redis_util = self.infrastructure_gateway.get_redis_util()
|
||
redis_error = self.infrastructure_gateway.get_redis_error_class()
|
||
try:
|
||
async with async_session_local() as session:
|
||
config_list = await config_service.get_config_list_services(
|
||
session,
|
||
config_vo_module.ConfigPageQueryModel(),
|
||
is_page=False,
|
||
)
|
||
redis = await redis_util.create_redis_pool(log_enabled=False)
|
||
await config_service.init_cache_sys_config_services(session, redis)
|
||
except redis_error as exc:
|
||
return {'ok': False, 'message': '刷新参数缓存失败', 'error': str(exc), 'exit_code': REDIS_ERROR}
|
||
except Exception as exc:
|
||
return {'ok': False, 'message': '刷新参数缓存失败', 'error': str(exc), 'exit_code': DATABASE_ERROR}
|
||
finally:
|
||
if redis is not None:
|
||
await redis.close()
|
||
|
||
return {'ok': True, 'message': '参数缓存刷新成功', 'count': len(config_list)}
|
||
|
||
async def diagnose_config(self, *, sample_limit: int = 10) -> dict[str, Any]:
|
||
"""
|
||
诊断数据库参数配置与 Redis 缓存的一致性状态。
|
||
|
||
:param sample_limit: 示例键名输出上限
|
||
:return: CLI 标准结果字典
|
||
"""
|
||
redis = None
|
||
async_session_local = self.infrastructure_gateway.get_async_session_local()
|
||
config_service = self.infrastructure_gateway.get_config_service()
|
||
config_vo_module = self.infrastructure_gateway.get_config_vo_module()
|
||
redis_util = self.infrastructure_gateway.get_redis_util()
|
||
redis_init_key_config = self.infrastructure_gateway.get_redis_init_key_config()
|
||
redis_error = self.infrastructure_gateway.get_redis_error_class()
|
||
try:
|
||
async with async_session_local() as session:
|
||
config_list = await config_service.get_config_list_services(
|
||
session,
|
||
config_vo_module.ConfigPageQueryModel(),
|
||
is_page=False,
|
||
)
|
||
redis = await redis_util.create_redis_pool(log_enabled=False)
|
||
cache_prefix = f'{redis_init_key_config.SYS_CONFIG.key}:'
|
||
cache_keys = sorted(await redis.keys(f'{cache_prefix}*'))
|
||
cache_values = await redis.mget(*cache_keys) if cache_keys else []
|
||
except redis_error as exc:
|
||
return {'ok': False, 'message': '读取参数缓存诊断信息失败', 'error': str(exc), 'exit_code': REDIS_ERROR}
|
||
except Exception as exc:
|
||
return {'ok': False, 'message': '读取参数配置诊断信息失败', 'error': str(exc), 'exit_code': DATABASE_ERROR}
|
||
finally:
|
||
if redis is not None:
|
||
await redis.close()
|
||
|
||
database_map = {str(item.get('configKey')): str(item.get('configValue', '')) for item in config_list}
|
||
cache_map = {
|
||
cache_key.replace(cache_prefix, '', 1): '' if cache_value is None else str(cache_value)
|
||
for cache_key, cache_value in zip(cache_keys, cache_values, strict=False)
|
||
}
|
||
missing_in_cache = sorted(key for key in database_map if key not in cache_map)
|
||
orphan_in_cache = sorted(key for key in cache_map if key not in database_map)
|
||
mismatch_keys = sorted(
|
||
key for key, database_value in database_map.items() if key in cache_map and database_value != cache_map[key]
|
||
)
|
||
in_sync = not missing_in_cache and not orphan_in_cache and not mismatch_keys
|
||
return {
|
||
'ok': in_sync,
|
||
'message': '参数配置数据库与缓存一致' if in_sync else '参数配置存在数据库与缓存不一致项',
|
||
'databaseCount': len(database_map),
|
||
'cacheCount': len(cache_map),
|
||
'missingInCacheCount': len(missing_in_cache),
|
||
'orphanInCacheCount': len(orphan_in_cache),
|
||
'mismatchCount': len(mismatch_keys),
|
||
'sampleLimit': sample_limit,
|
||
'missingInCache': missing_in_cache[:sample_limit],
|
||
'orphanInCache': orphan_in_cache[:sample_limit],
|
||
'mismatchKeys': mismatch_keys[:sample_limit],
|
||
}
|
||
|
||
|
||
CONFIG_RUNTIME = ConfigRuntimeService()
|