52 lines
2.2 KiB
Python
52 lines
2.2 KiB
Python
# utils/validator.py
|
|
from typing import Dict, Any, List, Optional, Callable
|
|
from core.exceptions import ValidationError
|
|
|
|
def validate_required_fields(data: Dict[str, Any], required_fields: List[str]) -> None:
|
|
"""验证必要字段"""
|
|
missing_fields = [field for field in required_fields if field not in data]
|
|
|
|
if missing_fields:
|
|
raise ValidationError(f"缺少必要字段: {', '.join(missing_fields)}")
|
|
|
|
def validate_field_type(data: Dict[str, Any], field: str, expected_type: type) -> None:
|
|
"""验证字段类型"""
|
|
if field in data and not isinstance(data[field], expected_type):
|
|
raise ValidationError(f"字段 {field} 类型错误，期望类型: {expected_type.__name__}")
|
|
|
|
def validate_field_value(data: Dict[str, Any], field: str, validator: Callable[[Any], bool], error_message: str) -> None:
|
|
"""验证字段值"""
|
|
if field in data and not validator(data[field]):
|
|
raise ValidationError(f"字段 {field} 值无效: {error_message}")
|
|
|
|
def validate_workflow_definition(workflow_data: Dict[str, Any]) -> None:
|
|
"""验证工作流定义"""
|
|
# 验证必要字段
|
|
validate_required_fields(workflow_data, ["name"])
|
|
|
|
# 验证字段类型
|
|
validate_field_type(workflow_data, "name", str)
|
|
validate_field_type(workflow_data, "workflow_type", str)
|
|
validate_field_type(workflow_data, "blocks", list)
|
|
validate_field_type(workflow_data, "variables", dict)
|
|
|
|
# 验证工作流类型
|
|
if "workflow_type" in workflow_data:
|
|
valid_types = ["normal", "scheduled"]
|
|
if workflow_data["workflow_type"] not in valid_types:
|
|
raise ValidationError(f"无效的工作流类型: {workflow_data['workflow_type']}, 有效类型: {', '.join(valid_types)}")
|
|
|
|
# 验证块定义
|
|
if "blocks" in workflow_data:
|
|
for i, block in enumerate(workflow_data["blocks"]):
|
|
validate_block_definition(block, i)
|
|
|
|
def validate_block_definition(block: Dict[str, Any], index: int) -> None:
|
|
"""验证块定义"""
|
|
# 验证必要字段
|
|
validate_required_fields(block, ["id", "type"])
|
|
|
|
# 验证字段类型
|
|
validate_field_type(block, "id", str)
|
|
validate_field_type(block, "type", str)
|
|
validate_field_type(block, "params", dict) |