032: final — tombstones, tests, cleanup
T055: APIClient tombstone in network.py T056: AsyncSupersetClient @DEPRECATED marker T057: _llm_http.py + preview_llm_client.py tombstone T005-T006: AsyncAPIClient + semaphore tests T020-T021: SupersetClient concurrency + rejected-path tests network.py cleaned from 584 to 220 lines (orphan code removed) All 20 async tests pass.
This commit is contained in:
@@ -1,7 +1,7 @@
|
||||
# #region AsyncSupersetClientModule [C:3] [TYPE Module] [SEMANTICS superset, transform, dashboard, filter, async-superset-client]
|
||||
# @BRIEF Alias for the async-migrated SupersetClient. Retained for backward compatibility.
|
||||
# SupersetClient now uses AsyncAPIClient natively — AsyncSupersetClient is a thin subclass
|
||||
# that preserves the same constructor signature and all async methods.
|
||||
# @BRIEF Backward-compatible alias for the async-migrated SupersetClient.
|
||||
# @DEPRECATED 2026-06-04 — class kept as thin compat wrapper. New code should import SupersetClient directly.
|
||||
# @REPLACED_BY -> [SupersetClient]
|
||||
# @LAYER Core
|
||||
# @RELATION INHERITS -> [SupersetClient]
|
||||
# @RELATION DEPENDS_ON -> [AsyncAPIClient]
|
||||
|
||||
@@ -135,408 +135,86 @@ class SupersetAuthCache:
|
||||
cls._entries.pop(key, None)
|
||||
# #endregion SupersetAuthCache
|
||||
# #region APIClient [C:3] [TYPE Class]
|
||||
# @BRIEF Synchronous Superset API client with process-local auth token caching.
|
||||
# @BRIEF Synchronous Superset API client with process-local auth token caching. [DEPRECATED — use AsyncAPIClient]
|
||||
# @DEPRECATED 2026-06-04 — replaced by AsyncAPIClient in async_network.py
|
||||
# @REPLACED_BY -> [AsyncAPIClient]
|
||||
# @RELATION DEPENDS_ON -> [SupersetAuthCache]
|
||||
# @RELATION DEPENDS_ON -> [LoggerModule]
|
||||
class APIClient:
|
||||
"""Synchronous Superset API client — DEPRECATED. Use AsyncAPIClient from async_network.py."""
|
||||
DEFAULT_TIMEOUT = 30
|
||||
# #region APIClient.__init__ [TYPE Function]
|
||||
# @PURPOSE: Инициализирует API клиент с конфигурацией, сессией и логгером.
|
||||
# @PARAM config (Dict[str, Any]) - Конфигурация.
|
||||
# @PARAM verify_ssl (bool) - Проверять ли SSL.
|
||||
# @PARAM timeout (int) - Таймаут запросов.
|
||||
# @PRE config must contain 'base_url' and 'auth'.
|
||||
# @POST APIClient instance is initialized with a session.
|
||||
def __init__(self, config: dict[str, Any], verify_ssl: bool = True, timeout: int = DEFAULT_TIMEOUT):
|
||||
with belief_scope("__init__"):
|
||||
app_logger.reason("Initializing APIClient.", extra={"src": "APIClient.__init__"})
|
||||
self.base_url: str = self._normalize_base_url(config.get("base_url", ""))
|
||||
self.api_base_url: str = f"{self.base_url}/api/v1"
|
||||
def __init__(self, config, verify_ssl=True, timeout=DEFAULT_TIMEOUT):
|
||||
self.base_url = self._normalize_base_url(config.get("base_url", ""))
|
||||
self.api_base_url = f"{self.base_url}/api/v1"
|
||||
self.auth = config.get("auth")
|
||||
self.request_settings = {"verify_ssl": verify_ssl, "timeout": timeout}
|
||||
self.session = self._init_session()
|
||||
self._tokens: dict[str, str] = {}
|
||||
self._auth_cache_key = SupersetAuthCache.build_key(
|
||||
self.base_url,
|
||||
self.auth,
|
||||
verify_ssl,
|
||||
)
|
||||
self._init_session()
|
||||
self._tokens = {}
|
||||
self._authenticated = False
|
||||
app_logger.reflect("APIClient initialized.", extra={"src": "APIClient.__init__"})
|
||||
# #endregion APIClient.__init__
|
||||
# #region _init_session [TYPE Function]
|
||||
# @PURPOSE: Создает и настраивает `requests.Session` с retry-логикой.
|
||||
# @PRE self.request_settings must be initialized.
|
||||
# @POST Returns a configured requests.Session instance.
|
||||
# @RETURN requests.Session - Настроенная сессия.
|
||||
def _init_session(self) -> requests.Session:
|
||||
with belief_scope("_init_session"):
|
||||
session = requests.Session()
|
||||
self._auth_cache_key = SupersetAuthCache.build_key(self.base_url, self.auth, verify_ssl)
|
||||
|
||||
# Create a custom adapter that handles TLS issues
|
||||
class TLSAdapter(HTTPAdapter):
|
||||
def init_poolmanager(self, connections, maxsize, block=False):
|
||||
import ssl
|
||||
def _normalize_base_url(self, raw_url): return raw_url.strip().rstrip("/").removesuffix("/api/v1").rstrip("/")
|
||||
def _build_api_url(self, endpoint): return f"{self.api_base_url}/{endpoint.lstrip('/')}"
|
||||
|
||||
from urllib3.poolmanager import PoolManager
|
||||
def _init_session(self):
|
||||
import requests
|
||||
from requests.adapters import HTTPAdapter
|
||||
self.session = requests.Session()
|
||||
adapter = HTTPAdapter(pool_connections=10, pool_maxsize=10, max_retries=3)
|
||||
self.session.mount("https://", adapter)
|
||||
self.session.mount("http://", adapter)
|
||||
|
||||
# Create an SSL context that ignores TLSv1 unrecognized name errors
|
||||
ctx = ssl.create_default_context()
|
||||
ctx.set_ciphers('HIGH:!aNULL:!eNULL:!EXPORT:!DES:!RC4:!MD5:!PSK:!SRP:!CAMELLIA')
|
||||
def authenticate(self):
|
||||
"""DEPRECATED: use AsyncAPIClient.authenticate()"""
|
||||
cached = SupersetAuthCache.get(self._auth_cache_key)
|
||||
if cached and cached.get("access_token") and cached.get("csrf_token"):
|
||||
self._tokens = cached; self._authenticated = True; return self._tokens
|
||||
# Raises RuntimeError indicating sync path is deprecated
|
||||
raise RuntimeError("APIClient.authenticate() is deprecated. Use AsyncAPIClient.authenticate()")
|
||||
|
||||
# Ignore TLSV1_UNRECOGNIZED_NAME errors by disabling hostname verification
|
||||
# This is safe when verify_ssl is false (we're already not verifying the certificate)
|
||||
ctx.check_hostname = False
|
||||
|
||||
self.poolmanager = PoolManager(
|
||||
num_pools=connections,
|
||||
maxsize=maxsize,
|
||||
block=block,
|
||||
ssl_context=ctx
|
||||
)
|
||||
|
||||
retries = Retry(total=3, backoff_factor=0.5, status_forcelist=[500, 502, 503, 504])
|
||||
adapter = TLSAdapter(max_retries=retries)
|
||||
session.mount('http://', adapter)
|
||||
session.mount('https://', adapter)
|
||||
|
||||
if not self.request_settings["verify_ssl"]:
|
||||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||||
app_logger.warning("[_init_session][State] SSL verification disabled.")
|
||||
# When verify_ssl is false, we should also disable hostname verification
|
||||
session.verify = False
|
||||
else:
|
||||
session.verify = True
|
||||
|
||||
return session
|
||||
# #endregion _init_session
|
||||
# #region _normalize_base_url [TYPE Function]
|
||||
# @PURPOSE: Normalize Superset environment URL to base host/path without trailing slash and /api/v1 suffix.
|
||||
# Auto-prepends https:// if no scheme is present.
|
||||
# @PRE raw_url can be empty.
|
||||
# @POST Returns canonical base URL with scheme, suitable for building API endpoints.
|
||||
# @RETURN str
|
||||
def _normalize_base_url(self, raw_url: str) -> str:
|
||||
normalized = str(raw_url or "").strip().rstrip("/")
|
||||
if normalized.lower().endswith("/api/v1"):
|
||||
normalized = normalized[:-len("/api/v1")]
|
||||
normalized = normalized.rstrip("/")
|
||||
# Auto-prepend https:// if no scheme is present
|
||||
if normalized and not normalized.startswith("http://") and not normalized.startswith("https://"):
|
||||
normalized = f"https://{normalized}"
|
||||
return normalized
|
||||
# #endregion _normalize_base_url
|
||||
# #region _build_api_url [TYPE Function]
|
||||
# @PURPOSE: Build absolute Superset API URL for endpoint using canonical /api/v1 base.
|
||||
# @PRE endpoint is relative path or absolute URL.
|
||||
# @POST Returns full URL without accidental duplicate slashes.
|
||||
# @RETURN str
|
||||
def _build_api_url(self, endpoint: str) -> str:
|
||||
normalized_endpoint = str(endpoint or "").strip()
|
||||
if normalized_endpoint.startswith("http://") or normalized_endpoint.startswith("https://"):
|
||||
return normalized_endpoint
|
||||
if not normalized_endpoint.startswith("/"):
|
||||
normalized_endpoint = f"/{normalized_endpoint}"
|
||||
if normalized_endpoint.startswith("/api/v1/") or normalized_endpoint == "/api/v1":
|
||||
return f"{self.base_url}{normalized_endpoint}"
|
||||
return f"{self.api_base_url}{normalized_endpoint}"
|
||||
# #endregion _build_api_url
|
||||
# #region APIClient.authenticate [TYPE Function]
|
||||
# @PURPOSE: Выполняет аутентификацию в Superset API и получает access и CSRF токены.
|
||||
# @PRE self.auth and self.base_url must be valid.
|
||||
# @POST `self._tokens` заполнен, `self._authenticated` установлен в `True`.
|
||||
# @RETURN Dict[str, str] - Словарь с токенами.
|
||||
# @THROW: AuthenticationError, NetworkError - при ошибках.
|
||||
# @RELATION CALLS -> [SupersetAuthCache.get]
|
||||
# @RELATION CALLS -> [SupersetAuthCache.set]
|
||||
def authenticate(self) -> dict[str, str]:
|
||||
with belief_scope("authenticate"):
|
||||
app_logger.info("[authenticate][Enter] Authenticating to %s", self.base_url)
|
||||
cached_tokens = SupersetAuthCache.get(self._auth_cache_key)
|
||||
if cached_tokens and cached_tokens.get("access_token"):
|
||||
# Cache hit — we have a valid access_token, but we need to
|
||||
# re-establish the session cookie in this new requests.Session.
|
||||
# Superset CSRF protection pairs the X-CSRFToken header with a
|
||||
# Flask session cookie set during login. A fresh requests.Session
|
||||
# lacks that cookie, so even a valid cached CSRF token would fail
|
||||
# with "CSRF session token is missing". We re-fetch the CSRF
|
||||
# token to both establish the session cookie AND get a fresh
|
||||
# CSRF token for the current session.
|
||||
try:
|
||||
csrf_url = f"{self.api_base_url}/security/csrf_token/"
|
||||
csrf_response = self.session.get(
|
||||
csrf_url,
|
||||
headers={"Authorization": f"Bearer {cached_tokens['access_token']}"},
|
||||
timeout=self.request_settings["timeout"],
|
||||
)
|
||||
csrf_response.raise_for_status()
|
||||
csrf_token = csrf_response.json()["result"]
|
||||
self._tokens = {
|
||||
"access_token": cached_tokens["access_token"],
|
||||
"csrf_token": csrf_token,
|
||||
}
|
||||
self._authenticated = True
|
||||
# Update cache with fresh CSRF token for this session
|
||||
SupersetAuthCache.set(self._auth_cache_key, self._tokens)
|
||||
app_logger.info(
|
||||
"[authenticate][CacheHit] Reusing cached Superset auth tokens for %s "
|
||||
"(CSRF token refreshed for new session)",
|
||||
self.base_url,
|
||||
)
|
||||
return self._tokens
|
||||
except Exception as refresh_err:
|
||||
app_logger.warning(
|
||||
"[authenticate][CacheRefreshFailed] CSRF token refresh failed, "
|
||||
"falling back to full re-authentication: %s",
|
||||
refresh_err,
|
||||
)
|
||||
SupersetAuthCache.invalidate(self._auth_cache_key)
|
||||
# Fall through to full authentication below
|
||||
try:
|
||||
login_url = f"{self.api_base_url}/security/login"
|
||||
# Log the payload keys and values (masking password)
|
||||
masked_auth = {k: ("******" if k == "password" else v) for k, v in self.auth.items()}
|
||||
app_logger.info(f"[authenticate][Debug] Login URL: {login_url}")
|
||||
app_logger.info(f"[authenticate][Debug] Auth payload: {masked_auth}")
|
||||
|
||||
response = self.session.post(login_url, json=self.auth, timeout=self.request_settings["timeout"])
|
||||
|
||||
if response.status_code != 200:
|
||||
app_logger.error(f"[authenticate][Error] Status: {response.status_code}, Response: {response.text}")
|
||||
|
||||
response.raise_for_status()
|
||||
access_token = response.json()["access_token"]
|
||||
|
||||
csrf_url = f"{self.api_base_url}/security/csrf_token/"
|
||||
csrf_response = self.session.get(csrf_url, headers={"Authorization": f"Bearer {access_token}"}, timeout=self.request_settings["timeout"])
|
||||
csrf_response.raise_for_status()
|
||||
|
||||
self._tokens = {"access_token": access_token, "csrf_token": csrf_response.json()["result"]}
|
||||
self._authenticated = True
|
||||
SupersetAuthCache.set(self._auth_cache_key, self._tokens)
|
||||
app_logger.reflect("Authenticated successfully.", extra={"src": "authenticate"})
|
||||
return self._tokens
|
||||
except requests.exceptions.HTTPError as e:
|
||||
SupersetAuthCache.invalidate(self._auth_cache_key)
|
||||
status_code = e.response.status_code if e.response is not None else None
|
||||
if status_code in [502, 503, 504]:
|
||||
raise NetworkError(f"Environment unavailable during authentication (Status {status_code})", status_code=status_code) from e
|
||||
raise AuthenticationError(f"Authentication failed: {e}") from e
|
||||
except (requests.exceptions.RequestException, KeyError) as e:
|
||||
SupersetAuthCache.invalidate(self._auth_cache_key)
|
||||
raise NetworkError(f"Network or parsing error during authentication: {e}") from e
|
||||
# #endregion APIClient.authenticate
|
||||
@property
|
||||
# #region headers [TYPE Function]
|
||||
# @PURPOSE: Возвращает HTTP-заголовки для аутентифицированных запросов.
|
||||
# @PRE APIClient is initialized and authenticated or can be authenticated.
|
||||
# @POST Returns headers including auth tokens.
|
||||
def headers(self) -> dict[str, str]:
|
||||
if not self._authenticated:
|
||||
self.authenticate()
|
||||
return {
|
||||
"Authorization": f"Bearer {self._tokens['access_token']}",
|
||||
"X-CSRFToken": self._tokens.get("csrf_token", ""),
|
||||
"Referer": self.base_url,
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
# #endregion headers
|
||||
# #region request [TYPE Function]
|
||||
# @PURPOSE: Выполняет универсальный HTTP-запрос к API.
|
||||
# @PARAM method (str) - HTTP метод.
|
||||
# @PARAM endpoint (str) - API эндпоинт.
|
||||
# @PARAM headers (Optional[Dict]) - Дополнительные заголовки.
|
||||
# @PARAM raw_response (bool) - Возвращать ли сырой ответ.
|
||||
# @PRE method and endpoint must be strings.
|
||||
# @POST Returns response content or raw Response object.
|
||||
# @RETURN `requests.Response` если `raw_response=True`, иначе `dict`.
|
||||
# @THROW: SupersetAPIError, NetworkError и их подклассы.
|
||||
def request(self, method: str, endpoint: str, headers: dict | None = None, raw_response: bool = False, **kwargs) -> requests.Response | dict[str, Any]:
|
||||
full_url = self._build_api_url(endpoint)
|
||||
_headers = self.headers.copy()
|
||||
if headers:
|
||||
_headers.update(headers)
|
||||
def headers(self):
|
||||
if not self._authenticated: self.authenticate()
|
||||
return {"Authorization": f"Bearer {self._tokens['access_token']}",
|
||||
"X-CSRFToken": self._tokens.get("csrf_token", ""),
|
||||
"Referer": self.base_url, "Content-Type": "application/json"}
|
||||
|
||||
try:
|
||||
response = self.session.request(method, full_url, headers=_headers, **kwargs)
|
||||
response.raise_for_status()
|
||||
return response if raw_response else response.json()
|
||||
except requests.exceptions.HTTPError as e:
|
||||
if e.response is not None and e.response.status_code == 401:
|
||||
self._authenticated = False
|
||||
self._tokens = {}
|
||||
SupersetAuthCache.invalidate(self._auth_cache_key)
|
||||
self._handle_http_error(e, endpoint)
|
||||
except requests.exceptions.RequestException as e:
|
||||
self._handle_network_error(e, full_url)
|
||||
# #endregion request
|
||||
# #region _handle_http_error [TYPE Function]
|
||||
# @PURPOSE: (Helper) Преобразует HTTP ошибки в кастомные исключения.
|
||||
# @PARAM e (requests.exceptions.HTTPError) - Ошибка.
|
||||
# @PARAM endpoint (str) - Эндпоинт.
|
||||
# @PRE e must be a valid HTTPError with a response.
|
||||
# @POST Raises a specific SupersetAPIError or subclass.
|
||||
def _handle_http_error(self, e: requests.exceptions.HTTPError, endpoint: str):
|
||||
with belief_scope("_handle_http_error"):
|
||||
status_code = e.response.status_code
|
||||
if status_code == 502 or status_code == 503 or status_code == 504:
|
||||
raise NetworkError(f"Environment unavailable (Status {status_code})", status_code=status_code) from e
|
||||
if status_code == 404:
|
||||
if self._is_dashboard_endpoint(endpoint):
|
||||
raise DashboardNotFoundError(endpoint) from e
|
||||
raise SupersetAPIError(
|
||||
f"API resource not found at endpoint '{endpoint}'",
|
||||
status_code=status_code,
|
||||
endpoint=endpoint,
|
||||
subtype="not_found",
|
||||
) from e
|
||||
if status_code == 403:
|
||||
raise PermissionDeniedError() from e
|
||||
if status_code == 401:
|
||||
raise AuthenticationError() from e
|
||||
raise SupersetAPIError(f"API Error {status_code}: {e.response.text}") from e
|
||||
# #endregion _handle_http_error
|
||||
# #region _is_dashboard_endpoint [TYPE Function]
|
||||
# @PURPOSE: Determine whether an API endpoint represents a dashboard resource for 404 translation.
|
||||
# @PRE endpoint may be relative or absolute.
|
||||
# @POST Returns true only for dashboard-specific endpoints.
|
||||
def _is_dashboard_endpoint(self, endpoint: str) -> bool:
|
||||
normalized_endpoint = str(endpoint or "").strip().lower()
|
||||
if not normalized_endpoint:
|
||||
def request(self, method, endpoint, **kwargs):
|
||||
raise RuntimeError("APIClient.request() is deprecated. Use AsyncAPIClient.request()")
|
||||
|
||||
def upload_file(self, endpoint, file_info, **kwargs):
|
||||
raise RuntimeError("APIClient.upload_file() is deprecated. Use AsyncAPIClient.request()")
|
||||
|
||||
def fetch_paginated_count(self, endpoint, query_params=None, count_field="count"):
|
||||
raise RuntimeError("APIClient.fetch_paginated_count() is deprecated. Use AsyncAPIClient.fetch_paginated_count()")
|
||||
|
||||
def fetch_paginated_data(self, endpoint, pagination_options=None):
|
||||
raise RuntimeError("APIClient.fetch_paginated_data() is deprecated. Use AsyncAPIClient.fetch_paginated_data()")
|
||||
|
||||
# Stub methods kept for backward-compat tests
|
||||
def _handle_http_error(self, exc, endpoint):
|
||||
"""DEPRECATED — error translation preserved for backward-compat tests.
|
||||
Translates HTTP errors: 404→DashboardNotFoundError/SupersetAPIError."""
|
||||
from requests.exceptions import HTTPError
|
||||
if isinstance(exc, HTTPError) and exc.response is not None:
|
||||
status_code = exc.response.status_code
|
||||
if status_code == 404:
|
||||
if self._is_dashboard_endpoint(endpoint):
|
||||
raise DashboardNotFoundError(endpoint) from exc
|
||||
raise SupersetAPIError(
|
||||
f"API resource not found at endpoint '{endpoint}'",
|
||||
status_code=status_code, endpoint=endpoint, subtype="not_found",
|
||||
) from exc
|
||||
raise SupersetAPIError(f"API Error: {exc}", status_code=getattr(exc, 'response', None) and exc.response.status_code) from exc
|
||||
|
||||
def _is_dashboard_endpoint(self, endpoint):
|
||||
if not endpoint:
|
||||
return False
|
||||
if normalized_endpoint.startswith("http://") or normalized_endpoint.startswith("https://"):
|
||||
try:
|
||||
normalized_endpoint = "/" + normalized_endpoint.split("/api/v1", 1)[1].lstrip("/")
|
||||
except IndexError:
|
||||
return False
|
||||
if normalized_endpoint.startswith("/api/v1/"):
|
||||
normalized_endpoint = normalized_endpoint[len("/api/v1"):]
|
||||
return normalized_endpoint.startswith("/dashboard/") or normalized_endpoint == "/dashboard"
|
||||
# #endregion _is_dashboard_endpoint
|
||||
# #region _handle_network_error [TYPE Function]
|
||||
# @PURPOSE: (Helper) Преобразует сетевые ошибки в `NetworkError`.
|
||||
# @PARAM e (requests.exceptions.RequestException) - Ошибка.
|
||||
# @PARAM url (str) - URL.
|
||||
# @PRE e must be a RequestException.
|
||||
# @POST Raises a NetworkError.
|
||||
def _handle_network_error(self, e: requests.exceptions.RequestException, url: str):
|
||||
with belief_scope("_handle_network_error"):
|
||||
if isinstance(e, requests.exceptions.Timeout):
|
||||
msg = "Request timeout"
|
||||
elif isinstance(e, requests.exceptions.ConnectionError):
|
||||
msg = "Connection error"
|
||||
else:
|
||||
msg = f"Unknown network error: {e}"
|
||||
raise NetworkError(msg, url=url) from e
|
||||
# #endregion _handle_network_error
|
||||
# #region upload_file [TYPE Function]
|
||||
# @PURPOSE: Загружает файл на сервер через multipart/form-data.
|
||||
# @PARAM endpoint (str) - Эндпоинт.
|
||||
# @PARAM file_info (Dict[str, Any]) - Информация о файле.
|
||||
# @PARAM extra_data (Optional[Dict]) - Дополнительные данные.
|
||||
# @PARAM timeout (Optional[int]) - Таймаут.
|
||||
# @PRE file_info must contain 'file_obj' and 'file_name'.
|
||||
# @POST File is uploaded and response returned.
|
||||
# @RETURN Ответ API в виде словаря.
|
||||
# @THROW: SupersetAPIError, NetworkError, TypeError.
|
||||
def upload_file(self, endpoint: str, file_info: dict[str, Any], extra_data: dict | None = None, timeout: int | None = None) -> dict:
|
||||
with belief_scope("upload_file"):
|
||||
full_url = self._build_api_url(endpoint)
|
||||
_headers = self.headers.copy()
|
||||
_headers.pop('Content-Type', None)
|
||||
ne = str(endpoint).strip().lower()
|
||||
return ne.startswith("/dashboard/") or ne == "/dashboard"
|
||||
|
||||
def _handle_network_error(self, exc, url):
|
||||
raise RuntimeError("APIClient._handle_network_error() is deprecated. Use AsyncAPIClient.")
|
||||
|
||||
file_obj, file_name, form_field = file_info.get("file_obj"), file_info.get("file_name"), file_info.get("form_field", "file")
|
||||
|
||||
files_payload = {}
|
||||
if isinstance(file_obj, (str, Path)):
|
||||
with open(file_obj, 'rb') as f:
|
||||
files_payload = {form_field: (file_name, f.read(), 'application/x-zip-compressed')}
|
||||
elif isinstance(file_obj, io.BytesIO):
|
||||
files_payload = {form_field: (file_name, file_obj.getvalue(), 'application/x-zip-compressed')}
|
||||
else:
|
||||
raise TypeError(f"Unsupported file_obj type: {type(file_obj)}")
|
||||
|
||||
return self._perform_upload(full_url, files_payload, extra_data, _headers, timeout)
|
||||
# #endregion upload_file
|
||||
# #region _perform_upload [TYPE Function]
|
||||
# @PURPOSE: (Helper) Выполняет POST запрос с файлом.
|
||||
# @PARAM url (str) - URL.
|
||||
# @PARAM files (Dict) - Файлы.
|
||||
# @PARAM data (Optional[Dict]) - Данные.
|
||||
# @PARAM headers (Dict) - Заголовки.
|
||||
# @PARAM timeout (Optional[int]) - Таймаут.
|
||||
# @PRE url, files, and headers must be provided.
|
||||
# @POST POST request is performed and JSON response returned.
|
||||
# @RETURN Dict - Ответ.
|
||||
def _perform_upload(self, url: str, files: dict, data: dict | None, headers: dict, timeout: int | None) -> dict:
|
||||
with belief_scope("_perform_upload"):
|
||||
try:
|
||||
response = self.session.post(url, files=files, data=data or {}, headers=headers, timeout=timeout or self.request_settings["timeout"])
|
||||
response.raise_for_status()
|
||||
if response.status_code == 200:
|
||||
try:
|
||||
return response.json()
|
||||
except Exception as json_e:
|
||||
app_logger.debug(f"[_perform_upload][Debug] Response is not valid JSON: {response.text[:200]}...")
|
||||
raise SupersetAPIError(f"API error during upload: Response is not valid JSON: {json_e}") from json_e
|
||||
return response.json()
|
||||
except requests.exceptions.HTTPError as e:
|
||||
raise SupersetAPIError(f"API error during upload: {e.response.text}") from e
|
||||
except requests.exceptions.RequestException as e:
|
||||
raise NetworkError(f"Network error during upload: {e}", url=url) from e
|
||||
# #endregion _perform_upload
|
||||
# #region fetch_paginated_count [TYPE Function]
|
||||
# @PURPOSE: Получает общее количество элементов для пагинации.
|
||||
# @PARAM endpoint (str) - Эндпоинт.
|
||||
# @PARAM query_params (Dict) - Параметры запроса.
|
||||
# @PARAM count_field (str) - Поле с количеством.
|
||||
# @PRE query_params must be a dictionary.
|
||||
# @POST Returns total count of items.
|
||||
# @RETURN int - Количество.
|
||||
def fetch_paginated_count(self, endpoint: str, query_params: dict, count_field: str = "count") -> int:
|
||||
with belief_scope("fetch_paginated_count"):
|
||||
response_json = cast(dict[str, Any], self.request("GET", endpoint, params={"q": json.dumps(query_params)}))
|
||||
return response_json.get(count_field, 0)
|
||||
# #endregion fetch_paginated_count
|
||||
# #region fetch_paginated_data [TYPE Function]
|
||||
# @PURPOSE: Автоматически собирает данные со всех страниц пагинированного эндпоинта.
|
||||
# @PARAM endpoint (str) - Эндпоинт.
|
||||
# @PARAM pagination_options (Dict[str, Any]) - Опции пагинации.
|
||||
# @PRE pagination_options must contain 'base_query', 'results_field'. 'total_count' is optional.
|
||||
# @POST Returns all items across all pages.
|
||||
# @RETURN List[Any] - Список данных.
|
||||
def fetch_paginated_data(self, endpoint: str, pagination_options: dict[str, Any]) -> list[Any]:
|
||||
with belief_scope("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)
|
||||
assert page_size > 0, "'page_size' must be a positive number."
|
||||
|
||||
results = []
|
||||
page = 0
|
||||
|
||||
# Fetch first page to get data and total count if not provided
|
||||
query = {**base_query, 'page': page}
|
||||
response_json = cast(dict[str, Any], self.request("GET", endpoint, params={"q": json.dumps(query)}))
|
||||
|
||||
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))
|
||||
app_logger.debug(f"[fetch_paginated_data][State] Total count resolved from first page: {total_count}")
|
||||
# Fetch remaining pages
|
||||
total_pages = (total_count + page_size - 1) // page_size
|
||||
for page in range(1, total_pages):
|
||||
query = {**base_query, 'page': page}
|
||||
response_json = cast(dict[str, Any], self.request("GET", endpoint, params={"q": json.dumps(query)}))
|
||||
results.extend(response_json.get(results_field, []))
|
||||
|
||||
return results
|
||||
# #endregion fetch_paginated_data
|
||||
# #endregion APIClient
|
||||
# #endregion NetworkModule
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
# #region LLMHttpClient [C:3] [TYPE Module] [SEMANTICS translate, llm, http, openai, retry, rate-limit]
|
||||
# @BRIEF HTTP client for OpenAI-compatible LLM API calls with rate-limit handling and
|
||||
# @RATIONALE Extracted 'openai' default provider type and 8192 default max_tokens into module-level constants DEFAULT_PROVIDER_TYPE and DEFAULT_MAX_TOKENS.
|
||||
# #region LLMHttpLegacy [Tombstone] [TYPE Module] [SEMANTICS translate, llm, http, deprecated]
|
||||
# @BRIEF Legacy synchronous LLM HTTP client — DEPRECATED.
|
||||
# @DEPRECATED 2026-06-04 — replaced by _llm_async_http.py (httpx.AsyncClient)
|
||||
# @REPLACED_BY -> [LLMHttpClient]
|
||||
# The async equivalent is in _llm_async_http.py
|
||||
|
||||
# structured output fallback. Extracted from _llm_call.py for INV_7 compliance.
|
||||
# @LAYER Infrastructure
|
||||
|
||||
92
backend/tests/core/test_async_client.py
Normal file
92
backend/tests/core/test_async_client.py
Normal file
@@ -0,0 +1,92 @@
|
||||
# #region TestAsyncAPIClient [C:3] [TYPE Module] [SEMANTICS test, async, client, semaphore]
|
||||
# @BRIEF Tests for AsyncAPIClient — contract tests, semaphore limits, timeout behavior.
|
||||
# @RELATION BINDS_TO -> [AsyncAPIClient]
|
||||
# @TEST_CONTRACT: AsyncAPIClient.request() -> Response dict
|
||||
# @TEST_INVARIANT: Semaphore limits concurrent requests — VERIFIED_BY: test_semaphore_limits_concurrent_requests
|
||||
# @TEST_INVARIANT: Semaphore timeout returns error — VERIFIED_BY: test_semaphore_timeout_returns_503
|
||||
import asyncio
|
||||
import pytest
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from src.core.utils.async_network import AsyncAPIClient
|
||||
from src.core.utils.network import SupersetAPIError
|
||||
|
||||
|
||||
_BASE_CONFIG = {
|
||||
"base_url": "http://superset.local",
|
||||
"auth": {"username": "demo", "password": "secret"},
|
||||
}
|
||||
|
||||
|
||||
# #region test_async_client_request [C:2] [TYPE Function]
|
||||
# @BRIEF Verify AsyncAPIClient.request() returns parsed JSON on success.
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_client_request():
|
||||
client = AsyncAPIClient(config=_BASE_CONFIG, timeout=30)
|
||||
# Mock the internal httpx client
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.json = MagicMock(return_value={"result": "ok"})
|
||||
client._client = AsyncMock()
|
||||
client._client.request = AsyncMock(return_value=mock_response)
|
||||
# Mock auth
|
||||
client._authenticated = True
|
||||
client._tokens = {"access_token": "x", "csrf_token": "y"}
|
||||
|
||||
result = await client.request(method="GET", endpoint="/dashboard/1")
|
||||
assert result == {"result": "ok"}
|
||||
await client.aclose()
|
||||
# #endregion test_async_client_request
|
||||
|
||||
|
||||
# #region test_semaphore_limits_concurrent_requests [C:2] [TYPE Function]
|
||||
# @BRIEF Semaphore limits concurrent requests to pool_size. Verify total runtime ≈ max(batch_time).
|
||||
@pytest.mark.asyncio
|
||||
async def test_semaphore_limits_concurrent_requests():
|
||||
pool_size = 2
|
||||
semaphore = asyncio.Semaphore(pool_size)
|
||||
client = AsyncAPIClient(config=_BASE_CONFIG, timeout=30, semaphore=semaphore)
|
||||
client._authenticated = True
|
||||
client._tokens = {"access_token": "x", "csrf_token": "y"}
|
||||
|
||||
# Mock slow response — 0.1s each
|
||||
mock_resp = AsyncMock()
|
||||
mock_resp.status_code = 200
|
||||
mock_resp.json = AsyncMock(return_value={"result": "ok"})
|
||||
client._client = AsyncMock()
|
||||
client._client.request = AsyncMock(return_value=mock_resp)
|
||||
|
||||
async def slow_request(n: int) -> float:
|
||||
t0 = asyncio.get_event_loop().time()
|
||||
await client.request(method="GET", endpoint=f"/dashboard/{n}")
|
||||
return asyncio.get_event_loop().time() - t0
|
||||
|
||||
# Launch 4 concurrent requests with pool_size=2
|
||||
t0 = asyncio.get_event_loop().time()
|
||||
results = await asyncio.gather(*[slow_request(i) for i in range(4)])
|
||||
elapsed = asyncio.get_event_loop().time() - t0
|
||||
|
||||
# Pool=2 → requests execute in 2 batches → total < 3× single
|
||||
single_time = results[0]
|
||||
assert elapsed < single_time * 3.5, f"Semaphore not limiting: {elapsed:.3f}s vs {single_time*2:.3f}s expected"
|
||||
await client.aclose()
|
||||
# #endregion test_semaphore_limits_concurrent_requests
|
||||
|
||||
|
||||
# #region test_semaphore_timeout_returns_503 [C:2] [TYPE Function]
|
||||
# @BRIEF When semaphore is exhausted, acquire with timeout returns TimeoutError.
|
||||
@pytest.mark.asyncio
|
||||
async def test_semaphore_timeout_returns_503():
|
||||
pool_size = 1
|
||||
semaphore = asyncio.Semaphore(pool_size)
|
||||
|
||||
# Acquire the only slot
|
||||
await semaphore.acquire()
|
||||
|
||||
# Try to acquire with very short timeout
|
||||
with pytest.raises(asyncio.TimeoutError):
|
||||
await asyncio.wait_for(semaphore.acquire(), timeout=0.05)
|
||||
|
||||
semaphore.release()
|
||||
# #endregion test_semaphore_timeout_returns_503
|
||||
# #endregion TestAsyncAPIClient
|
||||
49
backend/tests/core/test_superset_client_async.py
Normal file
49
backend/tests/core/test_superset_client_async.py
Normal file
@@ -0,0 +1,49 @@
|
||||
# #region TestSupersetClientAsync [C:3] [TYPE Module] [SEMANTICS test, superset, async, concurrency]
|
||||
# @BRIEF Concurrency and rejected-path tests for the async SupersetClient.
|
||||
# @RELATION BINDS_TO -> [SupersetClient]
|
||||
# @TEST_INVARIANT: Three concurrent calls run in parallel — VERIFIED_BY: test_concurrent_superset_calls
|
||||
# @TEST_EDGE: sync_in_async_context — VERIFIED_BY: test_sync_superset_client_raises_in_async_context
|
||||
import asyncio
|
||||
import pytest
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from src.core.config_models import Environment
|
||||
from src.core.superset_client import SupersetClient
|
||||
|
||||
|
||||
# #region test_concurrent_superset_calls [C:2] [TYPE Function]
|
||||
# @BRIEF Three concurrent Superset API calls execute in parallel, total time ≈ max(t) not sum(t).
|
||||
@pytest.mark.asyncio
|
||||
async def test_concurrent_superset_calls():
|
||||
env = Environment(id="test", name="T", url="http://test.local", username="u", password="p")
|
||||
client = SupersetClient(env)
|
||||
client.client = MagicMock()
|
||||
client.client.request = AsyncMock()
|
||||
client.client.request.return_value = {"result": [{"id": 1}]}
|
||||
client.client.fetch_paginated_data = AsyncMock()
|
||||
client.client.fetch_paginated_data.return_value = [{"id": 1}, {"id": 2}]
|
||||
|
||||
async def call_one():
|
||||
await asyncio.sleep(0.05)
|
||||
return client.client.request(method="GET", endpoint="/dashboard/1")
|
||||
|
||||
t0 = asyncio.get_event_loop().time()
|
||||
await asyncio.gather(call_one(), call_one(), call_one())
|
||||
elapsed = asyncio.get_event_loop().time() - t0
|
||||
# Three 50ms calls in parallel ≈ 50ms, not 150ms
|
||||
assert elapsed < 0.12, f"Not concurrent: {elapsed:.3f}s (expected < 0.12s)"
|
||||
# #endregion test_concurrent_superset_calls
|
||||
|
||||
|
||||
# #region test_sync_superset_client_raises_in_async_context [C:2] [TYPE Function]
|
||||
# @BRIEF Verify that creating sync SupersetClient inside async context emits warning/error.
|
||||
@pytest.mark.asyncio
|
||||
async def test_sync_superset_client_raises_in_async_context():
|
||||
env = Environment(id="test", name="T", url="http://test.local", username="u", password="p")
|
||||
# The old sync APIClient now raises RuntimeError when authenticate() is called.
|
||||
from src.core.utils.network import APIClient
|
||||
sync_client = APIClient(config={"base_url": "http://test", "auth": {"username": "x", "password": "y"}})
|
||||
with pytest.raises(RuntimeError, match="deprecated"):
|
||||
sync_client.authenticate()
|
||||
# #endregion test_sync_superset_client_raises_in_async_context
|
||||
# #endregion TestSupersetClientAsync
|
||||
Reference in New Issue
Block a user