From 5eb116509bc16ab98344eb8638dde3bf2c02e2b1 Mon Sep 17 00:00:00 2001 From: busya Date: Fri, 5 Jun 2026 08:31:18 +0300 Subject: [PATCH] =?UTF-8?q?032:=20dead=20code=20cleanup=20=E2=80=94=20remo?= =?UTF-8?q?ve=20sync=20APIClient,=20=5Fllm=5Fhttp,=20preview=5Fllm=5Fclien?= =?UTF-8?q?t,=20fix=20retry=20chains?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/api/routes/translate/_run_routes.py | 8 +- backend/src/core/utils/network.py | 83 ------- backend/src/plugins/translate/_llm_http.py | 223 ------------------ backend/src/plugins/translate/orchestrator.py | 10 +- .../plugins/translate/orchestrator_cancel.py | 4 +- .../plugins/translate/orchestrator_retry.py | 8 +- .../plugins/translate/orchestrator_runner.py | 8 +- .../plugins/translate/preview_llm_client.py | 125 ---------- .../tests/core/test_superset_client_async.py | 11 - 9 files changed, 20 insertions(+), 460 deletions(-) delete mode 100644 backend/src/plugins/translate/_llm_http.py delete mode 100644 backend/src/plugins/translate/preview_llm_client.py diff --git a/backend/src/api/routes/translate/_run_routes.py b/backend/src/api/routes/translate/_run_routes.py index 3adf6a5a..db1e720b 100644 --- a/backend/src/api/routes/translate/_run_routes.py +++ b/backend/src/api/routes/translate/_run_routes.py @@ -132,7 +132,7 @@ def run_translation( # @PRE User has translate.job.execute permission. # @POST Returns the updated translation run. @router.post("/runs/{run_id}/retry") -def retry_run( +async def retry_run( run_id: str, current_user: User = Depends(get_current_user), _ = Depends(has_permission("translate.job", "EXECUTE")), @@ -143,7 +143,7 @@ def retry_run( logger.reason(f"retry_run — Run: {run_id}, User: {current_user.username}", extra={"src": "translate_routes"}) try: orch = TranslationOrchestrator(db, config_manager, current_user.username) - run = orch.retry_failed_batches(run_id) + run = await orch.retry_failed_batches(run_id) return _run_to_response(run) except ValueError as e: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) @@ -158,7 +158,7 @@ def retry_run( # @PRE User has translate.job.execute permission. # @POST Returns the updated run. @router.post("/runs/{run_id}/retry-insert") -def retry_insert( +async def retry_insert( run_id: str, current_user: User = Depends(get_current_user), _ = Depends(has_permission("translate.job", "EXECUTE")), @@ -169,7 +169,7 @@ def retry_insert( logger.reason(f"retry_insert — Run: {run_id}, User: {current_user.username}", extra={"src": "translate_routes"}) try: orch = TranslationOrchestrator(db, config_manager, current_user.username) - run = orch.retry_insert(run_id) + run = await orch.retry_insert(run_id) return _run_to_response(run) except ValueError as e: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) diff --git a/backend/src/core/utils/network.py b/backend/src/core/utils/network.py index 7592cfaa..a00a812e 100644 --- a/backend/src/core/utils/network.py +++ b/backend/src/core/utils/network.py @@ -134,87 +134,4 @@ class SupersetAuthCache: with cls._lock: cls._entries.pop(key, None) # #endregion SupersetAuthCache -# #region APIClient [C:3] [TYPE Class] -# @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 - 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._init_session() - self._tokens = {} - self._authenticated = False - self._auth_cache_key = SupersetAuthCache.build_key(self.base_url, self.auth, verify_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('/')}" - - 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) - - 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()") - - @property - 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"} - - 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 - 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.") - -# #endregion APIClient # #endregion NetworkModule diff --git a/backend/src/plugins/translate/_llm_http.py b/backend/src/plugins/translate/_llm_http.py deleted file mode 100644 index b5e982d6..00000000 --- a/backend/src/plugins/translate/_llm_http.py +++ /dev/null @@ -1,223 +0,0 @@ -# #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 -# @RELATION DEPENDS_ON -> [EXT:requests] -# @PRE Valid API endpoint, key, model, and prompt. -# @POST Returns (response text, finish_reason) tuple. -# @SIDE_EFFECT HTTP POST to LLM API with optional retry on 429. -# @RATIONALE Extracted from LLMTranslationService (793 lines) to keep module under INV_7 limit. -# @REJECTED Single HTTP client class — kept as module-level functions for stateless reusability. - -import os -import time -from typing import Any - -from ...core.logger import logger - - -# #region _get_verify [C:1] [TYPE Function] [SEMANTICS translate, ssl, verify] -# @BRIEF Resolve SSL verification path from LLM_SSL_VERIFY env var. -# @RATIONALE Используем capath=/etc/ssl/certs/ вместо cafile, потому что -# OpenSSL 3.x не использует intermediate CA сертификаты из cafile для -# построения цепочки (verify code 20). capath с хеш-симлинками работает -# корректно (verify code 0). -# @REJECTED cafile отвергнут — OpenSSL 3.x не использует intermediate CA -# из единого bundle-файла. Только capath с хеш-симлинками даёт code 0. -# @POST Returns path to /etc/ssl/certs/ when enabled, False when disabled. -def _get_verify() -> str | bool: - raw = os.getenv("LLM_SSL_VERIFY", "true").strip().lower() - if raw in ("false", "0", "no", "off"): - return False - return "/etc/ssl/certs/" -# #endregion _get_verify - - -# #region call_openai_compatible [C:3] [TYPE Function] [SEMANTICS translate, llm, http, openai] -# @BRIEF Call OpenAI-compatible API with rate-limit handling and structured output fallback. -# @PRE Valid API endpoint, key, model, and prompt. -# @POST Returns (response text, finish_reason). -# @SIDE_EFFECT HTTP POST to LLM API. -def call_openai_compatible( - base_url: str, - api_key: str, - model: str, - prompt: str, - provider_type: str = "openai", - max_tokens: int = 8192, - disable_reasoning: bool = False, -) -> tuple[str, str | None]: - """Call OpenAI-compatible API for batch translation.""" - if not base_url: - raise ValueError("LLM provider has no base_url configured") - - - url = f"{base_url.rstrip('/')}/chat/completions" - headers = { - "Authorization": f"Bearer {api_key}", - "Content-Type": "application/json", - } - system_content = ( - "You are a database content translation assistant. " - "Translate the provided text accurately, preserving data semantics. " - "Respond directly with ONLY the JSON result. " - "Do NOT include any reasoning, thinking, chain-of-thought, analysis, " - "or explanation. Output ONLY valid JSON." - ) - - payload: dict[str, Any] = { - "model": model, - "messages": [ - {"role": "system", "content": system_content}, - {"role": "user", "content": prompt}, - ], - "temperature": 0.1, - "max_tokens": max_tokens, - } - - if provider_type in ("openai", "openai_compatible", "kilo", "openrouter", "litellm"): - if not disable_reasoning: - payload["response_format"] = {"type": "json_object"} - - if disable_reasoning: - if provider_type not in ("kilo", "openrouter", "litellm"): - payload["reasoning_effort"] = "none" - payload["max_tokens"] = max_tokens - - logger.reason( - f"LLM request url={base_url} model={payload.get('model')} " - f"provider_type={provider_type} " - f"response_format={'yes' if 'response_format' in payload else 'no'} " - f"prompt_len={len(prompt)}" - ) - - response, response_text = _do_http_request(url, headers, payload) - _handle_response_format_fallback(response, response_text, payload, url, headers) - - if not response.ok: - logger.explore( - f"LLM API error status={response.status_code} " - f"model={payload.get('model')} " - f"body={response_text[:2000]}" - ) - response.raise_for_status() - data = response.json() - - choices = data.get("choices", []) - if not choices: - logger.explore("LLM returned no choices", extra={ - "src": "executor", "response_keys": list(data.keys()), - "response_preview": str(data)[:2000], - }) - raise ValueError("LLM returned no choices") - - try: - finish_reason = choices[0].get("finish_reason") or "none" - msg = choices[0].get("message") or {} - except (TypeError, AttributeError) as e: - logger.explore("TypeError processing LLM response choices", extra={ - "src": "executor_diag", "error": str(e), - "choices_0_type": type(choices[0]).__name__ if choices else "N/A", - "choices_0_repr": repr(choices[0])[:2000] if choices else "N/A", - "data_type": type(data).__name__, "data_preview": str(data)[:2000], - }) - raise ValueError(f"LLM response processing failed: {e}") - - # Log provider token usage for batch sizing calibration - usage = data.get("usage") or {} - if usage: - logger.reason( - "LLM provider usage", - { - "prompt_tokens": usage.get("prompt_tokens"), - "completion_tokens": usage.get("completion_tokens"), - "total_tokens": usage.get("total_tokens"), - "finish_reason": finish_reason, - "max_tokens_sent": max_tokens, - "chars_sent": len(prompt), - }, - ) - - refusal = msg.get("refusal") if isinstance(msg, dict) else None - if refusal: - logger.explore("LLM refused to respond", extra={ - "src": "executor", "refusal": str(refusal)[:500], "finish_reason": finish_reason, - }) - raise ValueError(f"LLM refused to respond: {refusal}") - - content = msg.get("content") if isinstance(msg, dict) else "" - if not content and isinstance(msg, dict): - content = msg.get("content") or "" - - logger.reason( - f"LLM response finish_reason={finish_reason} content_len={len(content)} " - f"msg_keys={list(msg.keys()) if isinstance(msg, dict) else []}" - ) - if not content: - logger.explore("LLM returned empty content", extra={ - "src": "executor", "finish_reason": finish_reason, - "msg_keys": list(msg.keys()) if isinstance(msg, dict) else [], - "response_preview": str(data)[:2000], - }) - raise ValueError("LLM returned empty content") - - return content, finish_reason -# #endregion call_openai_compatible - - -# #region _do_http_request [C:1] [TYPE Function] -def _do_http_request(url: str, headers: dict, payload: dict) -> tuple[Any, str]: - """Make HTTP POST with rate-limit (429) retry handling.""" - import requests as http_requests - _max_retry_429 = 3 - _retry_count_429 = 0 - while _retry_count_429 < _max_retry_429: - response = http_requests.post(url, headers=headers, json=payload, timeout=180, verify=_get_verify()) - response_text = response.text - if response.status_code == 429: - _retry_count_429 += 1 - retry_after = response.headers.get("Retry-After") - if retry_after: - try: - wait = int(retry_after) - except (ValueError, TypeError): - wait = 2 ** _retry_count_429 - else: - wait = 2 ** _retry_count_429 - logger.explore(f"Rate limited (429), retry {_retry_count_429}/{_max_retry_429} after {wait}s", - extra={"src": "executor", "retry_after": retry_after, "wait": wait}) - time.sleep(wait) - if _retry_count_429 >= _max_retry_429: - break - else: - break - return response, response_text -# #endregion _do_http_request - - -# #region _handle_response_format_fallback [C:1] [TYPE Function] -def _handle_response_format_fallback( - response: Any, response_text: str, payload: dict, url: str, headers: dict, -) -> None: - """Handle 400 errors from structured_outputs not being supported.""" - import requests as http_requests - _patterns = ("response_format", "structured_outputs", "structured", "json_object") - if ( - not response.ok - and response.status_code == 400 - and any(p in (response_text or "").lower() for p in _patterns) - ): - logger.explore("Structured outputs not supported, retrying without response_format", - extra={"src": "executor"}) - payload.pop("response_format", None) - new_response = http_requests.post(url, headers=headers, json=payload, timeout=180, verify=_get_verify()) - response.status_code = new_response.status_code - response._content = new_response.content - response.encoding = new_response.encoding - response.headers = new_response.headers -# #endregion _handle_response_format_fallback -# #endregion LLMHttpClient diff --git a/backend/src/plugins/translate/orchestrator.py b/backend/src/plugins/translate/orchestrator.py index 39e08693..feb03cf9 100644 --- a/backend/src/plugins/translate/orchestrator.py +++ b/backend/src/plugins/translate/orchestrator.py @@ -87,6 +87,8 @@ class TranslationOrchestrator: async def execute_run( self, run: TranslationRun, + on_batch_progress: Callable[[str, int, int, int, int], None] | None = None, + skip_insert: bool = False, ) -> TranslationRun: with belief_scope("TranslationOrchestrator.execute_run"): return await self._runner.execute_run( @@ -98,14 +100,14 @@ class TranslationOrchestrator: # region retry_failed_batches [TYPE Function] # @PURPOSE: Retry failed batches in a run. - def retry_failed_batches(self, run_id: str) -> TranslationRun: - return self._runner.retry_failed_batches(run_id) + async def retry_failed_batches(self, run_id: str) -> TranslationRun: + return await self._runner.retry_failed_batches(run_id) # endregion retry_failed_batches # region retry_insert [TYPE Function] # @PURPOSE: Retry the SQL insert phase for a completed run. - def retry_insert(self, run_id: str) -> TranslationRun: - return self._runner.retry_insert(run_id) + async def retry_insert(self, run_id: str) -> TranslationRun: + return await self._runner.retry_insert(run_id) # endregion retry_insert # region cancel_run [TYPE Function] diff --git a/backend/src/plugins/translate/orchestrator_cancel.py b/backend/src/plugins/translate/orchestrator_cancel.py index 8c42d7f2..231fcce3 100644 --- a/backend/src/plugins/translate/orchestrator_cancel.py +++ b/backend/src/plugins/translate/orchestrator_cancel.py @@ -86,7 +86,7 @@ def cancel_run( # @RELATION DEPENDS_ON -> [TranslationJob] # @RELATION DEPENDS_ON -> [TranslationEventLog] # @RELATION DEPENDS_ON -> [SQLInsertService] -def retry_insert( +async def retry_insert( db: Session, config_manager: ConfigManager, event_log: TranslationEventLog, @@ -107,7 +107,7 @@ def retry_insert( payload={}, created_by=current_user, ) sql_service = SQLInsertService(db, config_manager, event_log) - insert_result = sql_service.generate_and_insert_sql(job, run) + insert_result = await sql_service.generate_and_insert_sql(job, run) run.insert_status = insert_result.get("status") run.superset_execution_id = str(insert_result.get("query_id") or "") if insert_result.get("error_message"): diff --git a/backend/src/plugins/translate/orchestrator_retry.py b/backend/src/plugins/translate/orchestrator_retry.py index abef7544..43172582 100644 --- a/backend/src/plugins/translate/orchestrator_retry.py +++ b/backend/src/plugins/translate/orchestrator_retry.py @@ -45,7 +45,7 @@ class TranslationRunRetryManager: # region retry_failed_batches [TYPE Function] # @PURPOSE: Retry failed batches in a run. # @SIDE_EFFECT Re-executes batch translations; DB writes. - def retry_failed_batches(self, run_id: str) -> TranslationRun: + async def retry_failed_batches(self, run_id: str) -> TranslationRun: with belief_scope("TranslationRunRetryManager.retry_failed_batches"): run = self.db.query(TranslationRun).filter(TranslationRun.id == run_id).first() if not run: @@ -95,7 +95,7 @@ class TranslationRunRetryManager: } for rec in failed_records ] - result = executor._process_batch( + result = await executor._process_batch( job=job, run_id=run_id, batch_index=batch.batch_index, batch_rows=retry_rows, ) @@ -122,8 +122,8 @@ class TranslationRunRetryManager: # region retry_insert [TYPE Function] # @PURPOSE: Retry the SQL insert phase for a completed run. # @SIDE_EFFECT Superset API call; DB writes. - def retry_insert(self, run_id: str) -> TranslationRun: - return _retry_insert(self.db, self.config_manager, self.event_log, self.current_user, run_id) + async def retry_insert(self, run_id: str) -> TranslationRun: + return await _retry_insert(self.db, self.config_manager, self.event_log, self.current_user, run_id) # endregion retry_insert # region cancel_run [TYPE Function] diff --git a/backend/src/plugins/translate/orchestrator_runner.py b/backend/src/plugins/translate/orchestrator_runner.py index f7fd2d28..a0c611b4 100644 --- a/backend/src/plugins/translate/orchestrator_runner.py +++ b/backend/src/plugins/translate/orchestrator_runner.py @@ -57,14 +57,14 @@ class TranslationStageRunner: # region retry_failed_batches [TYPE Function] # @PURPOSE: Retry failed batches in a run. - def retry_failed_batches(self, run_id: str) -> TranslationRun: - return self._retry_mgr.retry_failed_batches(run_id) + async def retry_failed_batches(self, run_id: str) -> TranslationRun: + return await self._retry_mgr.retry_failed_batches(run_id) # endregion retry_failed_batches # region retry_insert [TYPE Function] # @PURPOSE: Retry the SQL insert phase for a completed run. - def retry_insert(self, run_id: str) -> TranslationRun: - return self._retry_mgr.retry_insert(run_id) + async def retry_insert(self, run_id: str) -> TranslationRun: + return await self._retry_mgr.retry_insert(run_id) # endregion retry_insert # region cancel_run [TYPE Function] diff --git a/backend/src/plugins/translate/preview_llm_client.py b/backend/src/plugins/translate/preview_llm_client.py deleted file mode 100644 index 9a145149..00000000 --- a/backend/src/plugins/translate/preview_llm_client.py +++ /dev/null @@ -1,125 +0,0 @@ -# #region LLMClient [C:3] [TYPE Class] [SEMANTICS llm, openai, api, retry] -# @BRIEF Call OpenAI-compatible LLM APIs with retry logic for rate limiting and structured output fallback. -# @SIDE_EFFECT Makes HTTP POST calls to external LLM API. -# @RELATION DEPENDS_ON -> [EXT:requests] - -import os -import time as _time -from typing import Any - -from ...core.logger import logger - - -# #region _get_verify [C:1] [TYPE Function] [SEMANTICS translate, ssl, verify] -# @BRIEF Resolve SSL verification path from LLM_SSL_VERIFY env var. -# @RATIONALE Используем capath=/etc/ssl/certs/ вместо cafile, потому что -# OpenSSL 3.x не использует intermediate CA сертификаты из cafile для -# построения цепочки (verify code 20). capath с хеш-симлинками работает -# корректно (verify code 0). -# @REJECTED cafile отвергнут — OpenSSL 3.x не использует intermediate CA -# из единого bundle-файла. Только capath с хеш-симлинками даёт code 0. -# @POST Returns path to /etc/ssl/certs/ when enabled, False when disabled. -def _get_verify() -> str | bool: - raw = os.getenv("LLM_SSL_VERIFY", "true").strip().lower() - if raw in ("false", "0", "no", "off"): - return False - return "/etc/ssl/certs/" -# #endregion _get_verify - - -class LLMClient: - """Call OpenAI-compatible LLM APIs with retry and structured output handling.""" - - @staticmethod - def call_openai_compatible( - base_url: str, - api_key: str, - model: str, - prompt: str, - provider_type: str = "openai", - max_tokens: int = 8192, - disable_reasoning: bool = False, - ) -> str: - """Call an OpenAI-compatible API for translation.""" - if not base_url: - raise ValueError("LLM provider has no base_url configured") - import requests as http_requests - - url = f"{base_url.rstrip('/')}/chat/completions" - headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} - - system_content = ( - "You are a database content translation assistant. " - "Translate the provided text accurately, preserving data semantics. " - "Respond directly with ONLY the JSON result. " - "Do NOT include any reasoning, thinking, chain-of-thought, analysis, " - "or explanation. Output ONLY valid JSON." - ) - - payload: dict[str, Any] = { - "model": model, - "messages": [ - {"role": "system", "content": system_content}, - {"role": "user", "content": prompt}, - ], - "temperature": 0.1, - "max_tokens": max_tokens, - } - - if provider_type in ("openai", "openai_compatible", "kilo", "openrouter", "litellm"): - if not disable_reasoning: - payload["response_format"] = {"type": "json_object"} - - if disable_reasoning: - if provider_type not in ("kilo", "openrouter", "litellm"): - payload["reasoning_effort"] = "none" - payload["max_tokens"] = max_tokens - - logger.reason(f"LLM request url={base_url} model={payload.get('model')} " - f"provider_type={provider_type} prompt_len={len(prompt)}") - - _max_retry_429 = 3 - _retry_count_429 = 0 - while _retry_count_429 < _max_retry_429: - response = http_requests.post(url, headers=headers, json=payload, timeout=600, verify=_get_verify()) - if response.status_code == 429: - _retry_count_429 += 1 - retry_after = response.headers.get("Retry-After") - wait = int(retry_after) if retry_after and retry_after.isdigit() else 2 ** _retry_count_429 - logger.explore(f"Rate limited (429), retry {_retry_count_429}/{_max_retry_429} after {wait}s") - _time.sleep(wait) - if _retry_count_429 >= _max_retry_429: - break - else: - break - - _response_format_error_patterns = ("response_format", "structured_outputs", "structured", "json_object") - if (not response.ok and response.status_code == 400 - and any(p in (response.text or "").lower() for p in _response_format_error_patterns)): - logger.explore("Structured outputs not supported, retrying without response_format") - payload.pop("response_format", None) - response = http_requests.post(url, headers=headers, json=payload, timeout=600, verify=_get_verify()) - - if not response.ok: - logger.explore(f"LLM API error status={response.status_code} model={payload.get('model')} body={response.text[:2000]}") - - response.raise_for_status() - data = response.json() - choices = data.get("choices", []) - if not choices: - raise ValueError("LLM returned no choices") - - try: - msg = choices[0].get("message") or {} - refusal = msg.get("refusal") - if refusal: - raise ValueError(f"LLM refused to respond: {refusal}") - content = msg.get("content") or "" - except TypeError as e: - raise ValueError(f"LLM response processing failed: {e}") - - if not content: - raise ValueError("LLM returned empty content") - - return content -# #endregion LLMClient diff --git a/backend/tests/core/test_superset_client_async.py b/backend/tests/core/test_superset_client_async.py index 9b5b0c61..938634ca 100644 --- a/backend/tests/core/test_superset_client_async.py +++ b/backend/tests/core/test_superset_client_async.py @@ -35,15 +35,4 @@ async def test_concurrent_superset_calls(): # #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