387 lines
23 KiB
Markdown
387 lines
23 KiB
Markdown
# Contracts: Async Backend Refactoring
|
||
|
||
**Feature**: `032-translate-requests-httpx`
|
||
**Date**: 2026-06-04
|
||
|
||
## @{ AsyncAPIClient [C:4] [TYPE Class]
|
||
@BRIEF Асинхронный HTTP-клиент для Superset API — использует `httpx.AsyncClient` с connection pooling, CSRF-аутентификацией и глобальным per‑env семафором.
|
||
@PRE env_config — валидный EnvironmentConfig с url, username, password.
|
||
@POST Клиент аутентифицирован, connection pool готов. Все HTTP-методы возвращают response.
|
||
@SIDE_EFFECT HTTP-вызовы к Superset API; модификация CSRF-токена; захват/освобождение глобального семафора.
|
||
@RELATION DEPENDS_ON -> [EXT:httpx:AsyncClient]
|
||
@RELATION DEPENDS_ON -> [SupersetClientRegistry]
|
||
@RATIONALE httpx выбран вместо aiohttp — уже в зависимостях (0.28.1), лучший API для async HTTP.
|
||
@REJECTED requests.Session (sync) — блокирует event loop. aiohttp — требует новой зависимости.
|
||
|
||
### Client Lifecycle
|
||
|
||
**Production runtime**: Клиенты долгоживущие per‑environment через `SupersetClientRegistry` (синглтон). Создаются при первом обращении к `env_id`, закрываются на FastAPI shutdown. Request‑scoped клиенты запрещены в production.
|
||
|
||
**Tests**: разрешён `async with AsyncAPIClient(env) as client:` для изоляции тестов.
|
||
|
||
### Error Mapping
|
||
|
||
| Условие | Исключение | HTTP Status |
|
||
|---------|------------|-------------|
|
||
| Все слоты семафора заняты, таймаут | `SupersetPoolTimeout` | 503 Service Unavailable |
|
||
| Ошибка аутентификации (401 после retry) | `AuthenticationError` | 502 Bad Gateway |
|
||
| Таймаут upstream (httpx.Timeout) | `SupersetTimeout` | 504 Gateway Timeout |
|
||
| Сетевая ошибка (ConnectionError) | `NetworkError` | 502 Bad Gateway |
|
||
| Rate-limit от Superset (429) | `SupersetRateLimitError` | 502 Bad Gateway |
|
||
|
||
@RATIONALE 502 для rate‑limit выбран, чтобы не раскрывать upstream 429 как отдельный код ошибки — все ошибки Superset маппятся в 502/503/504 для унификации клиентского контракта.
|
||
|
||
### Auth Refresh Safety
|
||
|
||
Обновление CSRF/auth‑токенов защищено `asyncio.Lock` per‑environment через `SupersetClientRegistry`. Retry после 401 разрешён только для идемпотентных методов (GET). Для неидемпотентных (POST import, PUT) — ошибка всплывает вызывающему без автоматического retry.
|
||
|
||
@DATA_CONTRACT Input: (method, endpoint, params, data, headers, files) → Output: Response dict with status_code, data, headers.
|
||
|
||
### Контракты методов
|
||
|
||
```
|
||
async def request(method, endpoint, params, data, headers, files, timeout) -> dict
|
||
@PRE client аутентифицирован, semaphore доступен.
|
||
@POST HTTP-ответ получен, semaphore освобождён.
|
||
@SIDE_EFFECT HTTP-вызов к Superset; обновление CSRF-токена при 401.
|
||
|
||
async def authenticate() -> None
|
||
@POST CSRF-токен получен, сессия валидна.
|
||
|
||
async def upload_file(endpoint, file_path, ...) -> dict
|
||
@PRE файл существует на диске.
|
||
@POST файл загружен, ответ получен.
|
||
|
||
async def fetch_paginated_data(endpoint, query) -> list[dict]
|
||
@POST все страницы загружены, результат агрегирован.
|
||
|
||
async def fetch_paginated_count(endpoint, query) -> int
|
||
@POST возвращает общее количество записей.
|
||
```
|
||
## @} AsyncAPIClient
|
||
|
||
---
|
||
|
||
## @{ SupersetClientAuth [C:3] [TYPE Module]
|
||
@BRIEF CSRF-аутентификация для Superset API — извлечено из `network.py` для соблюдения INV_7 (<400 строк).
|
||
@RELATION DEPENDS_ON -> [AsyncAPIClient]
|
||
@RATIONALE Декомпозиция `network.py` (542 строки) на три модуля: `async_client.py`, `auth.py`, `paginated.py`. Аутентификация — отдельная ответственность.
|
||
@REJECTED Оставить в `network.py` — нарушает INV_7, модуль > 400 строк.
|
||
|
||
```
|
||
async def fetch_csrf_token(client: AsyncAPIClient) -> str
|
||
@BRIEF GET /api/v1/security/csrf_token/, возвращает CSRF-токен.
|
||
|
||
async def login(client: AsyncAPIClient, username: str, password: str) -> None
|
||
@BRIEF POST /api/v1/security/login, устанавливает сессионную куку.
|
||
@POST клиент аутентифицирован, CSRF-токен обновлён.
|
||
```
|
||
## @} SupersetClientAuth
|
||
|
||
---
|
||
|
||
## @{ SupersetClientPaginated [C:3] [TYPE Module]
|
||
@BRIEF Пагинированная загрузка данных из Superset API — извлечено из `network.py`.
|
||
@RELATION DEPENDS_ON -> [AsyncAPIClient]
|
||
@RATIONALE Пагинация — отдельная ответственность, вынесена для INV_7.
|
||
|
||
```
|
||
async def fetch_paginated(client: AsyncAPIClient, endpoint: str, query: dict) -> list[dict]
|
||
@BRIEF Итеративная загрузка всех страниц пагинированного ответа.
|
||
@POST все страницы загружены и агрегированы в единый список.
|
||
|
||
async def fetch_count(client: AsyncAPIClient, endpoint: str, query: dict) -> int
|
||
@BRIEF Загрузка количества записей (page_size=0, только count).
|
||
@POST возвращает total_count из ответа Superset.
|
||
```
|
||
## @} SupersetClientPaginated
|
||
|
||
---
|
||
|
||
## @{ SupersetClient [C:4] [TYPE Class]
|
||
@BRIEF Единый асинхронный клиент для Superset API. Объединяет sync `SupersetClient` и `AsyncSupersetClient` в полностью асинхронный фасад. Все 13 миксинов — async.
|
||
@PRE env_config — валидный EnvironmentConfig.
|
||
@POST Клиент готов к вызовам Superset API.
|
||
@SIDE_EFFECT Все методы выполняют асинхронные HTTP-вызовы к Superset.
|
||
@RELATION DEPENDS_ON -> [AsyncAPIClient]
|
||
@RELATION DEPENDS_ON -> [SupersetClientAuth]
|
||
@RELATION DEPENDS_ON -> [SupersetClientPaginated]
|
||
@RATIONALE Big-bang миграция: sync-код удаляется, dual-stack не поддерживается. AsyncSupersetClient поглощается.
|
||
@REJECTED Dual-stack (sync + async) — отвергнут на этапе clarification. Wrapper-only (asyncio.to_thread) — отвергнут, httpx.AsyncClient эффективнее.
|
||
|
||
### Сигнатуры ключевых методов (все async def)
|
||
|
||
```
|
||
async def get_dashboards(query=None) -> tuple[int, list[dict]]
|
||
async def get_dashboard(dashboard_id: str) -> dict
|
||
async def export_dashboard(dashboard_id: str) -> tuple[bytes, str]
|
||
async def import_dashboard(file_path: str) -> dict
|
||
async def get_datasets(query=None) -> tuple[int, list[dict]]
|
||
async def get_dataset_detail(dataset_id: int) -> dict
|
||
async def get_database(database_id: int) -> dict
|
||
async def get_databases(query=None) -> tuple[int, list[dict]]
|
||
async def get_chart_data(query_context: dict) -> dict
|
||
async def execute_sql_lab(payload: dict) -> dict
|
||
async def get_query_status(query_id: str) -> dict
|
||
async def get_query_results(query_id: str) -> dict
|
||
async def cache_screenshot(dashboard_id: str) -> dict
|
||
async def get_thumbnail(dashboard_id: str) -> bytes
|
||
```
|
||
|
||
### Контракты на 13 миксинов
|
||
|
||
Все миксины (`_base.py`, `_datasets.py`, `_dashboards.py`, `_charts.py`, `_sql_lab.py`, `_database.py`, `_import_export.py`, `_thumbnail.py`, `_logs.py`, `_css.py`, `_annotation.py`, `_filter.py`, `_query_context.py`) обновляются:
|
||
- Все методы: `def` → `async def`
|
||
- @RELATION DEPENDS_ON -> [SupersetClient] (было [APIClient] или [SupersetClient])
|
||
- @SIDE_EFFECT обновлён: «асинхронные HTTP-вызовы к Superset API»
|
||
|
||
Контракт `AsyncSupersetClient` → Tombstone:
|
||
- @DEPRECATED: 2026-06-04
|
||
- @REPLACED_BY: [SupersetClient]
|
||
## @} SupersetClient
|
||
|
||
---
|
||
|
||
## @{ SupersetClientRegistry [C:5] [TYPE Class]
|
||
@BRIEF Синглтон-реестр долгоживущих асинхронных клиентов по env_id. Хранит shared httpx.AsyncClient + shared asyncio.Semaphore + asyncio.Lock для auth refresh.
|
||
@PRE FastAPI event loop запущен. Environment configs загружены.
|
||
@POST Клиенты доступны по env_id, закрыты на shutdown.
|
||
@SIDE_EFFECT Создание/закрытие httpx.AsyncClient; управление семафорами и auth‑локами.
|
||
@INVARIANT Один клиент и один семафор на env_id за всё время жизни приложения.
|
||
@INVARIANT Все клиенты закрыты при shutdown.
|
||
@RELATION DEPENDS_ON -> [AsyncAPIClient]
|
||
@RATIONALE Глобальный per‑env семафор невозможен без синглтона. Request‑scoped клиенты теряют connection pooling. Shutdown без реестра оставляет висящие соединения.
|
||
@REJECTED Per‑request клиенты — теряют connection pooling, семафор не работает. Inline создание в каждом use‑case — утечки соединений.
|
||
|
||
```
|
||
async def get_client(env_id: str) -> AsyncAPIClient
|
||
@BRIEF Получить или создать клиент для окружения. Создание ленивое при первом обращении.
|
||
@POST Клиент готов, семафор активен.
|
||
|
||
async def shutdown() -> None
|
||
@BRIEF Закрыть все httpx.AsyncClient, освободить семафоры, очистить auth‑локи.
|
||
@POST Все сетевые ресурсы освобождены.
|
||
```
|
||
## @} SupersetClientRegistry
|
||
|
||
---
|
||
|
||
## @{ GracefulShutdown [C:4] [TYPE Block]
|
||
@BRIEF Контракт graceful shutdown: порядок закрытия ресурсов при остановке FastAPI.
|
||
@PRE lifespan shutdown event получен.
|
||
@POST Все ресурсы освобождены, незавершённые задачи помечены CANCELLED.
|
||
@SIDE_EFFECT Прекращение приёма задач; отмена активных asyncio.Task; закрытие клиентов через SupersetClientRegistry; слив/закрытие EventBus.
|
||
@INVARIANT Shutdown завершается в течение graceful_timeout (по умолчанию 30 секунд).
|
||
|
||
```
|
||
async def on_shutdown():
|
||
1. stop_accepting_new_tasks()
|
||
2. cancel_active_tasks(graceful_timeout=30)
|
||
3. await SupersetClientRegistry.shutdown()
|
||
4. await EventBus.drain()
|
||
5. mark_unfinished_tasks(CANCELLED)
|
||
```
|
||
## @} GracefulShutdown
|
||
|
||
---
|
||
|
||
## @{ SupersetSqlLabExecutor [C:4] [TYPE Class]
|
||
@BRIEF Асинхронный исполнитель SQL-запросов через Superset SQL Lab API. Поллинг с `asyncio.sleep`.
|
||
@PRE env_id — валидное окружение Superset. SQL — валидный и санитизированный.
|
||
@POST Результат выполнения SQL получен и распарсен.
|
||
@SIDE_EFFECT Асинхронные HTTP-вызовы к Superset SQL Lab API; поллинг статуса.
|
||
@RELATION DEPENDS_ON -> [SupersetClient]
|
||
@RATIONALE Поллинг с `asyncio.sleep` вместо `time.sleep` — не блокирует event loop между попытками.
|
||
@REJECTED time.sleep в цикле поллинга — блокирует event loop до 120 секунд (60 попыток × 2s).
|
||
|
||
```
|
||
async def execute_sql(sql: str, database_id: int, run_async: bool) -> dict
|
||
@POST query_id получен.
|
||
|
||
async def poll_execution_status(query_id: str, max_polls: int, poll_interval: float) -> dict
|
||
@BRIEF Поллинг с asyncio.sleep. Возвращает {"status": "success"|"failed"|"timeout"}.
|
||
@POST результат выполнения SQL получен или таймаут.
|
||
|
||
async def execute_and_poll(sql: str, database_id: int, ...) -> dict
|
||
@BRIEF execute_sql + poll_execution_status в одном вызове.
|
||
```
|
||
## @} SupersetSqlLabExecutor
|
||
|
||
---
|
||
|
||
## @{ LLMHttpClient [C:4] [TYPE Module]
|
||
@BRIEF Унифицированный асинхронный HTTP-клиент для OpenAI-совместимых LLM API. Замена `_llm_http.py` и `preview_llm_client.py`.
|
||
@PRE api_key, base_url, model — валидны.
|
||
@POST LLM-ответ получен и распарсен.
|
||
@SIDE_EFFECT HTTP POST к LLM API; rate-limit backoff с asyncio.sleep.
|
||
@RELATION DEPENDS_ON -> [EXT:httpx:AsyncClient]
|
||
@RATIONALE Объединение двух дублирующихся LLM-клиентов в один модуль. httpx.AsyncClient обеспечивает неблокирующий HTTP. asyncio.sleep вместо time.sleep для rate-limit пауз.
|
||
@REJECTED requests + time.sleep — блокирует event loop. AsyncOpenAI SDK — не поддерживает кастомные base_url для OpenRouter/LiteLLM/Kilo.
|
||
|
||
```
|
||
async def call_openai_compatible(base_url, api_key, model, prompt, max_tokens, provider_type, disable_reasoning) -> tuple[str, str|None]
|
||
@BRIEF Отправить prompt и получить (content, finish_reason).
|
||
@PRE api_key дешифрован, base_url не пустой.
|
||
@POST (response_text, finish_reason) или ValueError при ошибке.
|
||
@SIDE_EFFECT HTTP POST + retry при 429 (rate-limit) с exponential backoff через asyncio.sleep.
|
||
```
|
||
## @} LLMHttpClient
|
||
|
||
---
|
||
|
||
## @{ TaskManager [C:5] [TYPE Class]
|
||
@BRIEF Менеджер фоновых задач — полный переход с ThreadPoolExecutor на asyncio.create_task.
|
||
@PRE FastAPI event loop запущен. Плагины реализуют async def execute().
|
||
@POST Задачи запущены, выполняются конкурентно в основном event loop.
|
||
@SIDE_EFFECT Запуск/отмена asyncio.Task; обновление статуса в БД; публикация событий в EventBus.
|
||
@INVARIANT Задачи не блокируют event loop дольше, чем на 100ms между await-точками.
|
||
@INVARIANT Отмена задачи cascade-освобождает все ресурсы (HTTP-соединения, семафоры, файловые дескрипторы).
|
||
@RELATION DEPENDS_ON -> [EventBus]
|
||
@RELATION DEPENDS_ON -> [EXT:asyncio:Task]
|
||
@DATA_CONTRACT Input: TaskParams(plugin_name, params, user_id) → Output: TaskResult(status, result, error)
|
||
@RATIONALE asyncio.create_task вместо ThreadPoolExecutor — задачи разделяют event loop, не требуют тредовой синхронизации, поддерживают нативную отмену через task.cancel().
|
||
@REJECTED ThreadPoolExecutor — создаёт изоляцию между задачами, усложняет отмену и WebSocket-логирование. Per-task event loop (asyncio.run) — отвергнут, изоляция мешает общей отмене.
|
||
|
||
```
|
||
async def run_task(plugin_name: str, params: dict, user_id: str) -> str
|
||
@BRIEF Запустить задачу как asyncio.create_task. Вернуть task_id.
|
||
@POST задача зарегистрирована, статус PENDING → RUNNING.
|
||
|
||
async def cancel_task(task_id: str) -> None
|
||
@BRIEF Отменить задачу через task.cancel(). Обработать CancelledError.
|
||
@POST статус задачи CANCELLED, ресурсы освобождены.
|
||
|
||
async def get_task_status(task_id: str) -> dict
|
||
@POST {"status": "RUNNING"|"COMPLETED"|"FAILED"|"CANCELLED", ...}
|
||
```
|
||
## @} TaskManager
|
||
|
||
---
|
||
|
||
## @{ EventBus [C:4] [TYPE Class]
|
||
@BRIEF Асинхронная шина событий для WebSocket-логирования задач. asyncio.Queue(maxsize=10000, настраивается) вместо queue.Queue. Одна очередь на подписчика. При переполнении — отбрасывание нового события с логированием предупреждения.
|
||
@PRE TaskManager и WebSocket-ручки инициализированы.
|
||
@POST События доставлены через fan‑out на все subscriber queues. При task.cancel() — финальное событие cancelled, дальнейшая публикация для task_id прекращается.
|
||
@SIDE_EFFECT Публикация событий; WebSocket-отправка логов.
|
||
@RELATION DEPENDS_ON -> [EXT:asyncio:Queue]
|
||
@RATIONALE asyncio.Queue интегрируется с event loop'ом. maxsize предотвращает неограниченный рост памяти. Отбрасывание вместо блокировки защищает задачу от замедления. Fan‑out (одна очередь на подписчика) гарантирует broadcast.
|
||
@REJECTED queue.Queue + call_soon_threadsafe — избыточно. Одна общая очередь — не гарантирует broadcast.
|
||
|
||
```
|
||
async def publish(event: dict) -> None
|
||
@POST событие добавлено в очередь.
|
||
|
||
async def subscribe() -> AsyncIterator[dict]
|
||
@BRIEF Асинхронный генератор — читает события из очереди.
|
||
@POST yield каждого события. При отмене (WebSocket disconnect) — выход из генератора.
|
||
```
|
||
## @} EventBus
|
||
|
||
---
|
||
|
||
## @{ TaskLifecycle [C:4] [TYPE Class]
|
||
@BRIEF Асинхронный жизненный цикл задачи: подготовка, выполнение плагина, финализация.
|
||
@PRE TaskManager запущен, плагин загружен.
|
||
@POST Задача выполнена, результат записан в БД, событие опубликовано.
|
||
@SIDE_EFFECT Запуск плагина; обновление статуса задачи; публикация событий.
|
||
@RELATION DEPENDS_ON -> [TaskManager]
|
||
@RELATION DEPENDS_ON -> [EventBus]
|
||
@RATIONALE Переход на async context manager вместо ThreadPoolExecutor.submit().
|
||
|
||
```
|
||
async def execute_task(task_id: str, plugin_name: str, params: dict) -> None
|
||
@BRIEF Полный цикл: PENDING → RUNNING → execute → COMPLETED/FAILED.
|
||
@SIDE_EFFECT Сохраняет статус в БД; публикует start/progress/complete события.
|
||
```
|
||
## @} TaskLifecycle
|
||
|
||
---
|
||
|
||
## @{ SMTPProvider [C:3] [TYPE Class]
|
||
@BRIEF Асинхронный SMTP-провайдер уведомлений. aiosmtplib вместо smtplib.
|
||
@PRE host, port, username, password — валидны.
|
||
@POST Email отправлен или ошибка залогирована.
|
||
@SIDE_EFFECT Подключение к SMTP-серверу, отправка письма.
|
||
@RELATION DEPENDS_ON -> [EXT:aiosmtplib:SMTP]
|
||
@RATIONALE aiosmtplib — зрелая async-замена smtplib. Не блокирует event loop при TCP-подключении и SMTP-командах.
|
||
@REJECTED smtplib.SMTP — блокирует event loop. asyncio.to_thread для SMTP — работает, но aiosmtplib даёт нативную интеграцию с event loop.
|
||
|
||
```
|
||
async def send(recipient: str, subject: str, body: str) -> None
|
||
@POST письмо отправлено или ошибка залогирована (не пробрасывается).
|
||
```
|
||
## @} SMTPProvider
|
||
|
||
---
|
||
|
||
## @{ TelegramProvider [C:3] [TYPE Class]
|
||
@BRIEF Асинхронный Telegram-провайдер уведомлений. httpx.AsyncClient вместо requests.
|
||
@PRE bot_token, chat_id — валидны.
|
||
@POST Сообщение отправлено.
|
||
@SIDE_EFFECT HTTP POST к Telegram Bot API.
|
||
@RELATION DEPENDS_ON -> [EXT:httpx:AsyncClient]
|
||
@RATIONALE httpx.AsyncClient переиспользуется из пула соединений — эффективнее requests.
|
||
@REJECTED requests.post — блокирует event loop.
|
||
|
||
```
|
||
async def send(recipient: str, subject: str, body: str) -> None
|
||
@POST сообщение отправлено через Telegram Bot API.
|
||
```
|
||
## @} TelegramProvider
|
||
|
||
---
|
||
|
||
## @{ NotificationService [C:3] [TYPE Class]
|
||
@BRIEF Сервис отправки уведомлений — вызывает провайдеры асинхронно.
|
||
@PRE NotificationConfig загружен, провайдеры инициализированы.
|
||
@POST Уведомление отправлено через все настроенные каналы.
|
||
@SIDE_EFFECT Вызовы провайдеров (SMTP, Telegram, Slack) — каждый выполняется независимо.
|
||
@RELATION DEPENDS_ON -> [SMTPProvider]
|
||
@RELATION DEPENDS_ON -> [TelegramProvider]
|
||
@RELATION DEPENDS_ON -> [SlackProvider]
|
||
|
||
```
|
||
async def dispatch_report(report: Report, config: NotificationConfig) -> None
|
||
@BRIEF Отправить отчёт через все включённые каналы параллельно (asyncio.gather).
|
||
@POST все провайдеры вызваны, ошибки залогированы индивидуально.
|
||
```
|
||
## @} NotificationService
|
||
|
||
---
|
||
|
||
## @{ FileIO [C:3] [TYPE Module]
|
||
@BRIEF Гибридные асинхронные файловые операции: aiofiles для больших файлов, asyncio.to_thread для мелких.
|
||
@RATIONALE aiofiles для потокового I/O больших файлов, asyncio.to_thread для операций, не поддерживаемых aiofiles (shutil.rmtree, рекурсивные операции).
|
||
@REJECTED Только asyncio.to_thread — для 50MB ZIP блокирует тред на секунды. Только aiofiles — не поддерживает shutil-операции.
|
||
|
||
```
|
||
async def read_dashboard_from_disk(file_path: str) -> bytes
|
||
@BRIEF Чтение дашборда с диска. >1MB → aiofiles, иначе asyncio.to_thread.
|
||
|
||
async def save_and_unpack_dashboard(zip_content: bytes, target_dir: str) -> None
|
||
@BRIEF Сохранение и распаковка ZIP. Запись через asyncio.to_thread.
|
||
|
||
async def cleanup_directory(path: str) -> None
|
||
@BRIEF Удаление директории через asyncio.to_thread(shutil.rmtree).
|
||
```
|
||
## @} FileIO
|
||
|
||
---
|
||
|
||
## Tombstoned Contracts
|
||
|
||
### @{ APIClient [Tombstone] [TYPE Class]
|
||
@DEPRECATED 2026-06-04 — заменён на AsyncAPIClient. Синхронный requests.Session больше не используется.
|
||
@REPLACED_BY -> [AsyncAPIClient]
|
||
## @} APIClient
|
||
|
||
### @{ AsyncSupersetClient [Tombstone] [TYPE Class]
|
||
@DEPRECATED 2026-06-04 — поглощён единым асинхронным SupersetClient.
|
||
@REPLACED_BY -> [SupersetClient]
|
||
## @} AsyncSupersetClient
|
||
|
||
### @{ LLMHttpClient_Legacy [Tombstone] [TYPE Module]
|
||
@DEPRECATED 2026-06-04 — `_llm_http.py` и `preview_llm_client.py` заменены унифицированным `LLMHttpClient`.
|
||
@REPLACED_BY -> [LLMHttpClient]
|
||
## @} LLMHttpClient_Legacy
|