166 lines
4.8 KiB
Python
166 lines
4.8 KiB
Python
from typing import Optional, Dict, Any, List
|
||
from sqlalchemy.orm import Session
|
||
from app.models.models import ServiceConfig
|
||
import uuid
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class ConfigService:
|
||
"""配置服务"""
|
||
|
||
@staticmethod
|
||
def get_config(db: Session, config_key: str) -> Optional[Dict[str, Any]]:
|
||
"""获取配置
|
||
|
||
Args:
|
||
db: 数据库会话
|
||
config_key: 配置键
|
||
|
||
Returns:
|
||
配置值，如果不存在返回None
|
||
"""
|
||
config = db.query(ServiceConfig).filter_by(
|
||
config_key=config_key,
|
||
status="active"
|
||
).first()
|
||
|
||
if config:
|
||
return config.config_value
|
||
return None
|
||
|
||
@staticmethod
|
||
def set_config(db: Session, config_key: str, config_value: Dict[str, Any],
|
||
config_type: str = "system", service_id: Optional[str] = None,
|
||
description: str = "") -> bool:
|
||
"""设置配置
|
||
|
||
Args:
|
||
db: 数据库会话
|
||
config_key: 配置键
|
||
config_value: 配置值
|
||
config_type: 配置类型，默认为"system"
|
||
service_id: 服务ID，系统配置可为None
|
||
description: 配置描述
|
||
|
||
Returns:
|
||
是否设置成功
|
||
"""
|
||
try:
|
||
# 检查是否存在
|
||
existing_config = db.query(ServiceConfig).filter_by(
|
||
config_key=config_key
|
||
).first()
|
||
|
||
if existing_config:
|
||
# 更新现有配置
|
||
existing_config.config_value = config_value
|
||
existing_config.config_type = config_type
|
||
existing_config.service_id = service_id
|
||
existing_config.description = description
|
||
existing_config.status = "active"
|
||
else:
|
||
# 创建新配置
|
||
new_config = ServiceConfig(
|
||
id=f"config-{uuid.uuid4()}",
|
||
config_key=config_key,
|
||
config_value=config_value,
|
||
config_type=config_type,
|
||
service_id=service_id,
|
||
description=description,
|
||
status="active"
|
||
)
|
||
db.add(new_config)
|
||
|
||
db.commit()
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"Failed to set config: {str(e)}")
|
||
db.rollback()
|
||
return False
|
||
|
||
@staticmethod
|
||
def get_service_configs(db: Session, service_id: str) -> List[Dict[str, Any]]:
|
||
"""获取服务的所有配置
|
||
|
||
Args:
|
||
db: 数据库会话
|
||
service_id: 服务ID
|
||
|
||
Returns:
|
||
服务配置列表
|
||
"""
|
||
configs = db.query(ServiceConfig).filter_by(
|
||
service_id=service_id,
|
||
status="active"
|
||
).all()
|
||
|
||
return [
|
||
{
|
||
"key": config.config_key,
|
||
"value": config.config_value,
|
||
"type": config.config_type,
|
||
"description": config.description
|
||
}
|
||
for config in configs
|
||
]
|
||
|
||
@staticmethod
|
||
def delete_config(db: Session, config_key: str) -> bool:
|
||
"""删除配置
|
||
|
||
Args:
|
||
db: 数据库会话
|
||
config_key: 配置键
|
||
|
||
Returns:
|
||
是否删除成功
|
||
"""
|
||
try:
|
||
config = db.query(ServiceConfig).filter_by(
|
||
config_key=config_key
|
||
).first()
|
||
|
||
if config:
|
||
config.status = "inactive"
|
||
db.commit()
|
||
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"Failed to delete config: {str(e)}")
|
||
db.rollback()
|
||
return False
|
||
|
||
@staticmethod
|
||
def get_all_configs(db: Session, config_type: Optional[str] = None) -> List[Dict[str, Any]]:
|
||
"""获取所有配置
|
||
|
||
Args:
|
||
db: 数据库会话
|
||
config_type: 配置类型，可选
|
||
|
||
Returns:
|
||
配置列表
|
||
"""
|
||
query = db.query(ServiceConfig).filter_by(status="active")
|
||
|
||
if config_type:
|
||
query = query.filter_by(config_type=config_type)
|
||
|
||
configs = query.all()
|
||
|
||
return [
|
||
{
|
||
"id": config.id,
|
||
"key": config.config_key,
|
||
"value": config.config_value,
|
||
"type": config.config_type,
|
||
"service_id": config.service_id,
|
||
"description": config.description,
|
||
"created_at": config.created_at,
|
||
"updated_at": config.updated_at
|
||
}
|
||
for config in configs
|
||
]
|