326 lines
10 KiB
Python
326 lines
10 KiB
Python
"""
|
||
Metadata Schema Service.
|
||
动态元数据模式管理服务，支持租户自定义元数据字段配置。
|
||
"""
|
||
|
||
import logging
|
||
import uuid
|
||
from datetime import datetime
|
||
from typing import Any
|
||
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from sqlmodel import col
|
||
|
||
from app.models.entities import (
|
||
MetadataField,
|
||
MetadataFieldType,
|
||
MetadataSchema,
|
||
MetadataSchemaCreate,
|
||
MetadataSchemaUpdate,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class MetadataSchemaService:
|
||
"""
|
||
元数据模式服务
|
||
管理租户的动态元数据字段配置
|
||
"""
|
||
|
||
def __init__(self, session: AsyncSession):
|
||
self._session = session
|
||
|
||
async def get_schema(
|
||
self,
|
||
tenant_id: str,
|
||
schema_id: str | None = None,
|
||
) -> MetadataSchema | None:
|
||
"""
|
||
获取元数据模式
|
||
|
||
Args:
|
||
tenant_id: 租户 ID
|
||
schema_id: 模式 ID（可选，不传则获取默认模式）
|
||
|
||
Returns:
|
||
MetadataSchema 或 None
|
||
"""
|
||
if schema_id:
|
||
stmt = select(MetadataSchema).where(
|
||
MetadataSchema.tenant_id == tenant_id,
|
||
MetadataSchema.id == uuid.UUID(schema_id),
|
||
)
|
||
else:
|
||
stmt = select(MetadataSchema).where(
|
||
MetadataSchema.tenant_id == tenant_id,
|
||
MetadataSchema.is_default == True,
|
||
MetadataSchema.is_enabled == True,
|
||
).order_by(col(MetadataSchema.created_at).desc())
|
||
|
||
result = await self._session.execute(stmt)
|
||
return result.scalar_one_or_none()
|
||
|
||
async def list_schemas(
|
||
self,
|
||
tenant_id: str,
|
||
include_disabled: bool = False,
|
||
) -> list[MetadataSchema]:
|
||
"""
|
||
列出租户所有元数据模式
|
||
|
||
Args:
|
||
tenant_id: 租户 ID
|
||
include_disabled: 是否包含禁用的模式
|
||
|
||
Returns:
|
||
MetadataSchema 列表
|
||
"""
|
||
stmt = select(MetadataSchema).where(
|
||
MetadataSchema.tenant_id == tenant_id,
|
||
)
|
||
if not include_disabled:
|
||
stmt = stmt.where(MetadataSchema.is_enabled == True)
|
||
|
||
stmt = stmt.order_by(col(MetadataSchema.created_at).desc())
|
||
|
||
result = await self._session.execute(stmt)
|
||
return list(result.scalars().all())
|
||
|
||
async def create_schema(
|
||
self,
|
||
tenant_id: str,
|
||
schema_create: MetadataSchemaCreate,
|
||
) -> MetadataSchema:
|
||
"""
|
||
创建元数据模式
|
||
|
||
Args:
|
||
tenant_id: 租户 ID
|
||
schema_create: 创建数据
|
||
|
||
Returns:
|
||
创建的 MetadataSchema
|
||
"""
|
||
schema = MetadataSchema(
|
||
tenant_id=tenant_id,
|
||
name=schema_create.name,
|
||
description=schema_create.description,
|
||
fields=[f.model_dump() if hasattr(f, 'model_dump') else f for f in schema_create.fields],
|
||
is_default=schema_create.is_default,
|
||
is_enabled=True,
|
||
)
|
||
|
||
self._session.add(schema)
|
||
await self._session.flush()
|
||
|
||
logger.info(
|
||
f"[MetadataSchemaService] Created schema: tenant={tenant_id}, "
|
||
f"name={schema.name}, fields_count={len(schema.fields)}"
|
||
)
|
||
|
||
return schema
|
||
|
||
async def update_schema(
|
||
self,
|
||
tenant_id: str,
|
||
schema_id: str,
|
||
schema_update: MetadataSchemaUpdate,
|
||
) -> MetadataSchema | None:
|
||
"""
|
||
更新元数据模式
|
||
|
||
Args:
|
||
tenant_id: 租户 ID
|
||
schema_id: 模式 ID
|
||
schema_update: 更新数据
|
||
|
||
Returns:
|
||
更新后的 MetadataSchema 或 None
|
||
"""
|
||
stmt = select(MetadataSchema).where(
|
||
MetadataSchema.tenant_id == tenant_id,
|
||
MetadataSchema.id == uuid.UUID(schema_id),
|
||
)
|
||
result = await self._session.execute(stmt)
|
||
schema = result.scalar_one_or_none()
|
||
|
||
if not schema:
|
||
return None
|
||
|
||
if schema_update.name is not None:
|
||
schema.name = schema_update.name
|
||
if schema_update.description is not None:
|
||
schema.description = schema_update.description
|
||
if schema_update.fields is not None:
|
||
schema.fields = schema_update.fields
|
||
if schema_update.is_default is not None:
|
||
if schema_update.is_default:
|
||
await self._unset_other_defaults(tenant_id, schema_id)
|
||
schema.is_default = True
|
||
if schema_update.is_enabled is not None:
|
||
schema.is_enabled = schema_update.is_enabled
|
||
|
||
schema.updated_at = datetime.utcnow()
|
||
await self._session.flush()
|
||
|
||
logger.info(
|
||
f"[MetadataSchemaService] Updated schema: tenant={tenant_id}, "
|
||
f"schema_id={schema_id}"
|
||
)
|
||
|
||
return schema
|
||
|
||
async def delete_schema(
|
||
self,
|
||
tenant_id: str,
|
||
schema_id: str,
|
||
) -> bool:
|
||
"""
|
||
删除元数据模式
|
||
|
||
Args:
|
||
tenant_id: 租户 ID
|
||
schema_id: 模式 ID
|
||
|
||
Returns:
|
||
是否删除成功
|
||
"""
|
||
stmt = select(MetadataSchema).where(
|
||
MetadataSchema.tenant_id == tenant_id,
|
||
MetadataSchema.id == uuid.UUID(schema_id),
|
||
)
|
||
result = await self._session.execute(stmt)
|
||
schema = result.scalar_one_or_none()
|
||
|
||
if not schema:
|
||
return False
|
||
|
||
if schema.is_default:
|
||
logger.warning(
|
||
f"[MetadataSchemaService] Cannot delete default schema: "
|
||
f"tenant={tenant_id}, schema_id={schema_id}"
|
||
)
|
||
return False
|
||
|
||
await self._session.delete(schema)
|
||
await self._session.flush()
|
||
|
||
logger.info(
|
||
f"[MetadataSchemaService] Deleted schema: tenant={tenant_id}, "
|
||
f"schema_id={schema_id}"
|
||
)
|
||
|
||
return True
|
||
|
||
async def _unset_other_defaults(
|
||
self,
|
||
tenant_id: str,
|
||
exclude_schema_id: str,
|
||
) -> None:
|
||
"""取消其他模式的默认状态"""
|
||
stmt = select(MetadataSchema).where(
|
||
MetadataSchema.tenant_id == tenant_id,
|
||
MetadataSchema.is_default == True,
|
||
MetadataSchema.id != uuid.UUID(exclude_schema_id),
|
||
)
|
||
result = await self._session.execute(stmt)
|
||
other_schemas = result.scalars().all()
|
||
|
||
for other in other_schemas:
|
||
other.is_default = False
|
||
|
||
async def get_field_definitions(
|
||
self,
|
||
tenant_id: str,
|
||
schema_id: str | None = None,
|
||
) -> dict[str, dict[str, Any]]:
|
||
"""
|
||
获取字段定义映射
|
||
|
||
Args:
|
||
tenant_id: 租户 ID
|
||
schema_id: 模式 ID（可选）
|
||
|
||
Returns:
|
||
字段名到字段定义的映射，如 {"grade": {"label": "年级", "type": "select", "options": [...]}}
|
||
"""
|
||
schema = await self.get_schema(tenant_id, schema_id)
|
||
if not schema:
|
||
return {}
|
||
|
||
field_map = {}
|
||
for field in schema.fields:
|
||
field_name = field.get("name") if isinstance(field, dict) else field.name
|
||
field_map[field_name] = field if isinstance(field, dict) else field.model_dump()
|
||
|
||
return field_map
|
||
|
||
async def validate_metadata(
|
||
self,
|
||
tenant_id: str,
|
||
metadata: dict[str, Any],
|
||
schema_id: str | None = None,
|
||
) -> tuple[bool, list[str]]:
|
||
"""
|
||
验证元数据是否符合模式定义
|
||
|
||
Args:
|
||
tenant_id: 租户 ID
|
||
metadata: 元数据字典
|
||
schema_id: 模式 ID（可选）
|
||
|
||
Returns:
|
||
(是否有效, 错误消息列表)
|
||
"""
|
||
field_defs = await self.get_field_definitions(tenant_id, schema_id)
|
||
|
||
if not field_defs:
|
||
return True, []
|
||
|
||
errors = []
|
||
|
||
for field_name, field_def in field_defs.items():
|
||
field_type = field_def.get("field_type", "string")
|
||
required = field_def.get("required", False)
|
||
options = field_def.get("options", [])
|
||
value = metadata.get(field_name)
|
||
|
||
if required and value is None:
|
||
errors.append(f"字段 '{field_def.get('label', field_name)}' 是必填的")
|
||
continue
|
||
|
||
if value is None:
|
||
continue
|
||
|
||
if field_type == MetadataFieldType.SELECT.value:
|
||
if value not in options:
|
||
errors.append(
|
||
f"字段 '{field_def.get('label', field_name)}' 的值 '{value}' 不在允许选项中"
|
||
)
|
||
|
||
elif field_type == MetadataFieldType.MULTI_SELECT.value:
|
||
if not isinstance(value, list):
|
||
errors.append(f"字段 '{field_def.get('label', field_name)}' 应该是多选值")
|
||
else:
|
||
for v in value:
|
||
if v not in options:
|
||
errors.append(
|
||
f"字段 '{field_def.get('label', field_name)}' 的值 '{v}' 不在允许选项中"
|
||
)
|
||
|
||
elif field_type == MetadataFieldType.NUMBER.value:
|
||
if not isinstance(value, (int, float)):
|
||
try:
|
||
float(value)
|
||
except (ValueError, TypeError):
|
||
errors.append(f"字段 '{field_def.get('label', field_name)}' 应该是数字")
|
||
|
||
elif field_type == MetadataFieldType.BOOLEAN.value:
|
||
if not isinstance(value, bool):
|
||
if value not in ["true", "false", "1", "0", 1, 0]:
|
||
errors.append(f"字段 '{field_def.get('label', field_name)}' 应该是布尔值")
|
||
|
||
return len(errors) == 0, errors
|