"""Hydra ApiDocumentation surface — parse + operation lookup.""" from __future__ import annotations import json from dataclasses import dataclass, field from typing import Any, Dict, List, Optional @dataclass class Operation: """One Hydra operation — typically one tool on a JouleClaw service.""" id: str types: List[str] title: str description: str method: str expects: str returns: str status_codes: List[Dict[str, Any]] = field(default_factory=list) jc_tool_id: Optional[str] = None jc_mutating: Optional[bool] = None @classmethod def from_dict(cls, d: Dict[str, Any]) -> "Operation": return cls( id=d["@id"], types=d.get("@type", ["Operation"]), title=d.get("title", ""), description=d.get("description", ""), method=d.get("method", "GET"), expects=d.get("expects", "http://www.w3.org/2002/07/owl#Nothing"), returns=d.get("returns", ""), status_codes=d.get("statusCodes", []), jc_tool_id=d.get("jc_tool_id"), jc_mutating=d.get("jc_mutating"), ) @dataclass class Class: """One hydra:Class — groups operations.""" id: str title: str description: str supported_operation: List[Operation] = field(default_factory=list) @classmethod def from_dict(cls, d: Dict[str, Any]) -> "Class": return cls( id=d["@id"], title=d.get("title", ""), description=d.get("description", ""), supported_operation=[Operation.from_dict(o) for o in d.get("supportedOperation", [])], ) @dataclass class ApiDocumentation: """Root Hydra document — the contract a service publishes.""" id: str title: str description: str supported_class: List[Class] = field(default_factory=list) entrypoint: Optional[str] = None context: Optional[Any] = None @classmethod def from_dict(cls, d: Dict[str, Any]) -> "ApiDocumentation": return cls( id=d.get("@id", ""), title=d.get("title", ""), description=d.get("description", ""), supported_class=[Class.from_dict(c) for c in d.get("supportedClass", [])], entrypoint=d.get("entrypoint"), context=d.get("@context"), ) def parse_api(source: Any) -> ApiDocumentation: """Parse a Hydra ApiDocumentation. Accepts a dict, bytes, or str.""" if isinstance(source, (bytes, bytearray)): source = source.decode("utf-8") if isinstance(source, str): source = json.loads(source) if not isinstance(source, dict): raise TypeError(f"unsupported source type: {type(source)}") return ApiDocumentation.from_dict(source)