from __future__ import annotations from importlib import import_module from threading import Lock from django.apps import apps _CACHE = None _CACHE_LOCK = Lock() def list_error_code_groups(): global _CACHE if _CACHE is None: with _CACHE_LOCK: if _CACHE is None: _CACHE = _discover_error_code_groups() return _CACHE def clear_error_code_cache(): global _CACHE with _CACHE_LOCK: _CACHE = None def _discover_error_code_groups(): modules = ['flower'] modules.extend(config.name for config in apps.get_app_configs()) groups = [] seen_modules = set() for module_name in modules: if module_name in seen_modules: continue seen_modules.add(module_name) error_code_module = _import_error_code_module(module_name) if error_code_module is None: continue for group in getattr(error_code_module, 'ERROR_CODE_GROUPS', []): groups.append(_serialize_group(group)) return groups def _import_error_code_module(module_name: str): error_code_module_name = f'{module_name}.error_code' try: return import_module(error_code_module_name) except ModuleNotFoundError as exc: if exc.name == error_code_module_name: return None raise def _serialize_group(group: dict): enum_cls = group['enum'] details = group.get('details') or {} return { 'module': group['module'], 'title': group.get('title') or group['module'], 'codes': [_serialize_code(member, details.get(member, {})) for member in enum_cls], } def _serialize_code(member, detail: dict): return { 'error_code': int(member), 'name': member.name, 'code': detail.get('code', member.name.lower()), 'message': detail.get('message', ''), }