127 lines
4.1 KiB
Python
127 lines
4.1 KiB
Python
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
from drf_spectacular.extensions import OpenApiAuthenticationExtension
|
||
from rest_framework import serializers
|
||
|
||
|
||
class UnstructuredResponseSerializer(serializers.Serializer):
|
||
"""Явный placeholder для endpoint с динамическим JSON-ответом."""
|
||
|
||
pass
|
||
|
||
|
||
class SessionAwareJWTAuthenticationScheme(OpenApiAuthenticationExtension):
|
||
"""Описывает кастомный JWT backend как стандартный Bearer auth."""
|
||
|
||
target_class = "api.auth.authentication.SessionAwareJWTAuthentication"
|
||
name = "jwtAuth"
|
||
|
||
def get_security_definition(self, auto_schema: Any) -> dict[str, str]:
|
||
return {
|
||
"type": "http",
|
||
"scheme": "bearer",
|
||
"bearerFormat": "JWT",
|
||
}
|
||
|
||
|
||
TAG_METADATA = {
|
||
"Auth": "Аутентификация, JWT-токены и верификация аккаунта.",
|
||
"Communications": "Чаты, сообщения и опросы корпоративного мессенджера.",
|
||
"Documents": "Документы, папки, теги и связанные действия.",
|
||
"Employees": "Сотрудники, отделы, роли, должности, группы и навыки.",
|
||
"Feed": "Лента новостей и публикации.",
|
||
"Notifications": "Уведомления, счетчики, настройки каналов и web push.",
|
||
"Procurement": "Закупки, оборудование, поставщики и связанные процессы.",
|
||
"Requests": "Заявки сотрудников, комментарии и workflow согласования.",
|
||
"Schedule": "Календари, события, правила повторения и вхождения.",
|
||
"Search": "Поиск по системе и агрегированные поисковые выдачи.",
|
||
}
|
||
|
||
|
||
TAG_BY_PREFIX = {
|
||
"auth": "Auth",
|
||
"communications": "Communications",
|
||
"departments": "Employees",
|
||
"department-roles": "Employees",
|
||
"document-tags": "Documents",
|
||
"documents": "Documents",
|
||
"employee-actions": "Employees",
|
||
"employees": "Employees",
|
||
"folders": "Documents",
|
||
"groups": "Employees",
|
||
"notifications": "Notifications",
|
||
"positions": "Employees",
|
||
"posts": "Feed",
|
||
"procurement": "Procurement",
|
||
"requests": "Requests",
|
||
"schedule": "Schedule",
|
||
"search": "Search",
|
||
"skills": "Employees",
|
||
}
|
||
|
||
|
||
def exclude_duplicate_auth_routes(endpoints: list[tuple]) -> list[tuple]:
|
||
"""Оставить в OpenAPI канонический `/api/auth/` без v1-дубликата.
|
||
|
||
`/api/v1/auth/` пока остаётся runtime-алиасом для обратной совместимости.
|
||
"""
|
||
return [
|
||
endpoint
|
||
for endpoint in endpoints
|
||
if not str(endpoint[0]).startswith("/api/v1/auth/")
|
||
]
|
||
|
||
|
||
def _tag_for_path(path: str) -> str:
|
||
segments = [segment for segment in path.strip("/").split("/") if segment]
|
||
|
||
if segments[:2] == ["api", "v1"]:
|
||
segments = segments[2:]
|
||
elif segments[:1] == ["api"]:
|
||
segments = segments[1:]
|
||
|
||
if not segments:
|
||
return "API"
|
||
|
||
return TAG_BY_PREFIX.get(
|
||
segments[0],
|
||
segments[0].replace("-", " ").title(),
|
||
)
|
||
|
||
|
||
def assign_operation_tags(
|
||
result: dict[str, Any], generator: Any, request: Any, public: bool
|
||
) -> dict[str, Any]:
|
||
used_tags: set[str] = set()
|
||
valid_methods = {
|
||
"get",
|
||
"post",
|
||
"put",
|
||
"patch",
|
||
"delete",
|
||
"options",
|
||
"head",
|
||
"trace",
|
||
}
|
||
|
||
for path, path_item in result.get("paths", {}).items():
|
||
fallback_tag = _tag_for_path(path)
|
||
|
||
for method, operation in path_item.items():
|
||
if method not in valid_methods:
|
||
continue
|
||
|
||
tags = [fallback_tag]
|
||
operation["tags"] = tags
|
||
used_tags.update(tags)
|
||
|
||
result["tags"] = [
|
||
{"name": tag, "description": TAG_METADATA[tag]}
|
||
for tag in TAG_METADATA
|
||
if tag in used_tags
|
||
] + [{"name": tag} for tag in sorted(used_tags) if tag not in TAG_METADATA]
|
||
|
||
return result
|