(source: CatalogSource)
| 89 | """ |
| 90 | |
| 91 | def fetch(source: CatalogSource) -> dict: |
| 92 | url = source.url |
| 93 | parsed = urlparse(url) |
| 94 | scheme = parsed.scheme.lower() |
| 95 | |
| 96 | if scheme == "builtin": |
| 97 | payload = _BUILTIN_CATALOGS.get(url) |
| 98 | if payload is None: |
| 99 | raise BundlerError(f"Unknown built-in catalog '{url}'.") |
| 100 | return payload |
| 101 | |
| 102 | if scheme == "file": |
| 103 | path = _file_url_to_path(parsed) |
| 104 | if not path.exists(): |
| 105 | raise BundlerError(f"Catalog file not found: {path}") |
| 106 | return loads_json(path.read_text(encoding="utf-8"), origin=str(path)) |
| 107 | |
| 108 | if scheme == "" or _is_windows_drive_path(url): |
| 109 | path = Path(url) |
| 110 | if not path.exists(): |
| 111 | raise BundlerError(f"Catalog file not found: {path}") |
| 112 | return loads_json(path.read_text(encoding="utf-8"), origin=str(path)) |
| 113 | |
| 114 | if scheme in ("http", "https"): |
| 115 | if not allow_network: |
| 116 | raise BundlerError( |
| 117 | f"Network access disabled; cannot fetch catalog '{source.id}' " |
| 118 | f"from {url}." |
| 119 | ) |
| 120 | _validate_remote_url(source.id, url) |
| 121 | return _http_get_json(source.id, url) |
| 122 | |
| 123 | raise BundlerError(f"Unsupported catalog URL scheme: {url}") |
| 124 | |
| 125 | return fetch |
| 126 |
nothing calls this directly
no test coverage detected