114 lines
4.1 KiB
Python
114 lines
4.1 KiB
Python
import concurrent.futures
|
|
import copy
|
|
import logging
|
|
|
|
import requests
|
|
from django.conf import settings
|
|
from django.core.cache import cache
|
|
|
|
from utils.http_fallback import request_with_ssl_fallback
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
CACHE_KEY_TEMPLATE = "proxy_upstream_openapi_schema_{service}"
|
|
CACHE_TIMEOUT_SECONDS = 300
|
|
FETCH_TIMEOUT_SECONDS = 5
|
|
|
|
|
|
def _rewrite_schema_refs(node, rename):
|
|
"""Recursively repoint `$ref: '#/components/schemas/<Name>'` at the
|
|
renamed (service-prefixed) schema, in place."""
|
|
if isinstance(node, dict):
|
|
ref = node.get("$ref")
|
|
if isinstance(ref, str) and ref.startswith("#/components/schemas/"):
|
|
name = ref.rsplit("/", 1)[-1]
|
|
if name in rename:
|
|
node["$ref"] = f"#/components/schemas/{rename[name]}"
|
|
for value in node.values():
|
|
_rewrite_schema_refs(value, rename)
|
|
elif isinstance(node, list):
|
|
for item in node:
|
|
_rewrite_schema_refs(item, rename)
|
|
|
|
|
|
def _fetch_service_schema(service, base_url, schema_path):
|
|
cache_key = CACHE_KEY_TEMPLATE.format(service=service)
|
|
cached = cache.get(cache_key)
|
|
if cached is not None:
|
|
return cached
|
|
|
|
url = f"{base_url.rstrip('/')}{schema_path}"
|
|
try:
|
|
response = request_with_ssl_fallback("GET", url, timeout=FETCH_TIMEOUT_SECONDS)
|
|
response.raise_for_status()
|
|
schema = response.json()
|
|
except (requests.RequestException, ValueError) as exc:
|
|
logger.warning("Could not fetch OpenAPI schema for service %r from %s: %s", service, url, exc)
|
|
return None
|
|
|
|
cache.set(cache_key, schema, timeout=CACHE_TIMEOUT_SECONDS)
|
|
return schema
|
|
|
|
|
|
def merge_upstream_schemas(result, generator, request, public):
|
|
"""drf-spectacular POSTPROCESSING_HOOKS entry point.
|
|
|
|
Folds every proxied service's own OpenAPI document into this service's
|
|
schema under `/<service>/...` paths (mirroring apps.proxy.views), so the
|
|
combined swagger covers every endpoint this proxy forwards to, DTOs
|
|
included. Schemas are cached per service; a service that's unconfigured
|
|
or unreachable is skipped rather than failing the whole document.
|
|
"""
|
|
services = {
|
|
service: base_url
|
|
for service, base_url in settings.PROXY_SERVICE_BASE_URLS.items()
|
|
if base_url
|
|
}
|
|
if not services:
|
|
return result
|
|
|
|
schema_paths = settings.PROXY_SERVICE_SCHEMA_PATHS
|
|
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=len(services)) as executor:
|
|
futures = {
|
|
executor.submit(
|
|
_fetch_service_schema,
|
|
service,
|
|
base_url,
|
|
schema_paths.get(service, "/swagger/?format=json"),
|
|
): service
|
|
for service, base_url in services.items()
|
|
}
|
|
fetched = {}
|
|
for future in concurrent.futures.as_completed(futures):
|
|
service = futures[future]
|
|
schema = future.result()
|
|
if schema:
|
|
fetched[service] = schema
|
|
|
|
result.setdefault("paths", {})
|
|
result.setdefault("components", {}).setdefault("schemas", {})
|
|
|
|
for service, schema in fetched.items():
|
|
schema = copy.deepcopy(schema)
|
|
upstream_schemas = schema.get("components", {}).get("schemas", {})
|
|
rename = {name: f"{service}_{name}" for name in upstream_schemas}
|
|
|
|
_rewrite_schema_refs(schema.get("paths", {}), rename)
|
|
_rewrite_schema_refs(upstream_schemas, rename)
|
|
|
|
for name, definition in upstream_schemas.items():
|
|
result["components"]["schemas"][rename[name]] = definition
|
|
|
|
label = service.capitalize()
|
|
for path, operations in schema.get("paths", {}).items():
|
|
proxied_operations = copy.deepcopy(operations)
|
|
for operation in proxied_operations.values():
|
|
if not isinstance(operation, dict):
|
|
continue
|
|
operation["tags"] = [label]
|
|
if "operationId" in operation:
|
|
operation["operationId"] = f"{service}_{operation['operationId']}"
|
|
result["paths"][f"/{service}{path}"] = proxied_operations
|
|
|
|
return result
|