"""OpenAPI schema views that are safe under gevent concurrency. DRF's AutoSchema descriptor keeps mutable per-class state (``self.view``). When uWSGI serves requests with gevent, concurrent ``/api/schema/`` builds interleave on that state and raise: AssertionError: Schema generation REQUIRES a view instance Cold builds also frequently hang the gevent hub when introspection runs in a greenlet. Generation therefore runs in a real OS thread (gevent ThreadPool), with per-process single-flight coordination and a Django/Redis cache shared by all workers. """ from __future__ import annotations import copy import logging import threading from typing import Callable, Dict, Optional from django.conf import settings from django.core.cache import cache from rest_framework.response import Response from drf_spectacular.views import SpectacularAPIView logger = logging.getLogger(__name__) try: # Real OS threads; .get() waits without blocking the gevent hub. from gevent import monkey from gevent.threadpool import ThreadPool _SCHEMA_POOL: Optional[ThreadPool] = ThreadPool(maxsize=1) _GEVENT_PATCHED = monkey.is_module_patched("threading") except ImportError: # pragma: no cover - non-gevent runtimes _SCHEMA_POOL = None _GEVENT_PATCHED = False # Greenlet-safe after gevent monkey-patching. _guard = threading.Lock() _in_progress: Dict[str, threading.Event] = {} _flight_results: Dict[str, dict] = {} _SCHEMA_CACHE_PREFIX = "openapi:schema:" _SCHEMA_CACHE_VER_KEY = f"{_SCHEMA_CACHE_PREFIX}cache_ver" _BUILD_TIMEOUT_SECONDS = 120 def _run_schema_build(build: Callable[[], dict]) -> dict: """Run schema introspection off the gevent hub when monkey-patched.""" if _SCHEMA_POOL is not None and _GEVENT_PATCHED: return _SCHEMA_POOL.spawn(build).get(timeout=_BUILD_TIMEOUT_SECONDS) return build() def _cache_supports_delete_pattern() -> bool: return callable(getattr(cache, "delete_pattern", None)) def clear_schema_cache() -> None: """Invalidate cached schemas (tests / forced refresh after deploy).""" with _guard: _flight_results.clear() try: if _cache_supports_delete_pattern(): cache.delete_pattern(f"{_SCHEMA_CACHE_PREFIX}*") return ver = cache.get(_SCHEMA_CACHE_VER_KEY) or 0 cache.set(_SCHEMA_CACHE_VER_KEY, int(ver) + 1, timeout=None) except Exception: logger.warning("Failed to clear OpenAPI schema cache", exc_info=True) def _key_namespace() -> str: """Stable on Redis (pattern delete); versioned on LocMem for test clears.""" if _cache_supports_delete_pattern(): return "shared" try: ver = cache.get(_SCHEMA_CACHE_VER_KEY) if ver is None: cache.add(_SCHEMA_CACHE_VER_KEY, 1, timeout=None) ver = cache.get(_SCHEMA_CACHE_VER_KEY) or 1 return f"v{int(ver)}" except Exception: return "v1" class LockedSpectacularAPIView(SpectacularAPIView): """SpectacularAPIView with single-flight generation and Django cache.""" def _resolve_version(self, request): return self.api_version or request.version or self._get_version_parameter( request ) def _cache_key(self, request) -> str: version = self._resolve_version(request) lang = request.GET.get("lang") if settings.USE_I18N else None urlconf = self.urlconf if urlconf is None: urlconf_part = "default" else: urlconf_part = getattr(urlconf, "__name__", None) or type(urlconf).__name__ patterns_part = "custom" if self.patterns is not None else "default" return ( f"{_SCHEMA_CACHE_PREFIX}{_key_namespace()}:" f"{version}:{lang}:{self.serve_public}:{urlconf_part}:{patterns_part}" ) def _cache_get(self, key: str) -> Optional[dict]: try: return cache.get(key) except Exception: logger.warning("OpenAPI schema cache get failed", exc_info=True) return None def _cache_set(self, key: str, schema: dict) -> None: try: # Use CACHES['default'] TIMEOUT (3600s in settings). cache.set(key, schema) except Exception: logger.warning("OpenAPI schema cache set failed", exc_info=True) def _build_schema(self, request, version: Optional[str]) -> dict: generator = self.generator_class( urlconf=self.urlconf, api_version=version, patterns=self.patterns, ) return generator.get_schema(request=request, public=self.serve_public) def _get_schema_response(self, request): version = self._resolve_version(request) cache_key = self._cache_key(request) cached = self._cache_get(cache_key) if cached is not None: return self._schema_response(request, version, cached) with _guard: cached = self._cache_get(cache_key) if cached is not None: return self._schema_response(request, version, cached) event = _in_progress.get(cache_key) if event is None: event = threading.Event() _in_progress[cache_key] = event is_builder = True else: is_builder = False if not is_builder: if not event.wait(timeout=_BUILD_TIMEOUT_SECONDS): return Response( {"detail": "Timed out waiting for OpenAPI schema generation."}, status=503, ) cached = self._cache_get(cache_key) if cached is None: with _guard: cached = _flight_results.get(cache_key) if cached is None: return Response( {"detail": "OpenAPI schema generation failed."}, status=500, ) return self._schema_response(request, version, cached) try: schema = _run_schema_build( lambda: self._build_schema(request, version) ) with _guard: _flight_results[cache_key] = schema self._cache_set(cache_key, schema) return self._schema_response(request, version, schema) except Exception: logger.exception("OpenAPI schema generation failed") return Response( {"detail": "OpenAPI schema generation failed."}, status=500, ) finally: with _guard: _in_progress.pop(cache_key, None) event.set() def _schema_response(self, request, version, schema: dict) -> Response: # LocMem returns the same object; deepcopy keeps responses isolated. return Response( data=copy.deepcopy(schema), headers={ "Content-Disposition": ( f'inline; filename="{self._get_filename(request, version)}"' ) }, )