""" Утилиты для документирования API (OpenAPI/Swagger). Предоставляет декораторы и утилиты для улучшения автоматически генерируемой документации. """ from collections import OrderedDict from typing import Any from django.conf import settings from drf_yasg import openapi from drf_yasg.generators import OpenAPISchemaGenerator from drf_yasg.utils import swagger_auto_schema OPENAPI_PROJECT_TITLE = ( "СПО «Программный сервис, обеспечивающий обработку информации в защищенной " "части ГИСП на основе данных организаций ОПК, входящих в состав " "Госкорпорации «Росатом» и Госкорпорации «Роскосмос»" ) OPENAPI_PROJECT_DESCRIPTION = """ ## Документация API сервиса Сервис предоставляет доступ к данным организаций ОПК, отчетным формам, реестрам, статусам фоновых задач и служебным маршрутам мониторинга. ### Авторизация Для доступа к защищённым маршрутам используйте JWT-токен: 1. Получите токен через `POST /api/v1/users/login/` 2. Передавайте заголовок `Authorization: Bearer ` ### Обновление токена Для выпуска нового access-токена используйте `POST /api/v1/users/token/refresh/`. ### Общий принцип работы API предназначен в основном для чтения и мониторинга данных. Загрузка отчетных форм и импорт реестров выполняются через специализированные маршруты и фоновые задачи. """.strip() OPENAPI_TAG_DESCRIPTIONS = OrderedDict( [ ( "Аутентификация", "Регистрация, вход, выход, обновление и проверка JWT-токенов.", ), ( "Пользователь", "Текущий пользователь, профиль, смена пароля и операции личного кабинета.", ), ( "Организации", "Справочник организаций ОПК с актуальными данными и связями по реестрам.", ), ( "Аналитика", "Агрегированные аналитические endpoint'ы по организации и корпорации.", ), ( "Реестры", "Просмотр реестров, составов организаций и импорт резервных копий реестров.", ), ( "Внешние данные", "Внешние контуры и реестры: продукция, проверки, закупки и арбитраж.", ), ( "Обмен данными", "Импорт зашифрованных exchange-пакетов из dev API и ручной доставки.", ), ( "Форма Ф-1", "Загрузка и просмотр формы Ф-1 по выпуску продукции, НИОКР и кадровым показателям.", ), ( "Форма Ф-2", "Загрузка и просмотр формы Ф-2 с бухгалтерским балансом и финансовыми результатами.", ), ( "Форма Ф-3", "Загрузка и просмотр формы Ф-3 по кадрам, оборудованию и износу.", ), ( "Форма Ф-4", "Загрузка и просмотр формы Ф-4 со сводными финансовыми показателями.", ), ( "Форма Ф-5", "Загрузка и просмотр формы Ф-5 по единицам оборудования и их состоянию.", ), ( "Форма Ф-6", "Загрузка и просмотр формы Ф-6 по возрастной структуре оборудования и загрузке.", ), ( "Фоновые задачи", "Контроль статусов, прогресса и результатов длительных операций импорта и обработки.", ), ( "Мониторинг", "Служебные маршруты для проверки доступности, готовности и общего состояния системы.", ), ] ) OPENAPI_AUTH_PATH_PREFIXES = ( "/api/v1/users/register/", "/api/v1/users/login/", "/api/v1/users/logout/", "/api/v1/users/token/refresh/", "/api/v1/users/token/verify/", ) OPENAPI_PUBLIC_PATH_PREFIXES = ( "/health/", "/api/v1/users/register/", "/api/v1/users/login/", "/api/v1/users/token/refresh/", "/api/v1/users/token/verify/", "/auth/login/", ) OPENAPI_TAG_BY_PATH_PREFIX = OrderedDict( [ ("/health/", "Мониторинг"), ("/api/v1/analytics/", "Аналитика"), ("/api/v1/jobs/", "Фоновые задачи"), ("/api/v1/organizations/", "Организации"), ("/api/v1/exchange/", "Обмен данными"), ("/api/v1/industrial-products/", "Внешние данные"), ("/api/v1/prosecutor-checks/", "Внешние данные"), ("/api/v1/public-procurements/", "Внешние данные"), ("/api/v1/arbitration-cases/", "Внешние данные"), ("/api/v1/security-registries/", "Внешние данные"), ("/api/v1/registers/", "Реестры"), ("/api/v1/forms/f1/", "Форма Ф-1"), ("/api/v1/forms/f2/", "Форма Ф-2"), ("/api/v1/forms/f3/", "Форма Ф-3"), ("/api/v1/forms/f4/", "Форма Ф-4"), ("/api/v1/forms/f5/", "Форма Ф-5"), ("/api/v1/forms/f6/", "Форма Ф-6"), ] ) OPENAPI_TAG_ALIASES = { "monitoring": "Мониторинг", "background_jobs": "Фоновые задачи", "authentication": "Аутентификация", "auth": "Аутентификация", "user": "Пользователь", "users": "Пользователь", "organizations": "Организации", "registers": "Реестры", "exchange": "Обмен данными", "api": None, } class RussianTagSchemaGenerator(OpenAPISchemaGenerator): """OpenAPI generator with Russian tags and tag descriptions.""" def get_schema(self, request=None, public=False): schema = super().get_schema(request=request, public=public) self._localize_tags(schema) schema.tags = [ {"name": name, "description": description} for name, description in OPENAPI_TAG_DESCRIPTIONS.items() ] return schema def _localize_tags(self, schema) -> None: for path, path_item in schema.paths.items(): resolved_tag = self._resolve_tag_for_path(path) for method in ("get", "post", "put", "patch", "delete", "head", "options"): operation = path_item.get(method) if operation is None: continue tag = resolved_tag or self._localize_existing_tag( getattr(operation, "tags", None) ) if tag is not None: operation.tags = [tag] if self._is_public_path(path): operation.security = [] def _resolve_tag_for_path(self, path: str) -> str | None: for prefix in OPENAPI_AUTH_PATH_PREFIXES: if path.startswith(prefix): return "Аутентификация" if path.startswith("/api/v1/users/"): return "Пользователь" for prefix, tag in OPENAPI_TAG_BY_PATH_PREFIX.items(): if path.startswith(prefix): return tag return None @staticmethod def _is_public_path(path: str) -> bool: return any(path.startswith(prefix) for prefix in OPENAPI_PUBLIC_PATH_PREFIXES) @staticmethod def _localize_existing_tag(tags: list[str] | None) -> str | None: if not tags: return None for tag in tags: if tag in OPENAPI_TAG_DESCRIPTIONS: return tag if tag in OPENAPI_TAG_ALIASES: return OPENAPI_TAG_ALIASES[tag] return tags[0] def api_docs( *, summary: str, description: str | None = None, request_body: Any = None, responses: dict[int, Any] | None = None, tags: list[str] | None = None, operation_id: str | None = None, deprecated: bool = False, security: list[dict[str, list[str]]] | None = None, manual_parameters: list[openapi.Parameter] | None = None, ): """ Декоратор для документирования API эндпоинтов. Упрощённая обёртка над swagger_auto_schema с поддержкой типовых паттернов документирования. Args: summary: Краткое описание эндпоинта (отображается в списке) description: Подробное описание (отображается при раскрытии) request_body: Схема тела запроса (serializer или openapi.Schema) responses: Словарь возможных ответов {status_code: schema} tags: Теги для группировки в документации operation_id: Уникальный идентификатор операции deprecated: Пометить как устаревший security: Требования безопасности manual_parameters: Дополнительные параметры запроса Пример использования: class UserView(APIView): @api_docs( summary="Получить текущего пользователя", description="Возвращает данные аутентифицированного пользователя", responses={ 200: UserSerializer, 401: "Не авторизован", }, tags=["Пользователи"], ) def get(self, request): ... """ # Преобразуем упрощённые responses в формат openapi formatted_responses = {} if responses: for code, schema in responses.items(): if isinstance(schema, str): # Простое текстовое описание formatted_responses[code] = openapi.Response(description=schema) elif isinstance(schema, type): # Serializer class formatted_responses[code] = openapi.Response( description=_get_status_description(code), schema=schema, ) elif isinstance(schema, openapi.Response): formatted_responses[code] = schema else: formatted_responses[code] = schema return swagger_auto_schema( operation_summary=summary, operation_description=description, request_body=request_body, responses=formatted_responses or None, tags=tags, operation_id=operation_id, deprecated=deprecated, security=security, manual_parameters=manual_parameters, ) def swagger_tag(ru: str, en: str | None = None) -> str: """Возвращает тег для Swagger в зависимости от текущих настроек.""" use_english = getattr(settings, "OPENAPI_USE_ENGLISH_TAGS", False) if use_english and en: return en return ru def _get_status_description(status_code: int) -> str: """Возвращает описание HTTP статуса на русском.""" descriptions = { 200: "Успешный запрос", 201: "Ресурс создан", 202: "Запрос принят в обработку", 204: "Успешно, без содержимого", 400: "Некорректный запрос", 401: "Не авторизован", 403: "Доступ запрещён", 404: "Ресурс не найден", 409: "Конфликт", 422: "Ошибка валидации", 429: "Слишком много запросов", 500: "Внутренняя ошибка сервера", 503: "Сервис временно недоступен", } return descriptions.get(status_code, f"HTTP {status_code}") # Предопределённые схемы ответов class CommonResponses: """ Общие схемы ответов для документации. Пример использования: @api_docs( summary="Удалить ресурс", responses={ 204: CommonResponses.NO_CONTENT, 404: CommonResponses.NOT_FOUND, }, ) def delete(self, request, pk): ... """ SUCCESS = openapi.Response( description="Успешный запрос", schema=openapi.Schema( type=openapi.TYPE_OBJECT, properties={ "success": openapi.Schema(type=openapi.TYPE_BOOLEAN, default=True), "data": openapi.Schema(type=openapi.TYPE_OBJECT), }, ), ) CREATED = openapi.Response( description="Ресурс успешно создан", schema=openapi.Schema( type=openapi.TYPE_OBJECT, properties={ "success": openapi.Schema(type=openapi.TYPE_BOOLEAN, default=True), "data": openapi.Schema(type=openapi.TYPE_OBJECT), }, ), ) NO_CONTENT = openapi.Response(description="Успешно, без содержимого") BAD_REQUEST = openapi.Response( description="Некорректный запрос", schema=openapi.Schema( type=openapi.TYPE_OBJECT, properties={ "success": openapi.Schema(type=openapi.TYPE_BOOLEAN, default=False), "errors": openapi.Schema( type=openapi.TYPE_ARRAY, items=openapi.Schema( type=openapi.TYPE_OBJECT, properties={ "code": openapi.Schema(type=openapi.TYPE_STRING), "message": openapi.Schema(type=openapi.TYPE_STRING), }, ), ), }, ), ) UNAUTHORIZED = openapi.Response( description="Требуется аутентификация", schema=openapi.Schema( type=openapi.TYPE_OBJECT, properties={ "detail": openapi.Schema( type=openapi.TYPE_STRING, default="Учётные данные не предоставлены.", ), }, ), ) FORBIDDEN = openapi.Response( description="Доступ запрещён", schema=openapi.Schema( type=openapi.TYPE_OBJECT, properties={ "detail": openapi.Schema( type=openapi.TYPE_STRING, default="У вас нет прав для выполнения этого действия.", ), }, ), ) NOT_FOUND = openapi.Response( description="Ресурс не найден", schema=openapi.Schema( type=openapi.TYPE_OBJECT, properties={ "success": openapi.Schema(type=openapi.TYPE_BOOLEAN, default=False), "errors": openapi.Schema( type=openapi.TYPE_ARRAY, items=openapi.Schema( type=openapi.TYPE_OBJECT, properties={ "code": openapi.Schema( type=openapi.TYPE_STRING, default="not_found" ), "message": openapi.Schema( type=openapi.TYPE_STRING, default="Ресурс не найден" ), }, ), ), }, ), ) VALIDATION_ERROR = openapi.Response( description="Ошибка валидации", schema=openapi.Schema( type=openapi.TYPE_OBJECT, properties={ "success": openapi.Schema(type=openapi.TYPE_BOOLEAN, default=False), "errors": openapi.Schema( type=openapi.TYPE_ARRAY, items=openapi.Schema( type=openapi.TYPE_OBJECT, properties={ "code": openapi.Schema(type=openapi.TYPE_STRING), "message": openapi.Schema(type=openapi.TYPE_STRING), "details": openapi.Schema(type=openapi.TYPE_OBJECT), }, ), ), }, ), ) RATE_LIMITED = openapi.Response( description="Превышен лимит запросов", schema=openapi.Schema( type=openapi.TYPE_OBJECT, properties={ "detail": openapi.Schema( type=openapi.TYPE_STRING, default="Превышен лимит запросов. Повторите позже.", ), }, ), ) SERVER_ERROR = openapi.Response( description="Внутренняя ошибка сервера", schema=openapi.Schema( type=openapi.TYPE_OBJECT, properties={ "success": openapi.Schema(type=openapi.TYPE_BOOLEAN, default=False), "errors": openapi.Schema( type=openapi.TYPE_ARRAY, items=openapi.Schema( type=openapi.TYPE_OBJECT, properties={ "code": openapi.Schema( type=openapi.TYPE_STRING, default="internal_error" ), "message": openapi.Schema( type=openapi.TYPE_STRING, default="Внутренняя ошибка сервера", ), }, ), ), }, ), ) SERVICE_UNAVAILABLE = openapi.Response( description="Сервис временно недоступен", schema=openapi.Schema( type=openapi.TYPE_OBJECT, properties={ "success": openapi.Schema(type=openapi.TYPE_BOOLEAN, default=False), "errors": openapi.Schema( type=openapi.TYPE_ARRAY, items=openapi.Schema( type=openapi.TYPE_OBJECT, properties={ "code": openapi.Schema( type=openapi.TYPE_STRING, default="service_unavailable", ), "message": openapi.Schema( type=openapi.TYPE_STRING, default="Сервис временно недоступен", ), }, ), ), }, ), ) class ErrorResponses: """Переиспользуемые наборы ошибок для OpenAPI responses.""" PUBLIC = { 429: CommonResponses.RATE_LIMITED, 500: CommonResponses.SERVER_ERROR, } AUTHENTICATED = { 401: CommonResponses.UNAUTHORIZED, **PUBLIC, } AUTHENTICATED_VALIDATION = { 400: CommonResponses.BAD_REQUEST, **AUTHENTICATED, } AUTHENTICATED_NOT_FOUND = { 404: CommonResponses.NOT_FOUND, **AUTHENTICATED, } AUTHENTICATED_VALIDATION_NOT_FOUND = { 400: CommonResponses.BAD_REQUEST, **AUTHENTICATED_NOT_FOUND, } ADMIN = { 401: CommonResponses.UNAUTHORIZED, 403: CommonResponses.FORBIDDEN, **PUBLIC, } ADMIN_NOT_FOUND = { 404: CommonResponses.NOT_FOUND, **ADMIN, } # Параметры запроса class CommonParameters: """ Общие параметры для документации API. Пример использования: @api_docs( summary="Список ресурсов", manual_parameters=[ CommonParameters.PAGE, CommonParameters.PAGE_SIZE, CommonParameters.SEARCH, ], ) def get(self, request): ... """ PAGE = openapi.Parameter( name="page", in_=openapi.IN_QUERY, type=openapi.TYPE_INTEGER, description="Номер страницы", default=1, ) PAGE_SIZE = openapi.Parameter( name="page_size", in_=openapi.IN_QUERY, type=openapi.TYPE_INTEGER, description="Количество элементов на странице", default=20, ) SEARCH = openapi.Parameter( name="search", in_=openapi.IN_QUERY, type=openapi.TYPE_STRING, description="Поисковый запрос", ) ORDERING = openapi.Parameter( name="ordering", in_=openapi.IN_QUERY, type=openapi.TYPE_STRING, description="Поле сортировки (префикс '-' для убывания)", ) ID = openapi.Parameter( name="id", in_=openapi.IN_PATH, type=openapi.TYPE_INTEGER, description="ID ресурса", required=True, ) def paginated_response(serializer_class: type) -> openapi.Response: """ Создаёт схему пагинированного ответа. Пример использования: @api_docs( summary="Список пользователей", responses={200: paginated_response(UserSerializer)}, ) def get(self, request): ... """ return openapi.Response( description="Пагинированный список", schema=openapi.Schema( type=openapi.TYPE_OBJECT, properties={ "success": openapi.Schema(type=openapi.TYPE_BOOLEAN, default=True), "data": openapi.Schema( type=openapi.TYPE_ARRAY, items=openapi.Schema(type=openapi.TYPE_OBJECT), ), "meta": openapi.Schema( type=openapi.TYPE_OBJECT, properties={ "pagination": openapi.Schema( type=openapi.TYPE_OBJECT, properties={ "count": openapi.Schema(type=openapi.TYPE_INTEGER), "page": openapi.Schema(type=openapi.TYPE_INTEGER), "page_size": openapi.Schema(type=openapi.TYPE_INTEGER), "total_pages": openapi.Schema( type=openapi.TYPE_INTEGER ), }, ), }, ), }, ), )