| 327 | return key |
| 328 | |
| 329 | def resolve_refs(self, schema: Dict[str, Any], url: str) -> Dict[str, Any]: |
| 330 | def visit(node: Any) -> Any: |
| 331 | if isinstance(node, list): |
| 332 | return [visit(child) for child in node] |
| 333 | if isinstance(node, dict): |
| 334 | ref = node.get("$ref") |
| 335 | if ref is not None and ref not in self._refs: |
| 336 | if ref.startswith("https://"): |
| 337 | raise ValueError("remote schema fetch is not allowed") |
| 338 | elif ref.startswith("#/"): |
| 339 | target = schema |
| 340 | ref = f"{url}{ref}" |
| 341 | node["$ref"] = ref |
| 342 | else: |
| 343 | raise ValueError(f"Unsupported ref {ref}") |
| 344 | |
| 345 | for selector in ref.split("#")[-1].split("/")[1:]: |
| 346 | assert target is not None and selector in target, ( |
| 347 | f"Error resolving ref {ref}: {selector} not in {target}" |
| 348 | ) |
| 349 | target = target[selector] |
| 350 | self._refs[ref] = target |
| 351 | else: |
| 352 | for value in node.values(): |
| 353 | visit(value) |
| 354 | return node |
| 355 | |
| 356 | return cast(Dict[str, Any], visit(schema)) |
| 357 | |
| 358 | def _generate_union_rule(self, name: str, alt_schemas: List[Dict[str, Any]]) -> str: |
| 359 | return " | ".join( |