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/'` 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 `/api//...` 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"/api/{service}{path}"] = proxied_operations return result