Function validate_workflow_content

server/services/workflow_storage.py:65–81

(filename: str, content: str)

def validate_workflow_content(filename: str, content: str) -> Tuple[str, Any]:
    safe_filename = validate_workflow_filename(filename, require_yaml_extension=True)

    try:
        yaml_content = yaml.safe_load(content)
        if yaml_content is None:
            raise ValidationError("YAML content is empty", field="content")

        errors = check_config(yaml_content)
        if errors:
            raise ValidationError(f"YAML validation errors:\n{errors}", field="content")
    except yaml.YAMLError as exc:
        logger = get_server_logger()
        logger.warning("Invalid YAML content in upload", details={"error": str(exc)})
        raise ValidationError(f"Invalid YAML syntax: {exc}", field="content")

    return safe_filename, yaml_content


def persist_workflow(