178 lines
7.0 KiB
Python
178 lines
7.0 KiB
Python
"""把 Swagger 2 / OpenAPI 3 编译为连接器使用的统一 operation catalog。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
HTTP_METHODS = (
|
|
"get",
|
|
"head",
|
|
"post",
|
|
"put",
|
|
"patch",
|
|
"delete",
|
|
"options",
|
|
"trace",
|
|
)
|
|
|
|
|
|
def resolve_local_object(spec: dict[str, Any], value: Any) -> Any:
|
|
"""解析单个本地 JSON Pointer 引用；远端引用保留原值并由上层拒绝。"""
|
|
if not isinstance(value, dict):
|
|
return value
|
|
ref = value.get("$ref")
|
|
if not isinstance(ref, str) or not ref.startswith("#/"):
|
|
return value
|
|
current: Any = spec
|
|
for raw in ref[2:].split("/"):
|
|
part = raw.replace("~1", "/").replace("~0", "~")
|
|
if not isinstance(current, dict) or part not in current:
|
|
return value
|
|
current = current[part]
|
|
return current if isinstance(current, dict) else value
|
|
|
|
|
|
def operation_id(method: str, path: str, operation: dict[str, Any]) -> str:
|
|
explicit = operation.get("operationId")
|
|
if isinstance(explicit, str) and explicit.strip():
|
|
return explicit.strip()
|
|
safe_path = re.sub(r"[^a-zA-Z0-9]+", "_", path).strip("_")
|
|
return f"{method}_{safe_path}"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class OperationCatalog:
|
|
operations: tuple[dict[str, Any], ...]
|
|
|
|
def find(self, operation_id_value: str) -> tuple[dict[str, Any], ...]:
|
|
return tuple(
|
|
operation
|
|
for operation in self.operations
|
|
if operation["operation_id"] == operation_id_value
|
|
)
|
|
|
|
|
|
def compile_operation_catalog(spec: dict[str, Any]) -> OperationCatalog:
|
|
results: list[dict[str, Any]] = []
|
|
for path, path_item in (spec.get("paths") or {}).items():
|
|
if not isinstance(path, str) or not isinstance(path_item, dict):
|
|
continue
|
|
common = path_item.get("parameters") or []
|
|
for method in HTTP_METHODS:
|
|
operation = path_item.get(method)
|
|
if not isinstance(operation, dict):
|
|
continue
|
|
params = [
|
|
resolve_local_object(spec, param)
|
|
for param in list(common) + list(operation.get("parameters") or [])
|
|
]
|
|
results.append(
|
|
{
|
|
"operation_id": operation_id(method, path, operation),
|
|
"method": method.upper(),
|
|
"path": path,
|
|
"summary": operation.get("summary") or "",
|
|
"description": operation.get("description") or "",
|
|
"tags": operation.get("tags") or [],
|
|
"parameters": params,
|
|
"request_body": resolve_local_object(
|
|
spec, operation.get("requestBody")
|
|
),
|
|
}
|
|
)
|
|
return OperationCatalog(tuple(results))
|
|
|
|
|
|
def validate_json_value(
|
|
spec: dict[str, Any],
|
|
value: Any,
|
|
schema: Any,
|
|
*,
|
|
path: str = "body",
|
|
depth: int = 0,
|
|
) -> None:
|
|
"""验证调用前最关键的 JSON Schema 子集，复杂语义仍由上游最终判定。"""
|
|
if depth > 20 or not isinstance(schema, dict):
|
|
return
|
|
schema = resolve_local_object(spec, schema)
|
|
expected = schema.get("type")
|
|
type_ok = {
|
|
"object": isinstance(value, dict),
|
|
"array": isinstance(value, list),
|
|
"string": isinstance(value, str),
|
|
"integer": isinstance(value, int) and not isinstance(value, bool),
|
|
"number": isinstance(value, (int, float)) and not isinstance(value, bool),
|
|
"boolean": isinstance(value, bool),
|
|
"null": value is None,
|
|
}
|
|
if expected in type_ok and not type_ok[expected]:
|
|
raise ValueError(f"{path} 应为 {expected}")
|
|
if "enum" in schema and value not in schema.get("enum", []):
|
|
raise ValueError(f"{path} 不在允许值范围内")
|
|
if isinstance(value, (int, float)) and not isinstance(value, bool):
|
|
minimum = schema.get("minimum")
|
|
maximum = schema.get("maximum")
|
|
exclusive_minimum = schema.get("exclusiveMinimum")
|
|
exclusive_maximum = schema.get("exclusiveMaximum")
|
|
if isinstance(minimum, (int, float)):
|
|
if value < minimum or (exclusive_minimum is True and value == minimum):
|
|
operator = ">" if exclusive_minimum is True else ">="
|
|
raise ValueError(f"{path} 必须 {operator} {minimum}")
|
|
if isinstance(maximum, (int, float)):
|
|
if value > maximum or (exclusive_maximum is True and value == maximum):
|
|
operator = "<" if exclusive_maximum is True else "<="
|
|
raise ValueError(f"{path} 必须 {operator} {maximum}")
|
|
if isinstance(exclusive_minimum, (int, float)) and not isinstance(
|
|
exclusive_minimum, bool
|
|
):
|
|
if value <= exclusive_minimum:
|
|
raise ValueError(f"{path} 必须 > {exclusive_minimum}")
|
|
if isinstance(exclusive_maximum, (int, float)) and not isinstance(
|
|
exclusive_maximum, bool
|
|
):
|
|
if value >= exclusive_maximum:
|
|
raise ValueError(f"{path} 必须 < {exclusive_maximum}")
|
|
if isinstance(value, str):
|
|
minimum_length = schema.get("minLength")
|
|
maximum_length = schema.get("maxLength")
|
|
if isinstance(minimum_length, int) and len(value) < minimum_length:
|
|
raise ValueError(f"{path} 长度不能小于 {minimum_length}")
|
|
if isinstance(maximum_length, int) and len(value) > maximum_length:
|
|
raise ValueError(f"{path} 长度不能超过 {maximum_length}")
|
|
if isinstance(value, list):
|
|
minimum_items = schema.get("minItems")
|
|
maximum_items = schema.get("maxItems")
|
|
if isinstance(minimum_items, int) and len(value) < minimum_items:
|
|
raise ValueError(f"{path} 项数不能少于 {minimum_items}")
|
|
if isinstance(maximum_items, int) and len(value) > maximum_items:
|
|
raise ValueError(f"{path} 项数不能超过 {maximum_items}")
|
|
if isinstance(value, dict):
|
|
required = schema.get("required") or []
|
|
missing = [str(name) for name in required if name not in value]
|
|
if missing:
|
|
raise ValueError(f"{path} 缺少必填字段: " + ", ".join(missing))
|
|
properties = schema.get("properties") or {}
|
|
if isinstance(properties, dict):
|
|
for name, child in value.items():
|
|
if name in properties:
|
|
validate_json_value(
|
|
spec,
|
|
child,
|
|
properties[name],
|
|
path=f"{path}.{name}",
|
|
depth=depth + 1,
|
|
)
|
|
elif schema.get("additionalProperties") is False:
|
|
raise ValueError(f"{path} 包含未定义字段: {name}")
|
|
if isinstance(value, list) and isinstance(schema.get("items"), dict):
|
|
for index, item in enumerate(value):
|
|
validate_json_value(
|
|
spec,
|
|
item,
|
|
schema["items"],
|
|
path=f"{path}[{index}]",
|
|
depth=depth + 1,
|
|
)
|