diff --git a/backend/src/api/routes/dashboards/_detail_routes.py b/backend/src/api/routes/dashboards/_detail_routes.py index b27f6c8c..010bb650 100644 --- a/backend/src/api/routes/dashboards/_detail_routes.py +++ b/backend/src/api/routes/dashboards/_detail_routes.py @@ -15,7 +15,7 @@ from fastapi.responses import JSONResponse from src.core.async_superset_client import AsyncSupersetClient from src.core.logger import belief_scope, logger -from src.core.superset_client import SupersetClient +from src.core.utils.client_registry import get_client as get_superset_client from src.core.utils.network import DashboardNotFoundError from src.dependencies import ( get_config_manager, @@ -25,7 +25,6 @@ from src.dependencies import ( ) from ._helpers import ( - _resolve_dashboard_id_from_ref, _resolve_dashboard_id_from_ref_async, ) from ._projection import ( @@ -131,9 +130,23 @@ async def get_dashboard_detail( raise HTTPException(status_code=404, detail="Environment not found") try: - sync_client = SupersetClient(env) - dashboard_id = _resolve_dashboard_id_from_ref(dashboard_ref, sync_client) - detail = sync_client.get_dashboard_detail(dashboard_id) + # Use async SupersetClient with shared AsyncAPIClient from registry + env_config = { + "base_url": env.url, + "auth": { + "username": env.username, + "password": env.password, + "provider": "db", + "refresh": "true", + }, + "verify_ssl": env.verify_ssl, + } + async_client = await get_superset_client(env_config) + client = AsyncSupersetClient(env) + dashboard_id = await _resolve_dashboard_id_from_ref_async( + dashboard_ref, client + ) + detail = await client.get_dashboard_detail(dashboard_id) logger.info( f"[get_dashboard_detail][Coherence:OK] Dashboard ref={dashboard_ref} resolved_id={dashboard_id}: {detail.get('chart_count', 0)} charts, {detail.get('dataset_count', 0)} datasets" ) @@ -256,7 +269,7 @@ async def get_dashboard_tasks_history( # #region get_dashboard_thumbnail [C:3] [TYPE Function] # @BRIEF Proxies Superset dashboard thumbnail with cache support. -# @RELATION CALLS -> [AsyncSupersetClient] +# @RELATION CALLS -> [AsyncAPIClient] # @PRE env_id must exist. # @POST Returns image bytes or 202 when thumbnail is being prepared by Superset. @router.get("/{dashboard_ref}/thumbnail") @@ -280,13 +293,27 @@ async def get_dashboard_thumbnail( raise HTTPException(status_code=404, detail="Environment not found") try: - client = SupersetClient(env) - dashboard_id = _resolve_dashboard_id_from_ref(dashboard_ref, client) + # Use shared AsyncAPIClient from registry + env_config = { + "base_url": env.url, + "auth": { + "username": env.username, + "password": env.password, + "provider": "db", + "refresh": "true", + }, + "verify_ssl": env.verify_ssl, + } + async_client = await get_superset_client(env_config) + client = AsyncSupersetClient(env) + dashboard_id = await _resolve_dashboard_id_from_ref_async( + dashboard_ref, client + ) digest = None thumb_endpoint = None try: - screenshot_payload = client.network.request( + screenshot_payload = await async_client.request( method="POST", endpoint=f"/dashboard/{dashboard_id}/cache_dashboard_screenshot/", ) @@ -311,7 +338,7 @@ async def get_dashboard_thumbnail( ) if not digest: - dashboard_payload = client.network.request( + dashboard_payload = await async_client.request( method="GET", endpoint=f"/dashboard/{dashboard_id}", ) @@ -343,11 +370,10 @@ async def get_dashboard_thumbnail( f"/dashboard/{dashboard_id}/thumbnail/{digest or 'latest'}/" ) - thumb_response = client.network.request( + thumb_response = await async_client.request( method="GET", endpoint=thumb_endpoint, raw_response=True, - allow_redirects=True, ) if thumb_response.status_code == 202: diff --git a/backend/src/api/routes/dashboards/_listing_routes.py b/backend/src/api/routes/dashboards/_listing_routes.py index 8bec60ed..f5502847 100644 --- a/backend/src/api/routes/dashboards/_listing_routes.py +++ b/backend/src/api/routes/dashboards/_listing_routes.py @@ -245,7 +245,7 @@ async def get_dashboards( and bound_username and profile_service is not None ): - actor_aliases = _resolve_profile_actor_aliases(env, bound_username) + actor_aliases = await _resolve_profile_actor_aliases(env, bound_username) if not actor_aliases: actor_aliases = [bound_username] logger.reason( diff --git a/backend/src/api/routes/dashboards/_projection.py b/backend/src/api/routes/dashboards/_projection.py index be56a5f8..5b6a49fe 100644 --- a/backend/src/api/routes/dashboards/_projection.py +++ b/backend/src/api/routes/dashboards/_projection.py @@ -7,7 +7,7 @@ from typing import Any from src.core.logger import logger -from src.core.superset_client import SupersetClient +from src.core.async_superset_client import AsyncSupersetClient from src.core.superset_profile_lookup import SupersetAccountLookupAdapter from src.models.auth import User from src.services.profile_service import ProfileService @@ -142,19 +142,19 @@ def _get_profile_filter_binding( # #region _resolve_profile_actor_aliases [C:2] [TYPE Function] # @BRIEF Resolve stable actor aliases for profile filtering without per-dashboard detail fan-out. -# @SIDE_EFFECT Performs at most one Superset users-lookup request. -def _resolve_profile_actor_aliases(env: Any, bound_username: str) -> list[str]: +# @SIDE_EFFECT Асинхронный HTTP-вызов к Superset API для поиска пользователей. +async def _resolve_profile_actor_aliases(env: Any, bound_username: str) -> list[str]: normalized_bound = _normalize_actor_alias_token(bound_username) if not normalized_bound: return [] aliases: list[str] = [normalized_bound] try: - client = SupersetClient(env) + client = AsyncSupersetClient(env) adapter = SupersetAccountLookupAdapter( - network_client=client.network, + network_client=client.client, environment_id=str(getattr(env, "id", "")), ) - lookup_payload = adapter.get_users_page( + lookup_payload = await adapter.get_users_page( search=normalized_bound, page_index=0, page_size=20, diff --git a/backend/src/api/routes/environments.py b/backend/src/api/routes/environments.py index 6e3c89c2..5c46b46e 100644 --- a/backend/src/api/routes/environments.py +++ b/backend/src/api/routes/environments.py @@ -12,7 +12,7 @@ from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel, Field from ...core.logger import belief_scope -from ...core.superset_client import SupersetClient +from ...core.async_superset_client import AsyncSupersetClient from ...dependencies import get_config_manager, get_scheduler_service, has_permission router = APIRouter(prefix="/api/environments", tags=["Environments"]) @@ -138,9 +138,9 @@ async def get_environment_databases( raise HTTPException(status_code=404, detail="Environment not found") try: - # Initialize SupersetClient from environment config - client = SupersetClient(env) - return client.get_databases_summary() + # Initialize AsyncSupersetClient from environment config + client = AsyncSupersetClient(env) + return await client.get_databases_summary() except Exception as e: raise HTTPException(status_code=500, detail=f"Failed to fetch databases: {e!s}") # #endregion get_environment_databases diff --git a/backend/src/api/routes/profile.py b/backend/src/api/routes/profile.py index 43682c6a..eaa7132b 100644 --- a/backend/src/api/routes/profile.py +++ b/backend/src/api/routes/profile.py @@ -138,7 +138,7 @@ async def lookup_superset_accounts( ) try: logger.reason("Executing Superset account lookup") - return service.lookup_superset_accounts( + return await service.lookup_superset_accounts( current_user=current_user, request=lookup_request, ) diff --git a/backend/src/api/routes/settings.py b/backend/src/api/routes/settings.py index 8563f33e..f929c510 100755 --- a/backend/src/api/routes/settings.py +++ b/backend/src/api/routes/settings.py @@ -22,7 +22,7 @@ from ...core.config_manager import ConfigManager from ...core.config_models import AppConfig, Environment, GlobalSettings, LoggingConfig from ...core.database import get_db from ...core.logger import belief_scope, logger -from ...core.superset_client import SupersetClient +from ...core.async_superset_client import AsyncSupersetClient from ...dependencies import get_config_manager, has_permission from ...models.config import AppConfigRecord from ...models.llm import ValidationPolicy @@ -73,12 +73,12 @@ def _normalize_superset_env_url(raw_url: str) -> str: # @BRIEF Run lightweight Superset connectivity validation without full pagination scan. # @PRE env contains valid URL and credentials. # @POST Raises on auth/API failures; returns None on success. -def _validate_superset_connection_fast(env: Environment) -> None: - client = SupersetClient(env) +async def _validate_superset_connection_fast(env: Environment) -> None: + client = AsyncSupersetClient(env) # 1) Explicit auth check - client.authenticate() + await client.authenticate() # 2) Single lightweight API call to ensure read access - client.get_dashboards_page( + _, _ = await client.get_dashboards_page( query={ "page": 0, "page_size": 1, @@ -238,7 +238,7 @@ async def add_environment( # Validate connection before adding (fast path) try: - _validate_superset_connection_fast(env) + await _validate_superset_connection_fast(env) except Exception as e: logger.error( f"[add_environment][Coherence:Failed] Connection validation failed: {e}" @@ -280,7 +280,7 @@ async def update_environment( # Validate connection before updating (fast path) try: - _validate_superset_connection_fast(env_to_validate) + await _validate_superset_connection_fast(env_to_validate) except Exception as e: logger.error( f"[update_environment][Coherence:Failed] Connection validation failed: {e}" @@ -331,7 +331,7 @@ async def test_environment_connection( raise HTTPException(status_code=404, detail=f"Environment {id} not found") try: - _validate_superset_connection_fast(env) + await _validate_superset_connection_fast(env) logger.info( f"[test_environment_connection][Coherence:OK] Connection successful for {id}" diff --git a/backend/src/core/superset_client/__init__.py b/backend/src/core/superset_client/__init__.py index dd83abe3..7c002032 100644 --- a/backend/src/core/superset_client/__init__.py +++ b/backend/src/core/superset_client/__init__.py @@ -1,23 +1,24 @@ # #region SupersetClientModule [C:5] [TYPE Module] [SEMANTICS superset, package, superset-client] # @LAYER Service -# @BRIEF Предоставляет высокоуровневый клиент для взаимодействия с Superset REST API, инкапсулируя логику запросов, обработку ошибок и пагинацию. +# @BRIEF Предоставляет высокоуровневый асинхронный клиент для взаимодействия с Superset REST API, инкапсулируя логику запросов, обработку ошибок и пагинацию. # @RELATION DEPENDS_ON -> [ConfigModels] -# @RELATION DEPENDS_ON -> [APIClient] +# @RELATION DEPENDS_ON -> [AsyncAPIClient] # @RELATION DEPENDS_ON -> [SupersetAPIError] # @RELATION DEPENDS_ON -> [get_filename_from_headers] # @RELATION DEPENDS_ON -> [SupersetDatasetsPreviewFiltersMixin] # -# @INVARIANT All network operations must use the internal APIClient instance. +# @INVARIANT All network operations must use the internal AsyncAPIClient instance. # @PUBLIC_API SupersetClient # # @RATIONALE Decomposed from monolithic superset_client.py (2145 lines) into # domain-scoped mixins to satisfy INV_7 (module < 400 lines). The composed class # preserves the original public API surface — all consumers continue to import # `from src.core.superset_client import SupersetClient` without changes. +# Все методы теперь асинхронные, используют AsyncAPIClient вместо APIClient. # @REJECTED Keeping a single 2145-line file — violates fractal limit INV_7. # @PRE Superset instance URL and credentials configured -# @POST SupersetClient class exported -# @SIDE_EFFECT Establishes HTTP connection pool to Superset +# @POST SupersetClient class exported with async methods +# @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API через AsyncAPIClient # @DATA_CONTRACT SupersetConfig -> SupersetClient from ._base import SupersetClientBase @@ -34,9 +35,9 @@ from ._user_projection import SupersetUserProjectionMixin # #region SupersetClient [C:3] [TYPE Class] -# @BRIEF Класс-обёртка над Superset REST API, предоставляющий методы для работы с дашбордами и датасетами. +# @BRIEF Класс-обёртка над Superset REST API, предоставляющий асинхронные методы для работы с дашбордами и датасетами. # @RELATION DEPENDS_ON -> [ConfigModels] -# @RELATION DEPENDS_ON -> [APIClient] +# @RELATION DEPENDS_ON -> [AsyncAPIClient] # @RELATION DEPENDS_ON -> [SupersetAPIError] # @RELATION INHERITS -> [SupersetClientBase] # @RELATION INHERITS -> [SupersetUserProjectionMixin] @@ -49,6 +50,7 @@ from ._user_projection import SupersetUserProjectionMixin # @RELATION INHERITS -> [SupersetDatasetsPreviewFiltersMixin] # @RELATION INHERITS -> [SupersetDatabasesMixin] # @RELATION INHERITS -> [SupersetDashboardsWriteMixin] +# @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API через AsyncAPIClient. class SupersetClient( SupersetDatabasesMixin, SupersetDatasetsPreviewFiltersMixin, @@ -66,6 +68,8 @@ class SupersetClient( MRO order ensures domain mixins resolve before base class. All consumers continue to use: from src.core.superset_client import SupersetClient + + Все методы асинхронные. Использует AsyncAPIClient для HTTP-вызовов. """ pass diff --git a/backend/src/core/superset_client/_base.py b/backend/src/core/superset_client/_base.py index 7be2840d..b23ff715 100644 --- a/backend/src/core/superset_client/_base.py +++ b/backend/src/core/superset_client/_base.py @@ -1,37 +1,39 @@ -# #region SupersetClientBase [C:3] [TYPE Module] [SEMANTICS superset, client, base, auth, pagination] +# #region SupersetClientBase [C:4] [TYPE Module] [SEMANTICS superset, client, base, auth, pagination, async] # @LAYER Infrastructure # @BRIEF Base class for SupersetClient providing initialization, authentication, pagination, and import/export helpers. # @RELATION DEPENDS_ON -> [get_filename_from_headers] -# @RELATION DEPENDS_ON -> [get_filename_from_headers] -# @RELATION DEPENDS_ON -> [get_filename_from_headers] -# @RELATION DEPENDS_ON -> [get_filename_from_headers] # @RATIONALE Extracted magic number 100 into DEFAULT_PAGE_SIZE constant at module level. Value unchanged — Superset API caps page_size at 100. +# @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API для аутентификации, пагинации, импорта/экспорта. +# @RELATION DEPENDS_ON -> [AsyncAPIClient] +from datetime import datetime from pathlib import Path from typing import Any, Optional, cast import zipfile -from requests import Response +import httpx DEFAULT_PAGE_SIZE = 100 from ..config_models import Environment from ..logger import belief_scope, logger as app_logger from ..utils.fileio import get_filename_from_headers -from ..utils.network import APIClient, SupersetAPIError +from ..utils.async_network import AsyncAPIClient +from ..utils.network import SupersetAPIError app_logger = cast(Any, app_logger) -# #region SupersetClientBase [C:3] [TYPE Class] +# #region SupersetClientBase [C:4] [TYPE Class] # @BRIEF Base class providing Superset client initialization, auth, pagination, and import/export plumbing. # @RELATION DEPENDS_ON -> [ConfigModels] -# @RELATION DEPENDS_ON -> [APIClient] +# @RELATION DEPENDS_ON -> [AsyncAPIClient] # @RELATION DEPENDS_ON -> [SupersetAPIError] +# @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API для аутентификации, пагинации, импорта/экспорта. class SupersetClientBase: - # #region SupersetClientInit [TYPE Function] [C:3] + # #region SupersetClientInit [TYPE Function] [C:4] # @PURPOSE Инициализирует клиент, проверяет конфигурацию и создает сетевой клиент. # @RELATION DEPENDS_ON -> [Environment] - # @RELATION DEPENDS_ON -> [APIClient] - def __init__(self, env: Environment): + # @RELATION DEPENDS_ON -> [AsyncAPIClient] + def __init__(self, env: Environment, client: AsyncAPIClient | None = None): with belief_scope("SupersetClientInit"): app_logger.reason( "Initializing Superset client for environment", @@ -45,11 +47,15 @@ class SupersetClientBase: "provider": "db", "refresh": "true", } - self.network = APIClient( - config={"base_url": env.url, "auth": auth_payload}, - verify_ssl=env.verify_ssl, - timeout=env.timeout, - ) + if client is not None: + self.client = client + else: + self.client = AsyncAPIClient( + config={"base_url": env.url, "auth": auth_payload}, + verify_ssl=env.verify_ssl, + timeout=env.timeout, + ) + self.network = self.client # alias for backward compat during migration self.delete_before_reimport: bool = False app_logger.reflect( "Superset client initialized", @@ -58,14 +64,15 @@ class SupersetClientBase: # #endregion SupersetClientInit # #region SupersetClientAuthenticate [TYPE Function] [C:3] # @PURPOSE Authenticates the client using the configured credentials. - # @RELATION CALLS -> [APIClient] - def authenticate(self) -> dict[str, str]: + # @SIDE_EFFECT Асинхронный HTTP-вызов к Superset API для аутентификации. + # @RELATION CALLS -> [AsyncAPIClient] + async def authenticate(self) -> dict[str, str]: with belief_scope("SupersetClientAuthenticate"): app_logger.reason( "Authenticating Superset client", extra={"environment": getattr(self.env, "id", None)}, ) - tokens = self.network.authenticate() + tokens = await self.client.authenticate() app_logger.reflect( "Superset client authentication completed", extra={ @@ -75,29 +82,30 @@ class SupersetClientBase: ) return tokens # #endregion SupersetClientAuthenticate - @property # #region SupersetClientHeaders [TYPE Function] [C:1] # @PURPOSE Возвращает базовые HTTP-заголовки, используемые сетевым клиентом. - def headers(self) -> dict: + # @SIDE_EFFECT Асинхронный HTTP-вызов к Superset API для получения заголовков. + async def get_headers(self) -> dict: with belief_scope("headers"): - return self.network.headers + return await self.client.get_headers() # #endregion SupersetClientHeaders # --- Pagination helpers --- # #region SupersetClientValidateQueryParams [TYPE Function] [C:1] # @PURPOSE Ensures query parameters have default page and page_size. + # @POST Returns query dict with defaults applied. def _validate_query_params(self, query: dict | None) -> dict: - with belief_scope("_validate_query_params"): - # Superset list endpoints commonly cap page_size at 100. - # Using 100 avoids partial fetches when larger values are silently truncated. - base_query = {"page": 0, "page_size": DEFAULT_PAGE_SIZE} - return {**base_query, **(query or {})} + # Superset list endpoints commonly cap page_size at 100. + # Using 100 avoids partial fetches when larger values are silently truncated. + base_query = {"page": 0, "page_size": DEFAULT_PAGE_SIZE} + return {**base_query, **(query or {})} # #endregion SupersetClientValidateQueryParams # #region SupersetClientFetchTotalObjectCount [TYPE Function] [C:1] # @PURPOSE Fetches the total number of items for a given endpoint. - # @RELATION CALLS -> [APIClient] - def _fetch_total_object_count(self, endpoint: str) -> int: + # @SIDE_EFFECT Асинхронный HTTP-вызов к Superset API для получения количества объектов. + # @RELATION CALLS -> [AsyncAPIClient] + async def _fetch_total_object_count(self, endpoint: str) -> int: with belief_scope("_fetch_total_object_count"): - return self.network.fetch_paginated_count( + return await self.client.fetch_paginated_count( endpoint=endpoint, query_params={"page": 0, "page_size": 1}, count_field="count", @@ -105,18 +113,20 @@ class SupersetClientBase: # #endregion SupersetClientFetchTotalObjectCount # #region SupersetClientFetchAllPages [TYPE Function] [C:1] # @PURPOSE Iterates through all pages to collect all data items. - # @RELATION CALLS -> [APIClient] - def _fetch_all_pages(self, endpoint: str, pagination_options: dict) -> list[dict]: + # @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API для пагинации. + # @RELATION CALLS -> [AsyncAPIClient] + async def _fetch_all_pages(self, endpoint: str, pagination_options: dict) -> list[dict]: with belief_scope("_fetch_all_pages"): - return self.network.fetch_paginated_data( + return await self.client.fetch_paginated_data( endpoint=endpoint, pagination_options=pagination_options ) # #endregion SupersetClientFetchAllPages # --- Import/Export helpers --- # #region SupersetClientDoImport [TYPE Function] [C:1] # @PURPOSE Performs the actual multipart upload for import. - # @RELATION CALLS -> [APIClient] - def _do_import(self, file_name: str | Path) -> dict: + # @SIDE_EFFECT Асинхронный multipart HTTP-запрос к Superset API для импорта дашборда. + # @RELATION CALLS -> [AsyncAPIClient] + async def _do_import(self, file_name: str | Path) -> dict: with belief_scope("_do_import"): app_logger.debug(f"[_do_import][State] Uploading file: {file_name}") file_path = Path(file_name) @@ -125,7 +135,7 @@ class SupersetClientBase: f"[_do_import][Failure] File does not exist: {file_name}" ) raise FileNotFoundError(f"File does not exist: {file_name}") - return self.network.upload_file( + return await self.client.upload_file( endpoint="/dashboard/import/", file_info={ "file_obj": file_path, @@ -138,7 +148,7 @@ class SupersetClientBase: # #endregion SupersetClientDoImport # #region SupersetClientValidateExportResponse [TYPE Function] [C:1] # @PURPOSE Validates that the export response is a non-empty ZIP archive. - def _validate_export_response(self, response: Response, dashboard_id: int) -> None: + def _validate_export_response(self, response: httpx.Response, dashboard_id: int) -> None: with belief_scope("_validate_export_response"): content_type = response.headers.get("Content-Type", "") if "application/zip" not in content_type: @@ -150,11 +160,10 @@ class SupersetClientBase: # #endregion SupersetClientValidateExportResponse # #region SupersetClientResolveExportFilename [TYPE Function] [C:1] # @PURPOSE Determines the filename for an exported dashboard. - def _resolve_export_filename(self, response: Response, dashboard_id: int) -> str: + def _resolve_export_filename(self, response: httpx.Response, dashboard_id: int) -> str: with belief_scope("_resolve_export_filename"): filename = get_filename_from_headers(dict(response.headers)) if not filename: - from datetime import datetime timestamp = datetime.now().strftime("%Y%m%dT%H%M%S") filename = f"dashboard_export_{dashboard_id}_{timestamp}.zip" app_logger.warning( @@ -181,7 +190,7 @@ class SupersetClientBase: # #region SupersetClientResolveTargetIdForDelete [TYPE Function] [C:1] # @PURPOSE Resolves a dashboard ID from either an ID or a slug. # @RELATION CALLS -> [SupersetClientGetDashboards] - def _resolve_target_id_for_delete( + async def _resolve_target_id_for_delete( self, dash_id: int | None, dash_slug: str | None ) -> int | None: with belief_scope("_resolve_target_id_for_delete"): @@ -193,7 +202,7 @@ class SupersetClientBase: dash_slug, ) try: - _, candidates = self.get_dashboards( + _, candidates = await self.get_dashboards( query={ "filters": [{"col": "slug", "op": "eq", "value": dash_slug}] } @@ -215,8 +224,9 @@ class SupersetClientBase: # #endregion SupersetClientResolveTargetIdForDelete # #region SupersetClientGetAllResources [TYPE Function] [C:3] # @PURPOSE Fetches all resources of a given type with id, uuid, and name columns. + # @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API для получения ресурсов. # @RELATION CALLS -> [SupersetClientFetchAllPages] - def get_all_resources( + async def get_all_resources( self, resource_type: str, since_dttm: Optional["datetime"] = None ) -> list[dict]: with belief_scope( @@ -247,13 +257,12 @@ class SupersetClientBase: query = {"columns": config["columns"]} if since_dttm: import math - # Use int milliseconds to be safe timestamp_ms = math.floor(since_dttm.timestamp() * 1000) query["filters"] = [ {"col": "changed_on_dttm", "opr": "gt", "value": timestamp_ms} ] validated = self._validate_query_params(query) - data = self._fetch_all_pages( + data = await self._fetch_all_pages( endpoint=config["endpoint"], pagination_options={"base_query": validated, "results_field": "result"}, ) diff --git a/backend/src/core/superset_client/_charts.py b/backend/src/core/superset_client/_charts.py index 913255be..448afe9e 100644 --- a/backend/src/core/superset_client/_charts.py +++ b/backend/src/core/superset_client/_charts.py @@ -2,6 +2,7 @@ # @LAYER Infrastructure # @BRIEF Chart domain mixin for SupersetClient — list, get, extract IDs from layout. # @RELATION DEPENDS_ON -> [SupersetClientBase] +# @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API для работы с чартами. import re from typing import Any, cast @@ -11,24 +12,27 @@ app_logger = cast(Any, app_logger) # #region SupersetChartsMixin [C:3] [TYPE Class] # @BRIEF Mixin providing all chart-related Superset API operations. # @RELATION DEPENDS_ON -> [SupersetClientBase] +# @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API. class SupersetChartsMixin: # #region SupersetClientGetChart [TYPE Function] [C:3] # @PURPOSE: Fetches a single chart by ID. - # @RELATION CALLS -> [APIClient] - def get_chart(self, chart_id: int) -> dict: + # @SIDE_EFFECT Асинхронный HTTP-вызов к Superset API. + # @RELATION CALLS -> [AsyncAPIClient] + async def get_chart(self, chart_id: int) -> dict: with belief_scope("SupersetClient.get_chart", f"id={chart_id}"): - response = self.network.request(method="GET", endpoint=f"/chart/{chart_id}") + response = await self.client.request(method="GET", endpoint=f"/chart/{chart_id}") return cast(dict, response) # #endregion SupersetClientGetChart # #region SupersetClientGetCharts [TYPE Function] [C:3] # @PURPOSE: Fetches all charts with pagination support. + # @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API для пагинации. # @RELATION CALLS -> [SupersetClientFetchAllPages] - def get_charts(self, query: dict | None = None) -> tuple[int, list[dict]]: + async def get_charts(self, query: dict | None = None) -> tuple[int, list[dict]]: with belief_scope("get_charts"): validated_query = self._validate_query_params(query or {}) if "columns" not in validated_query: validated_query["columns"] = ["id", "uuid", "slice_name", "viz_type"] - paginated_data = self._fetch_all_pages( + paginated_data = await self._fetch_all_pages( endpoint="/chart/", pagination_options={ "base_query": validated_query, diff --git a/backend/src/core/superset_client/_dashboards_crud.py b/backend/src/core/superset_client/_dashboards_crud.py index 15375141..31868b7c 100644 --- a/backend/src/core/superset_client/_dashboards_crud.py +++ b/backend/src/core/superset_client/_dashboards_crud.py @@ -4,12 +4,13 @@ # @RELATION DEPENDS_ON -> [SupersetClientBase] # @RELATION DEPENDS_ON -> [SupersetDashboardsFiltersMixin] # @RELATION DEPENDS_ON -> [SupersetChartsMixin] +# @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API для CRUD-операций с дашбордами. import json from pathlib import Path import re from typing import Any, cast -from requests import Response +import httpx from ..logger import belief_scope, logger as app_logger @@ -19,16 +20,18 @@ app_logger = cast(Any, app_logger) # @RELATION DEPENDS_ON -> [SupersetClientBase] # @RELATION DEPENDS_ON -> [SupersetDashboardsFiltersMixin] # @RELATION DEPENDS_ON -> [SupersetChartsMixin] +# @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API. class SupersetDashboardsCrudMixin: # #region SupersetClientGetDashboardDetail [TYPE Function] [C:3] # @PURPOSE: Fetches detailed dashboard information including related charts and datasets. + # @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API. # @RELATION CALLS -> [SupersetClientGetDashboard] # @RELATION CALLS -> [SupersetClientGetChart] - def get_dashboard_detail(self, dashboard_ref: int | str) -> dict: + async def get_dashboard_detail(self, dashboard_ref: int | str) -> dict: with belief_scope( "SupersetClient.get_dashboard_detail", f"ref={dashboard_ref}" ): - dashboard_response = self.get_dashboard(dashboard_ref) + dashboard_response = await self.get_dashboard(dashboard_ref) dashboard_data = dashboard_response.get("result", dashboard_response) charts: list[dict] = [] datasets: list[dict] = [] @@ -59,7 +62,7 @@ class SupersetDashboardsCrudMixin: return None # #endregion extract_dataset_id_from_form_data try: - charts_response = self.network.request( + charts_response = await self.client.request( method="GET", endpoint=f"/dashboard/{dashboard_ref}/charts" ) charts_payload = ( @@ -101,7 +104,7 @@ class SupersetDashboardsCrudMixin: "[get_dashboard_detail][Warning] Failed to fetch dashboard charts: %s", e, ) try: - datasets_response = self.network.request( + datasets_response = await self.client.request( method="GET", endpoint=f"/dashboard/{dashboard_ref}/datasets" ) datasets_payload = ( @@ -174,7 +177,7 @@ class SupersetDashboardsCrudMixin: ) for chart_id in sorted(chart_ids_from_position): try: - chart_response = self.get_chart(int(chart_id)) + chart_response = await self.get_chart(int(chart_id)) chart_data = chart_response.get("result", chart_response) charts.append({ "id": int(chart_id), @@ -208,7 +211,7 @@ class SupersetDashboardsCrudMixin: continue for dataset_id in missing_dataset_ids: try: - dataset_response = self.get_dataset(int(dataset_id)) + dataset_response = await self.get_dataset(int(dataset_id)) dataset_data = dataset_response.get("result", dataset_response) db_payload = dataset_data.get("database") db_name = ( @@ -255,21 +258,20 @@ class SupersetDashboardsCrudMixin: # #endregion SupersetClientGetDashboardDetail # #region SupersetClientExportDashboard [TYPE Function] [C:3] # @PURPOSE: Экспортирует дашборд в виде ZIP-архива. - # @SIDE_EFFECT Performs network I/O to download archive. - # @RELATION CALLS -> [APIClient] - def export_dashboard(self, dashboard_id: int) -> tuple[bytes, str]: + # @SIDE_EFFECT Асинхронный HTTP-вызов к Superset API для экспорта дашборда. + # @RELATION CALLS -> [AsyncAPIClient] + async def export_dashboard(self, dashboard_id: int) -> tuple[bytes, str]: with belief_scope("export_dashboard"): app_logger.info( "[export_dashboard][Enter] Exporting dashboard %s.", dashboard_id ) - response = self.network.request( + response = await self.client.request( method="GET", endpoint="/dashboard/export/", params={"q": json.dumps([dashboard_id])}, - stream=True, raw_response=True, ) - response = cast(Response, response) + response = cast(httpx.Response, response) self._validate_export_response(response, dashboard_id) filename = self._resolve_export_filename(response, dashboard_id) app_logger.reflect( @@ -280,10 +282,10 @@ class SupersetDashboardsCrudMixin: # #endregion SupersetClientExportDashboard # #region SupersetClientImportDashboard [TYPE Function] [C:3] # @PURPOSE: Импортирует дашборд из ZIP-файла. - # @SIDE_EFFECT Performs network I/O to upload archive. + # @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API для импорта дашборда. # @RELATION CALLS -> [SupersetClientDoImport] - # @RELATION CALLS -> [APIClient] - def import_dashboard( + # @RELATION CALLS -> [AsyncAPIClient] + async def import_dashboard( self, file_name: str | Path, dash_id: int | None = None, @@ -295,7 +297,7 @@ class SupersetDashboardsCrudMixin: file_path = str(file_name) self._validate_import_file(file_path) try: - return self._do_import(file_path) + return await self._do_import(file_path) except Exception as exc: app_logger.error( "[import_dashboard][Failure] First import attempt failed: %s", @@ -303,29 +305,29 @@ class SupersetDashboardsCrudMixin: ) if not self.delete_before_reimport: raise - target_id = self._resolve_target_id_for_delete(dash_id, dash_slug) + target_id = await self._resolve_target_id_for_delete(dash_id, dash_slug) if target_id is None: app_logger.error( "[import_dashboard][Failure] No ID available for delete-retry." ) raise - self.delete_dashboard(target_id) + await self.delete_dashboard(target_id) app_logger.info( "[import_dashboard][State] Deleted dashboard ID %s, retrying import.", target_id, ) - return self._do_import(file_path) + return await self._do_import(file_path) # #endregion SupersetClientImportDashboard # #region SupersetClientDeleteDashboard [TYPE Function] [C:3] # @PURPOSE: Удаляет дашборд по его ID или slug. - # @SIDE_EFFECT Deletes resource from upstream Superset environment. - # @RELATION CALLS -> [APIClient] - def delete_dashboard(self, dashboard_id: int | str) -> None: + # @SIDE_EFFECT Асинхронный HTTP-вызов к Superset API для удаления дашборда. + # @RELATION CALLS -> [AsyncAPIClient] + async def delete_dashboard(self, dashboard_id: int | str) -> None: with belief_scope("delete_dashboard"): app_logger.info( "[delete_dashboard][Enter] Deleting dashboard %s.", dashboard_id ) - response = self.network.request( + response = await self.client.request( method="DELETE", endpoint=f"/dashboard/{dashboard_id}" ) response = cast(dict, response) diff --git a/backend/src/core/superset_client/_dashboards_filters.py b/backend/src/core/superset_client/_dashboards_filters.py index 50326898..0cd1e9b7 100644 --- a/backend/src/core/superset_client/_dashboards_filters.py +++ b/backend/src/core/superset_client/_dashboards_filters.py @@ -2,6 +2,7 @@ # @LAYER Infrastructure # @BRIEF Dashboard native filter extraction mixin for SupersetClient. # @RELATION DEPENDS_ON -> [SupersetClientBase] +# @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API для работы с фильтрами. import json from typing import Any, cast @@ -11,40 +12,44 @@ app_logger = cast(Any, app_logger) # #region SupersetDashboardsFiltersMixin [C:3] [TYPE Class] # @BRIEF Mixin providing dashboard native filter extraction from permalink and URL state. # @RELATION DEPENDS_ON -> [SupersetClientBase] +# @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API. class SupersetDashboardsFiltersMixin: # #region SupersetClientGetDashboard [TYPE Function] [C:3] # @PURPOSE: Fetches a single dashboard by ID or slug. - # @RELATION CALLS -> [APIClient] - def get_dashboard(self, dashboard_ref: int | str) -> dict: + # @SIDE_EFFECT Асинхронный HTTP-вызов к Superset API. + # @RELATION CALLS -> [AsyncAPIClient] + async def get_dashboard(self, dashboard_ref: int | str) -> dict: with belief_scope("SupersetClient.get_dashboard", f"ref={dashboard_ref}"): - response = self.network.request( + response = await self.client.request( method="GET", endpoint=f"/dashboard/{dashboard_ref}" ) return cast(dict, response) # #endregion SupersetClientGetDashboard # #region SupersetClientGetDashboardPermalinkState [TYPE Function] [C:2] # @PURPOSE: Fetches stored dashboard permalink state by permalink key. - # @RELATION CALLS -> [APIClient] - def get_dashboard_permalink_state(self, permalink_key: str) -> dict: + # @SIDE_EFFECT Асинхронный HTTP-вызов к Superset API. + # @RELATION CALLS -> [AsyncAPIClient] + async def get_dashboard_permalink_state(self, permalink_key: str) -> dict: with belief_scope( "SupersetClient.get_dashboard_permalink_state", f"key={permalink_key}" ): - response = self.network.request( + response = await self.client.request( method="GET", endpoint=f"/dashboard/permalink/{permalink_key}" ) return cast(dict, response) # #endregion SupersetClientGetDashboardPermalinkState # #region SupersetClientGetNativeFilterState [TYPE Function] [C:2] # @PURPOSE: Fetches stored native filter state by filter state key. - # @RELATION CALLS -> [APIClient] - def get_native_filter_state( + # @SIDE_EFFECT Асинхронный HTTP-вызов к Superset API. + # @RELATION CALLS -> [AsyncAPIClient] + async def get_native_filter_state( self, dashboard_id: int | str, filter_state_key: str ) -> dict: with belief_scope( "SupersetClient.get_native_filter_state", f"dashboard={dashboard_id}, key={filter_state_key}", ): - response = self.network.request( + response = await self.client.request( method="GET", endpoint=f"/dashboard/{dashboard_id}/filter_state/{filter_state_key}", ) @@ -53,12 +58,12 @@ class SupersetDashboardsFiltersMixin: # #region SupersetClientExtractNativeFiltersFromPermalink [TYPE Function] [C:3] # @PURPOSE: Extract native filters dataMask from a permalink key. # @RELATION CALLS -> [SupersetClientGetDashboardPermalinkState] - def extract_native_filters_from_permalink(self, permalink_key: str) -> dict: + async def extract_native_filters_from_permalink(self, permalink_key: str) -> dict: with belief_scope( "SupersetClient.extract_native_filters_from_permalink", f"key={permalink_key}", ): - permalink_response = self.get_dashboard_permalink_state(permalink_key) + permalink_response = await self.get_dashboard_permalink_state(permalink_key) result = permalink_response.get("result", permalink_response) state = result.get("state", result) data_mask = state.get("dataMask", {}) @@ -82,14 +87,14 @@ class SupersetDashboardsFiltersMixin: # #region SupersetClientExtractNativeFiltersFromKey [TYPE Function] [C:3] # @PURPOSE: Extract native filters from a native_filters_key URL parameter. # @RELATION CALLS -> [SupersetClientGetNativeFilterState] - def extract_native_filters_from_key( + async def extract_native_filters_from_key( self, dashboard_id: int | str, filter_state_key: str ) -> dict: with belief_scope( "SupersetClient.extract_native_filters_from_key", f"dashboard={dashboard_id}, key={filter_state_key}", ): - filter_response = self.get_native_filter_state( + filter_response = await self.get_native_filter_state( dashboard_id, filter_state_key ) result = filter_response.get("result", filter_response) @@ -134,7 +139,7 @@ class SupersetDashboardsFiltersMixin: # @PURPOSE: Parse a Superset dashboard URL and extract native filter state if present. # @RELATION CALLS -> [SupersetClientExtractNativeFiltersFromPermalink] # @RELATION CALLS -> [SupersetClientExtractNativeFiltersFromKey] - def parse_dashboard_url_for_filters(self, url: str) -> dict: + async def parse_dashboard_url_for_filters(self, url: str) -> dict: with belief_scope( "SupersetClient.parse_dashboard_url_for_filters", f"url={url}" ): @@ -154,7 +159,7 @@ class SupersetDashboardsFiltersMixin: p_index = path_parts.index("p") if p_index + 1 < len(path_parts): permalink_key = path_parts[p_index + 1] - filter_data = self.extract_native_filters_from_permalink( + filter_data = await self.extract_native_filters_from_permalink( permalink_key ) result["filter_type"] = "permalink" @@ -181,7 +186,7 @@ class SupersetDashboardsFiltersMixin: resolved_id = int(dashboard_ref) except (ValueError, TypeError): try: - dash_resp = self.get_dashboard(dashboard_ref) + dash_resp = await self.get_dashboard(dashboard_ref) dash_data = ( dash_resp.get("result", dash_resp) if isinstance(dash_resp, dict) @@ -197,7 +202,7 @@ class SupersetDashboardsFiltersMixin: e, ) if resolved_id is not None: - filter_data = self.extract_native_filters_from_key( + filter_data = await self.extract_native_filters_from_key( resolved_id, native_filters_key ) result["filter_type"] = "native_filters_key" diff --git a/backend/src/core/superset_client/_dashboards_list.py b/backend/src/core/superset_client/_dashboards_list.py index 4798bd9b..6b388ae6 100644 --- a/backend/src/core/superset_client/_dashboards_list.py +++ b/backend/src/core/superset_client/_dashboards_list.py @@ -3,6 +3,7 @@ # @BRIEF Dashboard listing mixin for SupersetClient — paginated list, summary projection. # @RELATION DEPENDS_ON -> [SupersetClientBase] # @RELATION DEPENDS_ON -> [SupersetUserProjectionMixin] +# @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API для получения списка дашбордов. import json from typing import Any, cast @@ -13,11 +14,13 @@ app_logger = cast(Any, app_logger) # @BRIEF Mixin providing dashboard listing and summary projection operations. # @RELATION DEPENDS_ON -> [SupersetClientBase] # @RELATION DEPENDS_ON -> [SupersetUserProjectionMixin] +# @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API. class SupersetDashboardsListMixin: # #region SupersetClientGetDashboards [TYPE Function] [C:3] # @PURPOSE: Получает полный список дашбордов, автоматически обрабатывая пагинацию. + # @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API для пагинации. # @RELATION CALLS -> [SupersetClientFetchAllPages] - def get_dashboards(self, query: dict | None = None) -> tuple[int, list[dict]]: + async def get_dashboards(self, query: dict | None = None) -> tuple[int, list[dict]]: with belief_scope("get_dashboards"): app_logger.info("[get_dashboards][Enter] Fetching dashboards.") validated_query = self._validate_query_params(query or {}) @@ -26,7 +29,7 @@ class SupersetDashboardsListMixin: "slug", "id", "url", "changed_on_utc", "dashboard_title", "published", "created_by", "changed_by", "changed_by_name", "owners", ] - paginated_data = self._fetch_all_pages( + paginated_data = await self._fetch_all_pages( endpoint="/dashboard/", pagination_options={ "base_query": validated_query, @@ -39,8 +42,9 @@ class SupersetDashboardsListMixin: # #endregion SupersetClientGetDashboards # #region SupersetClientGetDashboardsPage [TYPE Function] [C:3] # @PURPOSE: Fetches a single dashboards page from Superset without iterating all pages. - # @RELATION CALLS -> [APIClient] - def get_dashboards_page( + # @SIDE_EFFECT Асинхронный HTTP-вызов к Superset API. + # @RELATION CALLS -> [AsyncAPIClient] + async def get_dashboards_page( self, query: dict | None = None ) -> tuple[int, list[dict]]: with belief_scope("get_dashboards_page"): @@ -52,7 +56,7 @@ class SupersetDashboardsListMixin: ] response_json = cast( dict[str, Any], - self.network.request( + await self.client.request( method="GET", endpoint="/dashboard/", params={"q": json.dumps(validated_query)}, @@ -64,13 +68,14 @@ class SupersetDashboardsListMixin: # #endregion SupersetClientGetDashboardsPage # #region SupersetClientGetDashboardsSummary [TYPE Function] [C:3] # @PURPOSE: Fetches dashboard metadata optimized for the grid. + # @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API. # @RELATION CALLS -> [SupersetClientGetDashboards] - def get_dashboards_summary(self, require_slug: bool = False) -> list[dict]: + async def get_dashboards_summary(self, require_slug: bool = False) -> list[dict]: with belief_scope("SupersetClient.get_dashboards_summary"): query: dict[str, Any] = {} if require_slug: query["filters"] = [{"col": "slug", "opr": "neq", "value": ""}] - _, dashboards = self.get_dashboards(query=query) + _, dashboards = await self.get_dashboards(query=query) result = [] max_debug_samples = 12 for index, dash in enumerate(dashboards): @@ -118,8 +123,9 @@ class SupersetDashboardsListMixin: # #endregion SupersetClientGetDashboardsSummary # #region SupersetClientGetDashboardsSummaryPage [TYPE Function] [C:3] # @PURPOSE: Fetches one page of dashboard metadata optimized for the grid. + # @SIDE_EFFECT Асинхронный HTTP-вызов к Superset API. # @RELATION CALLS -> [SupersetClientGetDashboardsPage] - def get_dashboards_summary_page( + async def get_dashboards_summary_page( self, page: int, page_size: int, @@ -141,7 +147,7 @@ class SupersetDashboardsListMixin: }) if filters: query["filters"] = filters - total_count, dashboards = self.get_dashboards_page(query=query) + total_count, dashboards = await self.get_dashboards_page(query=query) result = [] for dash in dashboards: owners = self._extract_owner_labels(dash.get("owners")) diff --git a/backend/src/core/superset_client/_dashboards_write.py b/backend/src/core/superset_client/_dashboards_write.py index 39ed74c0..e770c22c 100644 --- a/backend/src/core/superset_client/_dashboards_write.py +++ b/backend/src/core/superset_client/_dashboards_write.py @@ -1,12 +1,12 @@ -# #region SupersetDashboardsWriteMixin [C:4] [TYPE Module] [SEMANTICS superset, dashboard, chart, write, maintenance, banner] +# #region SupersetDashboardsWriteMixin [C:4] [TYPE Module] [SEMANTICS superset, dashboard, chart, write, maintenance, banner, async] # @BRIEF Dashboard write mixin for SupersetClient — markdown chart CRUD and layout manipulation for maintenance banners. # @LAYER Infrastructure # @RELATION DEPENDS_ON -> [SupersetClientBase] # @RELATION DEPENDS_ON -> [SupersetChartsMixin] # @RELATION DEPENDS_ON -> [LayoutUtils] -# @SIDE_EFFECT Creates, updates, deletes markdown charts & modifies dashboard layout via Superset API. +# @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API: создаёт, обновляет, удаляет markdown-диаграммы и изменяет лейаут дашбордов. # @DATA_CONTRACT Input: dashboard_id / chart_id / markdown_text → Output: API response dict or chart_id int -# @INVARIANT All network operations use self.network.request() +# @INVARIANT All network operations use self.client.request() # @RATIONALE Extracted from monolithic superset_client to satisfy INV_7. # Uses belief_scope with REASON/REFLECT/EXPLORE markers for all C4 operations per Std.Semantics.Python. @@ -28,6 +28,7 @@ app_logger = cast(Any, app_logger) # @BRIEF Mixin providing markdown chart CRUD and dashboard layout manipulation for maintenance banners. # @RELATION DEPENDS_ON -> [SupersetClientBase] # @RELATION DEPENDS_ON -> [SupersetChartsMixin] +# @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API. class SupersetDashboardsWriteMixin: """Mixin for creating/updating/deleting markdown charts and manipulating dashboard layouts.""" @@ -35,9 +36,9 @@ class SupersetDashboardsWriteMixin: # @BRIEF Create a markdown chart and return the new chart ID. # @PRE dashboard_id exists in Superset. markdown_text not empty. # @POST Returns chart_id (int). Chart is NOT yet placed in dashboard layout. - # @SIDE_EFFECT Creates a new chart resource in Superset. - # @RELATION CALLS -> [EXT:method:self.network.request] - def create_markdown_chart(self, dashboard_id: int, markdown_text: str) -> int: + # @SIDE_EFFECT Асинхронный HTTP-вызов: создаёт новый chart resource в Superset. + # @RELATION CALLS -> [EXT:method:self.client.request] + async def create_markdown_chart(self, dashboard_id: int, markdown_text: str) -> int: with belief_scope( "SupersetDashboardsWriteMixin.create_markdown_chart", f"dashboard_id={dashboard_id}", @@ -46,7 +47,7 @@ class SupersetDashboardsWriteMixin: "Creating markdown chart", extra={"dashboard_id": dashboard_id, "len": len(markdown_text)}, ) - datasource_id = self._resolve_markdown_datasource(dashboard_id) + datasource_id = await self._resolve_markdown_datasource(dashboard_id) payload = { "dashboards": [dashboard_id], "datasource_id": datasource_id, @@ -59,8 +60,8 @@ class SupersetDashboardsWriteMixin: }), } try: - response = cast(dict, self.network.request( - method="POST", endpoint="/chart/", json=payload + response = cast(dict, await self.client.request( + method="POST", endpoint="/chart/", data=payload )) chart_id = int(response.get("id") or response.get("result", {}).get("id", 0)) app_logger.reflect("Chart created", extra={"chart_id": chart_id}) @@ -74,8 +75,8 @@ class SupersetDashboardsWriteMixin: # @BRIEF Update the markdown text of an existing markdown chart. # @PRE chart_id exists and is a markdown chart. # @POST Chart markdown content updated. - # @SIDE_EFFECT Modifies a chart resource in Superset. - def update_markdown_chart(self, chart_id: int, markdown_text: str) -> dict: + # @SIDE_EFFECT Асинхронный HTTP-вызов: изменяет chart resource в Superset. + async def update_markdown_chart(self, chart_id: int, markdown_text: str) -> dict: with belief_scope( "SupersetDashboardsWriteMixin.update_markdown_chart", f"chart_id={chart_id}", @@ -83,7 +84,7 @@ class SupersetDashboardsWriteMixin: app_logger.reason("Updating markdown chart", extra={"chart_id": chart_id, "len": len(markdown_text)}) payload = {"params": json.dumps({"markdown": markdown_text, "viz_type": "markdown"})} try: - response = cast(dict, self.network.request(method="PUT", endpoint=f"/chart/{chart_id}", json=payload)) + response = cast(dict, await self.client.request(method="PUT", endpoint=f"/chart/{chart_id}", data=payload)) app_logger.reflect("Chart updated", extra={"chart_id": chart_id}) return response except Exception as e: @@ -96,8 +97,8 @@ class SupersetDashboardsWriteMixin: # shift existing charts down. Uses native MARKDOWN for reliable rendering. # @PRE dashboard_id exists in Superset. chart_id is used for tracking/DB. # @POST MARKDOWN element placed at top; existing items shifted down. - # @SIDE_EFFECT Modifies dashboard layout JSON. - def update_dashboard_layout(self, dashboard_id: int, chart_id: int) -> dict: + # @SIDE_EFFECT Асинхронные HTTP-вызовы: читает и изменяет layout JSON дашборда. + async def update_dashboard_layout(self, dashboard_id: int, chart_id: int) -> dict: with belief_scope( "SupersetDashboardsWriteMixin.update_dashboard_layout", f"dashboard_id={dashboard_id}, chart_id={chart_id}", @@ -107,7 +108,7 @@ class SupersetDashboardsWriteMixin: extra={"dashboard_id": dashboard_id, "chart_id": chart_id}, ) try: - dashboard = cast(dict, self.network.request( + dashboard = cast(dict, await self.client.request( method="GET", endpoint=f"/dashboard/{dashboard_id}" )) result = dashboard.get("result", dashboard) @@ -117,9 +118,9 @@ class SupersetDashboardsWriteMixin: placeholder = "*Maintenance banner*" insert_banner_markdown_at_top(position_json, chart_id, placeholder) - response = cast(dict, self.network.request( + response = cast(dict, await self.client.request( method="PUT", endpoint=f"/dashboard/{dashboard_id}", - json={"position_json": json.dumps(position_json)}, + data={"position_json": json.dumps(position_json)}, )) app_logger.reflect( "Layout updated with native MARKDOWN", @@ -139,8 +140,8 @@ class SupersetDashboardsWriteMixin: # @BRIEF Remove banner markdown element from dashboard layout without deleting the chart. # @PRE dashboard_id exists. chart_id exists in layout as MARKDOWN-banner-{chart_id}. # @POST MARKDOWN element removed from position_json; items shifted up. - # @SIDE_EFFECT Modifies dashboard layout JSON. - def remove_chart_from_layout(self, dashboard_id: int, chart_id: int) -> dict: + # @SIDE_EFFECT Асинхронные HTTP-вызовы: читает и изменяет layout JSON дашборда. + async def remove_chart_from_layout(self, dashboard_id: int, chart_id: int) -> dict: with belief_scope( "SupersetDashboardsWriteMixin.remove_chart_from_layout", f"dashboard_id={dashboard_id}, chart_id={chart_id}", @@ -150,17 +151,16 @@ class SupersetDashboardsWriteMixin: extra={"dashboard_id": dashboard_id, "chart_id": chart_id}, ) try: - dashboard = cast(dict, self.network.request( + dashboard = cast(dict, await self.client.request( method="GET", endpoint=f"/dashboard/{dashboard_id}" )) result = dashboard.get("result", dashboard) position_json = parse_position_json(result.get("position_json", "{}")) - # Remove new MARKDOWN+ROW pair (old CHART-{id} format no longer created) remove_banner_from_position(position_json, chart_id) - response = cast(dict, self.network.request( + response = cast(dict, await self.client.request( method="PUT", endpoint=f"/dashboard/{dashboard_id}", - json={"position_json": json.dumps(position_json)}, + data={"position_json": json.dumps(position_json)}, )) app_logger.reflect( "Banner removed from layout", @@ -180,12 +180,12 @@ class SupersetDashboardsWriteMixin: # @BRIEF Delete a chart from Superset by ID. # @PRE chart_id exists. # @POST Chart permanently deleted from Superset. - # @SIDE_EFFECT Destroys chart resource. - def delete_chart(self, chart_id: int) -> dict: + # @SIDE_EFFECT Асинхронный HTTP-вызов: удаляет chart resource. + async def delete_chart(self, chart_id: int) -> dict: with belief_scope("SupersetDashboardsWriteMixin.delete_chart", f"chart_id={chart_id}"): app_logger.reason("Deleting chart", extra={"chart_id": chart_id}) try: - response = cast(dict, self.network.request(method="DELETE", endpoint=f"/chart/{chart_id}")) + response = cast(dict, await self.client.request(method="DELETE", endpoint=f"/chart/{chart_id}")) app_logger.reflect("Chart deleted", extra={"chart_id": chart_id}) return response except Exception as e: @@ -197,8 +197,8 @@ class SupersetDashboardsWriteMixin: # @BRIEF Update the content of a native MARKDOWN banner element on a dashboard. # @PRE dashboard_id exists. chart_id used for key lookup. # @POST MARKDOWN element content updated on dashboard. - # @SIDE_EFFECT Modifies dashboard layout JSON in Superset. - def update_banner_on_dashboard( + # @SIDE_EFFECT Асинхронные HTTP-вызовы: читает и изменяет layout JSON дашборда. + async def update_banner_on_dashboard( self, dashboard_id: int, chart_id: int, content: str ) -> dict: with belief_scope( @@ -210,16 +210,16 @@ class SupersetDashboardsWriteMixin: extra={"dashboard_id": dashboard_id, "chart_id": chart_id, "len": len(content)}, ) try: - dashboard = cast(dict, self.network.request( + dashboard = cast(dict, await self.client.request( method="GET", endpoint=f"/dashboard/{dashboard_id}" )) result = dashboard.get("result", dashboard) position_json = parse_position_json(result.get("position_json", "{}")) markdown_key = f"MARKDOWN-banner-{chart_id}" update_banner_markdown_content(position_json, markdown_key, content) - response = cast(dict, self.network.request( + response = cast(dict, await self.client.request( method="PUT", endpoint=f"/dashboard/{dashboard_id}", - json={"position_json": json.dumps(position_json)}, + data={"position_json": json.dumps(position_json)}, )) app_logger.reflect( "Banner content updated on dashboard", @@ -240,11 +240,12 @@ class SupersetDashboardsWriteMixin: # @BRIEF Fetch the position_json layout of a dashboard. # @PRE dashboard_id exists. # @POST Returns the dashboard layout dict. - def get_dashboard_layout(self, dashboard_id: int) -> dict: + # @SIDE_EFFECT Асинхронный HTTP-вызов к Superset API. + async def get_dashboard_layout(self, dashboard_id: int) -> dict: with belief_scope("SupersetDashboardsWriteMixin.get_dashboard_layout", f"dashboard_id={dashboard_id}"): app_logger.reason("Fetching layout", extra={"dashboard_id": dashboard_id}) try: - dashboard = cast(dict, self.network.request(method="GET", endpoint=f"/dashboard/{dashboard_id}")) + dashboard = cast(dict, await self.client.request(method="GET", endpoint=f"/dashboard/{dashboard_id}")) result = dashboard.get("result", dashboard) layout = parse_position_json(result.get("position_json", "{}")) app_logger.reflect("Layout fetched", extra={"dashboard_id": dashboard_id, "items": len(layout)}) @@ -258,10 +259,11 @@ class SupersetDashboardsWriteMixin: # @BRIEF Find a valid datasource_id for markdown chart creation. # @PRE dashboard_id exists. # @POST Returns an integer datasource_id. - def _resolve_markdown_datasource(self, dashboard_id: int) -> int: + # @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API. + async def _resolve_markdown_datasource(self, dashboard_id: int) -> int: with belief_scope("_resolve_markdown_datasource", f"dashboard_id={dashboard_id}"): try: - datasets_response = cast(dict, self.network.request( + datasets_response = cast(dict, await self.client.request( method="GET", endpoint=f"/dashboard/{dashboard_id}/datasets" )) datasets = datasets_response.get("result", []) @@ -273,7 +275,7 @@ class SupersetDashboardsWriteMixin: app_logger.explore("Dashboard datasets fetch failed, falling back", extra={"dashboard_id": dashboard_id}, error="Dashboard datasets endpoint failed") try: - all_ds = cast(dict, self.network.request( + all_ds = cast(dict, await self.client.request( method="GET", endpoint="/dataset/", params={"q": json.dumps({"columns": ["id"], "page_size": 1, "page": 0})}, )) diff --git a/backend/src/core/superset_client/_databases.py b/backend/src/core/superset_client/_databases.py index e22792a4..201348d2 100644 --- a/backend/src/core/superset_client/_databases.py +++ b/backend/src/core/superset_client/_databases.py @@ -2,6 +2,7 @@ # @LAYER Infrastructure # @BRIEF Database domain mixin for SupersetClient — list, get, summary, by_uuid. # @RELATION DEPENDS_ON -> [SupersetClientBase] +# @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API для работы с базами данных. from typing import Any, cast from ..logger import belief_scope, logger as app_logger @@ -10,17 +11,19 @@ app_logger = cast(Any, app_logger) # #region SupersetDatabasesMixin [C:3] [TYPE Class] # @BRIEF Mixin providing all database-related Superset API operations. # @RELATION DEPENDS_ON -> [SupersetClientBase] +# @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API. class SupersetDatabasesMixin: # #region SupersetClientGetDatabases [TYPE Function] [C:3] # @PURPOSE: Получает полный список баз данных. + # @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API для пагинации. # @RELATION CALLS -> [SupersetClientFetchAllPages] - def get_databases(self, query: dict | None = None) -> tuple[int, list[dict]]: + async def get_databases(self, query: dict | None = None) -> tuple[int, list[dict]]: with belief_scope("get_databases"): app_logger.info("[get_databases][Enter] Fetching databases.") validated_query = self._validate_query_params(query or {}) if "columns" not in validated_query: validated_query["columns"] = [] - paginated_data = self._fetch_all_pages( + paginated_data = await self._fetch_all_pages( endpoint="/database/", pagination_options={ "base_query": validated_query, @@ -33,11 +36,12 @@ class SupersetDatabasesMixin: # #endregion SupersetClientGetDatabases # #region SupersetClientGetDatabase [TYPE Function] [C:3] # @PURPOSE: Получает информацию о конкретной базе данных по её ID. - # @RELATION CALLS -> [APIClient] - def get_database(self, database_id: int) -> dict: + # @SIDE_EFFECT Асинхронный HTTP-вызов к Superset API. + # @RELATION CALLS -> [AsyncAPIClient] + async def get_database(self, database_id: int) -> dict: with belief_scope("get_database"): app_logger.info("[get_database][Enter] Fetching database %s.", database_id) - response = self.network.request( + response = await self.client.request( method="GET", endpoint=f"/database/{database_id}" ) response = cast(dict, response) @@ -46,23 +50,24 @@ class SupersetDatabasesMixin: # #endregion SupersetClientGetDatabase # #region SupersetClientGetDatabasesSummary [TYPE Function] [C:3] # @PURPOSE: Fetch a summary of databases including uuid, name, and engine. + # @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API. # @RELATION CALLS -> [SupersetClientGetDatabases] - def get_databases_summary(self) -> list[dict]: + async def get_databases_summary(self) -> list[dict]: with belief_scope("SupersetClient.get_databases_summary"): query = {"columns": ["id", "uuid", "database_name", "backend"]} - _, databases = self.get_databases(query=query) - # Map 'backend' to 'engine' for consistency with contracts + _, databases = await self.get_databases(query=query) for db in databases: db["engine"] = db.pop("backend", None) return databases # #endregion SupersetClientGetDatabasesSummary # #region SupersetClientGetDatabaseByUuid [TYPE Function] [C:3] # @PURPOSE: Find a database by its UUID. + # @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API. # @RELATION CALLS -> [SupersetClientGetDatabases] - def get_database_by_uuid(self, db_uuid: str) -> dict | None: + async def get_database_by_uuid(self, db_uuid: str) -> dict | None: with belief_scope("SupersetClient.get_database_by_uuid", f"uuid={db_uuid}"): query = {"filters": [{"col": "uuid", "op": "eq", "value": db_uuid}]} - _, databases = self.get_databases(query=query) + _, databases = await self.get_databases(query=query) return databases[0] if databases else None # #endregion SupersetClientGetDatabaseByUuid # #endregion SupersetDatabasesMixin diff --git a/backend/src/core/superset_client/_datasets.py b/backend/src/core/superset_client/_datasets.py index a618a42a..a15ac1cb 100644 --- a/backend/src/core/superset_client/_datasets.py +++ b/backend/src/core/superset_client/_datasets.py @@ -2,6 +2,7 @@ # @LAYER Infrastructure # @BRIEF Dataset domain mixin for SupersetClient — list, get, detail, update. # @RELATION DEPENDS_ON -> [SupersetClientBase] +# @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API для работы с датасетами. import json from typing import Any, cast @@ -11,15 +12,17 @@ app_logger = cast(Any, app_logger) # #region SupersetDatasetsMixin [C:3] [TYPE Class] # @BRIEF Mixin providing basic dataset CRUD operations (list, get, detail, update). # @RELATION DEPENDS_ON -> [SupersetClientBase] +# @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API. class SupersetDatasetsMixin: # #region SupersetClientGetDatasets [TYPE Function] [C:3] # @PURPOSE: Получает полный список датасетов, автоматически обрабатывая пагинацию. + # @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API для пагинации. # @RELATION CALLS -> [SupersetClientFetchAllPages] - def get_datasets(self, query: dict | None = None) -> tuple[int, list[dict]]: + async def get_datasets(self, query: dict | None = None) -> tuple[int, list[dict]]: with belief_scope("get_datasets"): app_logger.info("[get_datasets][Enter] Fetching datasets.") validated_query = self._validate_query_params(query) - paginated_data = self._fetch_all_pages( + paginated_data = await self._fetch_all_pages( endpoint="/dataset/", pagination_options={ "base_query": validated_query, @@ -32,11 +35,12 @@ class SupersetDatasetsMixin: # #endregion SupersetClientGetDatasets # #region SupersetClientGetDatasetsSummary [TYPE Function] [C:3] # @PURPOSE: Fetches dataset metadata optimized for the Dataset Hub grid. + # @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API. # @RELATION CALLS -> [SupersetClientGetDatasets] - def get_datasets_summary(self) -> list[dict]: + async def get_datasets_summary(self) -> list[dict]: with belief_scope("SupersetClient.get_datasets_summary"): query = {"columns": ["id", "table_name", "schema", "database"]} - _, datasets = self.get_datasets(query=query) + _, datasets = await self.get_datasets(query=query) result = [] for ds in datasets: result.append( @@ -53,15 +57,16 @@ class SupersetDatasetsMixin: # #endregion SupersetClientGetDatasetsSummary # #region SupersetClientGetDatasetLinkedDashboardCount [TYPE Function] [C:2] # @PURPOSE: Fetch the number of dashboards linked to a dataset via related_objects endpoint. - # @RELATION CALLS -> [APIClient] + # @SIDE_EFFECT Асинхронный HTTP-вызов к Superset API. + # @RELATION CALLS -> [AsyncAPIClient] # @RATIONALE Added to fix linked_count=0 in StatsBar. Reuses the same /dataset/{id}/related_objects call # that get_dataset_detail() was already making, but as a lightweight count-only call. # @REJECTED Enriching get_datasets_summary() with more columns rejected — Superset list endpoint # doesn't include linked_dashboard_count (it's a computed field from related_objects). - def get_dataset_linked_dashboard_count(self, dataset_id: int) -> int: + async def get_dataset_linked_dashboard_count(self, dataset_id: int) -> int: with belief_scope("get_dataset_linked_dashboard_count", f"id={dataset_id}"): try: - related_objects = self.network.request( + related_objects = await self.client.request( method="GET", endpoint=f"/dataset/{dataset_id}/related_objects" ) if isinstance(related_objects, dict): @@ -83,9 +88,10 @@ class SupersetDatasetsMixin: # #endregion SupersetClientGetDatasetLinkedDashboardCount # #region SupersetClientGetDatasetDetail [TYPE Function] [C:3] # @PURPOSE: Fetches detailed dataset information including columns and linked dashboards. + # @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API. # @RELATION CALLS -> [SupersetClientGetDataset] - # @RELATION CALLS -> [APIClient] - def get_dataset_detail(self, dataset_id: int) -> dict: + # @RELATION CALLS -> [AsyncAPIClient] + async def get_dataset_detail(self, dataset_id: int) -> dict: with belief_scope("SupersetClient.get_dataset_detail", f"id={dataset_id}"): def as_bool(value, default=False): if value is None: @@ -95,7 +101,7 @@ class SupersetDatasetsMixin: if isinstance(value, str): return value.strip().lower() in ("1", "true", "yes", "y", "on") return bool(value) - response = self.get_dataset(dataset_id) + response = await self.get_dataset(dataset_id) if isinstance(response, dict) and "result" in response: dataset = response["result"] else: @@ -134,7 +140,7 @@ class SupersetDatasetsMixin: ) linked_dashboards = [] try: - related_objects = self.network.request( + related_objects = await self.client.request( method="GET", endpoint=f"/dataset/{dataset_id}/related_objects" ) if isinstance(related_objects, dict): @@ -146,7 +152,6 @@ class SupersetDatasetsMixin: dashboards_data = related_objects["result"].get("dashboards", []) else: dashboards_data = [] - # related_objects returns {"count": N, "result": [...]} — extract the array dashboards_list = ( dashboards_data.get("result", []) if isinstance(dashboards_data, dict) @@ -208,11 +213,12 @@ class SupersetDatasetsMixin: # #endregion SupersetClientGetDatasetDetail # #region SupersetClientGetDataset [TYPE Function] [C:3] # @PURPOSE: Получает информацию о конкретном датасете по его ID. - # @RELATION CALLS -> [APIClient] - def get_dataset(self, dataset_id: int) -> dict: + # @SIDE_EFFECT Асинхронный HTTP-вызов к Superset API. + # @RELATION CALLS -> [AsyncAPIClient] + async def get_dataset(self, dataset_id: int) -> dict: with belief_scope("SupersetClient.get_dataset", f"id={dataset_id}"): app_logger.info("[get_dataset][Enter] Fetching dataset %s.", dataset_id) - response = self.network.request( + response = await self.client.request( method="GET", endpoint=f"/dataset/{dataset_id}" ) response = cast(dict, response) @@ -221,13 +227,13 @@ class SupersetDatasetsMixin: # #endregion SupersetClientGetDataset # #region SupersetClientUpdateDataset [TYPE Function] [C:3] # @PURPOSE: Обновляет данные датасета по его ID. - # @SIDE_EFFECT Modifies resource in upstream Superset environment. - # @RELATION CALLS -> [APIClient] - def update_dataset(self, dataset_id: int, data: dict, override_columns: bool = False) -> dict: + # @SIDE_EFFECT Асинхронный HTTP-вызов к Superset API. + # @RELATION CALLS -> [AsyncAPIClient] + async def update_dataset(self, dataset_id: int, data: dict, override_columns: bool = False) -> dict: with belief_scope("SupersetClient.update_dataset", f"id={dataset_id}"): app_logger.info("[update_dataset][Enter] Updating dataset %s.", dataset_id) endpoint = f"/dataset/{dataset_id}?override_columns={'true' if override_columns else 'false'}" - response = self.network.request( + response = await self.client.request( method="PUT", endpoint=endpoint, data=json.dumps(data), diff --git a/backend/src/core/superset_client/_datasets_preview.py b/backend/src/core/superset_client/_datasets_preview.py index 8db8e80c..1726da41 100644 --- a/backend/src/core/superset_client/_datasets_preview.py +++ b/backend/src/core/superset_client/_datasets_preview.py @@ -3,6 +3,7 @@ # @BRIEF Dataset preview compilation mixin for SupersetClient — build query context, compile SQL. # @RELATION DEPENDS_ON -> [SupersetClientBase] # @RELATION DEPENDS_ON -> [SupersetDatasetsMixin] +# @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API для превью датасетов. from copy import deepcopy import json from typing import Any @@ -13,13 +14,15 @@ from ..utils.network import SupersetAPIError # #region SupersetDatasetsPreviewMixin [C:4] [TYPE Class] # @BRIEF Mixin providing dataset preview compilation and query context building. +# @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API. class SupersetDatasetsPreviewMixin: # #region SupersetClientCompileDatasetPreview [TYPE Function] [C:4] # @PURPOSE: Compile dataset preview SQL through the strongest supported Superset preview endpoint family and return normalized SQL output. - def compile_dataset_preview(self, dataset_id: int, template_params: dict[str, Any] | None = None, effective_filters: list[dict[str, Any]] | None = None) -> dict[str, Any]: + # @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API для компиляции SQL превью. + async def compile_dataset_preview(self, dataset_id: int, template_params: dict[str, Any] | None = None, effective_filters: list[dict[str, Any]] | None = None) -> dict[str, Any]: with belief_scope('SupersetClientCompileDatasetPreview'): app_logger.reason('Belief protocol reasoning checkpoint for SupersetClientCompileDatasetPreview') - dataset_response = self.get_dataset(dataset_id) + dataset_response = await self.get_dataset(dataset_id) dataset_record = dataset_response.get('result', dataset_response) if isinstance(dataset_response, dict) else {} query_context = self.build_dataset_preview_query_context(dataset_id=dataset_id, dataset_record=dataset_record, template_params=template_params or {}, effective_filters=effective_filters or []) legacy_form_data = self.build_dataset_preview_legacy_form_data(dataset_id=dataset_id, dataset_record=dataset_record, template_params=template_params or {}, effective_filters=effective_filters or []) @@ -48,7 +51,7 @@ class SupersetDatasetsPreviewMixin: strategy_diagnostics = {'endpoint': endpoint_path, 'endpoint_kind': endpoint_kind, 'request_transport': request_transport, 'contains_root_datasource': endpoint_kind == 'v1_chart_data' and 'datasource' in query_context, 'contains_form_datasource': endpoint_kind.startswith('legacy_') and 'datasource' in legacy_form_data, 'contains_query_object_datasource': bool(query_context.get('queries')) and isinstance(query_context['queries'][0], dict) and ('datasource' in query_context['queries'][0]), 'request_param_keys': request_param_keys, 'request_payload_keys': request_payload_keys} app_logger.reason('Attempting Superset dataset preview compilation strategy', extra={'dataset_id': dataset_id, **strategy_diagnostics, 'request_params': request_params, 'request_payload': request_body, 'legacy_form_data': legacy_form_data if endpoint_kind.startswith('legacy_') else None, 'query_context': query_context if endpoint_kind == 'v1_chart_data' else None, 'template_param_count': len(template_params or {}), 'filter_count': len(effective_filters or [])}) try: - response = self.network.request(method='POST', endpoint=endpoint_path, params=request_params or None, data=request_body, headers=request_headers or None) + response = await self.client.request(method='POST', endpoint=endpoint_path, params=request_params or None, data=request_body, headers=request_headers or None) normalized = self._extract_compiled_sql_from_preview_response(response) normalized['query_context'] = query_context normalized['legacy_form_data'] = legacy_form_data diff --git a/backend/src/core/superset_client/_datasets_preview_filters.py b/backend/src/core/superset_client/_datasets_preview_filters.py index 779d1d93..50292a79 100644 --- a/backend/src/core/superset_client/_datasets_preview_filters.py +++ b/backend/src/core/superset_client/_datasets_preview_filters.py @@ -3,6 +3,7 @@ # @BRIEF Filter normalization and SQL extraction helpers for dataset preview compilation. # @RELATION DEPENDS_ON -> [SupersetClientBase] # @RELATION DEPENDS_ON -> [SupersetDatasetsPreviewMixin] +# @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API для превью датасетов. from copy import deepcopy from typing import Any @@ -14,6 +15,7 @@ from ..utils.network import SupersetAPIError # @BRIEF Mixin providing filter normalization and compiled-SQL extraction for dataset preview operations. # @RELATION DEPENDS_ON -> [SupersetClientBase] # @RELATION DEPENDS_ON -> [SupersetDatasetsPreviewMixin] +# @SIDE_EFFECT Асинхронные HTTP-вызовы к Superset API. class SupersetDatasetsPreviewFiltersMixin: # #region SupersetClientNormalizeEffectiveFiltersForQueryContext [TYPE Function] [C:3] # @PURPOSE: Convert execution mappings into Superset chart-data filter objects. diff --git a/backend/src/core/superset_profile_lookup.py b/backend/src/core/superset_profile_lookup.py index 2c168da9..ef7025a7 100644 --- a/backend/src/core/superset_profile_lookup.py +++ b/backend/src/core/superset_profile_lookup.py @@ -13,7 +13,8 @@ import json from typing import Any from .logger import belief_scope, logger -from .utils.network import APIClient, AuthenticationError, SupersetAPIError +from .utils.async_network import AsyncAPIClient +from .utils.network import AuthenticationError, SupersetAPIError # #region SupersetAccountLookupAdapter [C:3] [TYPE Class] @@ -22,19 +23,20 @@ from .utils.network import APIClient, AuthenticationError, SupersetAPIError # @RELATION DEPENDS_ON -> [SupersetProfileLookup] class SupersetAccountLookupAdapter: # #region __init__ [TYPE Function] - # @PURPOSE: Initializes lookup adapter with authenticated API client and environment context. - # @PRE network_client supports request(method, endpoint, params=...). - # @POST Adapter is ready to perform users lookup requests. - def __init__(self, network_client: APIClient, environment_id: str): + # @PURPOSE: Initializes lookup adapter with authenticated async API client and environment context. + # @PRE network_client supports async request(method, endpoint, params=...). + # @POST Adapter is ready to perform async users lookup requests. + def __init__(self, network_client: AsyncAPIClient, environment_id: str): self.network_client = network_client self.environment_id = str(environment_id or "") # #endregion __init__ # #region get_users_page [TYPE Function] - # @PURPOSE: Fetch one users page from Superset with passthrough search/sort parameters. + # @PURPOSE: Fetch one users page from Superset with passthrough search/sort parameters asynchronously. + # @SIDE_EFFECT Асинхронный HTTP-вызов к Superset API для поиска пользователей. # @PRE page_index >= 0 and page_size >= 1. # @POST Returns deterministic payload with normalized items and total count. # @RETURN Dict[str, Any] - def get_users_page( + async def get_users_page( self, search: str | None = None, page_index: int = 0, @@ -74,7 +76,7 @@ class SupersetAccountLookupAdapter: "Users lookup request attempt", extra={"src": "SupersetAccountLookupAdapter.get_users_page", "payload": {"env": self.environment_id, "attempt": attempt_index, "endpoint": endpoint}}, ) - response = self.network_client.request( + response = await self.network_client.request( method="GET", endpoint=endpoint, params={"q": json.dumps(query)}, diff --git a/backend/src/core/utils/async_network.py b/backend/src/core/utils/async_network.py index 6654ee4c..aa0dc3bc 100644 --- a/backend/src/core/utils/async_network.py +++ b/backend/src/core/utils/async_network.py @@ -9,6 +9,9 @@ # @RELATION DEPENDS_ON -> [SupersetAuthCache] # @INVARIANT Async client reuses cached auth tokens per environment credentials and invalidates on 401. import asyncio +import io +import json +from pathlib import Path from typing import Any import httpx @@ -295,6 +298,135 @@ class AsyncAPIClient: message = f"Unknown network error: {exc}" raise NetworkError(message, url=url) from exc # #endregion AsyncAPIClient._handle_network_error + # #region AsyncAPIClient.fetch_paginated_count [TYPE Function] [C:3] + # @PURPOSE: Fetch total count of items for a paginated endpoint. + # @POST Returns integer count from API response. + # @RELATION CALLS -> [AsyncAPIClient.request] + async def fetch_paginated_count( + self, + endpoint: str, + query_params: dict[str, Any], + count_field: str = "count", + ) -> int: + with belief_scope("AsyncAPIClient.fetch_paginated_count"): + response = await self.request( + method="GET", + endpoint=endpoint, + params={"q": json.dumps(query_params)}, + ) + return response.get(count_field, 0) if isinstance(response, dict) else 0 + # #endregion AsyncAPIClient.fetch_paginated_count + + # #region AsyncAPIClient.fetch_paginated_data [TYPE Function] [C:3] + # @PURPOSE: Fetch all pages of a paginated endpoint with automatic page iteration. + # @POST Returns combined list of all items across pages. + # @SIDE_EFFECT Performs multiple sequential GET requests to iterate pages. + # @RELATION CALLS -> [AsyncAPIClient.request] + async def fetch_paginated_data( + self, + endpoint: str, + pagination_options: dict[str, Any], + ) -> list[Any]: + with belief_scope("AsyncAPIClient.fetch_paginated_data"): + base_query = pagination_options["base_query"] + total_count = pagination_options.get("total_count") + results_field = pagination_options["results_field"] + count_field = pagination_options.get("count_field", "count") + page_size = base_query.get("page_size", 1000) + + results: list[Any] = [] + + # Fetch first page + query = {**base_query, "page": 0} + response_json = await self.request( + method="GET", + endpoint=endpoint, + params={"q": json.dumps(query)}, + ) + if isinstance(response_json, dict): + first_page_results = response_json.get(results_field, []) + results.extend(first_page_results) + if total_count is None: + total_count = response_json.get(count_field, len(first_page_results)) + else: + return results + + # Fetch remaining pages + total_pages = max(0, (total_count + page_size - 1) // page_size) + for page in range(1, total_pages): + query = {**base_query, "page": page} + page_response = await self.request( + method="GET", + endpoint=endpoint, + params={"q": json.dumps(query)}, + ) + if isinstance(page_response, dict): + results.extend(page_response.get(results_field, [])) + + return results + # #endregion AsyncAPIClient.fetch_paginated_data + + # #region AsyncAPIClient.upload_file [TYPE Function] [C:4] + # @PURPOSE: Upload a file via multipart/form-data POST with optional extra data fields. + # @PRE endpoint is valid relative Superset API endpoint. file_info contains file_obj and file_name. + # @POST Returns parsed JSON API response. + # @SIDE_EFFECT Performs multipart HTTP upload via httpx.AsyncClient. + # @RELATION CALLS -> [AsyncAPIClient.get_headers] + async def upload_file( + self, + endpoint: str, + file_info: dict[str, Any], + extra_data: dict[str, Any] | None = None, + timeout: int | None = None, + ) -> dict: + with belief_scope("AsyncAPIClient.upload_file"): + url = self._build_api_url(endpoint) + req_headers = await self.get_headers() + req_headers.pop("Content-Type", None) + + file_obj = file_info.get("file_obj") + file_name = file_info.get("file_name") + form_field = file_info.get("form_field", "file") + + if isinstance(file_obj, (str, Path)): + file_path = Path(file_obj) + if not file_path.exists(): + raise FileNotFoundError(f"Upload file does not exist: {file_obj}") + files = { + form_field: (file_name, file_path.read_bytes(), "application/x-zip-compressed"), + } + elif isinstance(file_obj, io.BytesIO): + files = { + form_field: (file_name, file_obj.getvalue(), "application/x-zip-compressed"), + } + else: + raise TypeError(f"Unsupported file_obj type: {type(file_obj)}") + + try: + response = await self._client.post( + url, + data=extra_data or {}, + files=files, + headers=req_headers, + timeout=httpx.Timeout(timeout or self.request_settings.get("timeout", self.DEFAULT_TIMEOUT)), + ) + response.raise_for_status() + if response.status_code == 200: + try: + return response.json() + except Exception as json_e: + raise SupersetAPIError( + f"API error during upload: Response is not valid JSON: {json_e}" + ) from json_e + return response.json() + except httpx.HTTPStatusError as exc: + raise SupersetAPIError( + f"API error during upload: {exc.response.text}" + ) from exc + except httpx.HTTPError as exc: + raise NetworkError(f"Network error during upload: {exc}", url=url) from exc + # #endregion AsyncAPIClient.upload_file + # #region AsyncAPIClient.aclose [TYPE Function] [C:3] # @PURPOSE: Close underlying httpx client. # @POST Client resources are released.