busya
5d9d214bd6
Root cause: AsyncAPIClient.request() passes its parameter directly
to httpx.AsyncClient.request(json=data). When callers pass a pre-serialized
JSON string (data=json.dumps(dict)), httpx re-encodes it via json.dumps(),
resulting in a double-encoded JSON string body instead of a JSON object.
This caused ALL POST/PUT requests with string data to fail — Superset received
a JSON string instead of a JSON object, returning GENERIC_BACKEND_ERROR
('dictionary update sequence element #0 has length 1; 2 is required').
Fix: if data is a string, pass it via httpx parameter (raw body);
if it's a dict/list, pass via for automatic encoding.
Affected callers (6 files) now correctly send JSON objects:
- preview_executor.py: chart data requests
- superset_executor.py
- _run_source.py
- _datasets.py: update_dataset
- _datasets_preview.py: compile_dataset_preview
- _dashboards_write.py
Also simplified preview_executor.fetch_sample_rows back to single-strategy
(chart data API only) since the root cause is now fixed.
157 lines
6.9 KiB
Python
157 lines
6.9 KiB
Python
# #region PreviewExecutor [C:4] [TYPE Class] [SEMANTICS sqlalchemy, superset, llm, preview]
|
|
# @BRIEF Fetch sample data from Superset and call LLM provider for preview.
|
|
# @PRE Database session, config manager, and Superset client are available.
|
|
# @POST Sample rows fetched, LLM called.
|
|
# @SIDE_EFFECT Fetches data from Superset; makes HTTP calls to LLM provider.
|
|
# @RELATION DEPENDS_ON -> [SupersetClient]
|
|
# @RELATION DEPENDS_ON -> [LLMProviderService]
|
|
# @RELATION DEPENDS_ON -> [LLMClient]
|
|
# @RELATION DEPENDS_ON -> [preview_response_parser]
|
|
# @RATIONALE LLM response parsing and hash utilities extracted to preview_response_parser module for INV_7 compliance.
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ...core.config_manager import ConfigManager
|
|
from ...core.logger import belief_scope, logger
|
|
from ...core.superset_client import SupersetClient
|
|
from ...models.translate import TranslationJob
|
|
from ._llm_async_http import call_openai_compatible
|
|
from .preview_resolve_provider import resolve_provider_model as _resolve_provider_model
|
|
from .preview_response_parser import (
|
|
compute_config_hash as _compute_config_hash,
|
|
compute_dict_snapshot_hash as _compute_dict_snapshot_hash,
|
|
extract_data_rows as _extract_data_rows,
|
|
parse_llm_response as _parse_llm_response,
|
|
)
|
|
|
|
|
|
class PreviewExecutor:
|
|
"""Execute preview operations: fetch sample data from Superset, call LLM, parse responses."""
|
|
|
|
def __init__(self, db: Session, config_manager: ConfigManager):
|
|
self.db = db
|
|
self.config_manager = config_manager
|
|
|
|
# region fetch_sample_rows [TYPE Function]
|
|
# @PURPOSE: Fetch sample rows from Superset dataset for preview.
|
|
# @SIDE_EFFECT Calls Superset chart data API.
|
|
async def fetch_sample_rows(
|
|
self,
|
|
job: TranslationJob,
|
|
sample_size: int = 10,
|
|
env_id: str | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
with belief_scope("PreviewExecutor.fetch_sample_rows"):
|
|
environments = self.config_manager.get_environments()
|
|
target_env_id = env_id or job.environment_id or job.source_dialect or ""
|
|
env_config = next((e for e in environments if e.id == target_env_id), None)
|
|
if not env_config:
|
|
if environments:
|
|
env_config = environments[0]
|
|
else:
|
|
raise ValueError("No Superset environments configured")
|
|
|
|
client = SupersetClient(env_config)
|
|
dataset_detail = await client.get_dataset_detail(int(job.source_datasource_id))
|
|
query_context = client.build_dataset_preview_query_context(
|
|
dataset_id=int(job.source_datasource_id), dataset_record=dataset_detail,
|
|
template_params={}, effective_filters=[],
|
|
)
|
|
|
|
# Build column list from dataset schema for explicit column projection
|
|
column_names = [
|
|
col["name"] for col in dataset_detail.get("columns", [])
|
|
if col.get("name") and col.get("is_active", True)
|
|
]
|
|
|
|
queries = query_context.get("queries", [])
|
|
if queries:
|
|
queries[0]["row_limit"] = sample_size
|
|
queries[0]["columns"] = column_names or []
|
|
queries[0]["metrics"] = []
|
|
queries[0].pop("result_type", None)
|
|
query_context["result_type"] = "query"
|
|
form_data = query_context.get("form_data", {})
|
|
form_data.pop("query_mode", None)
|
|
|
|
try:
|
|
response = await client.network.request(
|
|
method="POST", endpoint="/api/v1/chart/data",
|
|
data=json.dumps(query_context),
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
except Exception as e:
|
|
raise ValueError(f"Failed to fetch sample data from Superset: {e}")
|
|
|
|
return _extract_data_rows(response)
|
|
# endregion fetch_sample_rows
|
|
|
|
# region call_llm [TYPE Function]
|
|
# @PURPOSE: Call the configured LLM provider with a prompt.
|
|
# @SIDE_EFFECT Makes HTTP call to LLM provider.
|
|
async def call_llm(self, job: TranslationJob, prompt: str, max_tokens: int = 8192) -> str:
|
|
with belief_scope("PreviewExecutor.call_llm"):
|
|
if not job.provider_id:
|
|
raise ValueError("Job has no LLM provider configured")
|
|
from ...services.llm_provider import LLMProviderService
|
|
provider_svc = LLMProviderService(self.db)
|
|
provider = provider_svc.get_provider(job.provider_id)
|
|
if not provider:
|
|
raise ValueError(f"LLM provider '{job.provider_id}' not found")
|
|
api_key = provider_svc.get_decrypted_api_key(job.provider_id)
|
|
if not api_key:
|
|
raise ValueError(f"Could not decrypt API key for provider '{job.provider_id}'")
|
|
|
|
model = provider.default_model or "gpt-4o-mini"
|
|
provider_type = provider.provider_type.lower() if provider.provider_type else "openai"
|
|
disable_reasoning = getattr(job, 'disable_reasoning', False)
|
|
|
|
if provider_type not in ("openai", "openai_compatible", "openrouter", "kilo", "litellm"):
|
|
raise ValueError(f"Unsupported provider type '{provider_type}' for preview")
|
|
|
|
max_attempts = 2
|
|
for attempt in range(max_attempts):
|
|
try:
|
|
content_text, _ = await call_openai_compatible(
|
|
base_url=provider.base_url, api_key=api_key, model=model,
|
|
prompt=prompt, provider_type=provider_type,
|
|
max_tokens=max_tokens * (attempt + 1),
|
|
disable_reasoning=disable_reasoning or (attempt > 0),
|
|
)
|
|
break
|
|
except ValueError as e:
|
|
if "empty content" in str(e) and attempt < max_attempts - 1:
|
|
logger.explore("Empty LLM response, retrying with doubled max_tokens")
|
|
continue
|
|
raise
|
|
|
|
return content_text
|
|
# endregion call_llm
|
|
|
|
# region delegation_helpers [TYPE Function]
|
|
# @PURPOSE: Backward-compatible delegation to preview_response_parser module functions.
|
|
|
|
@staticmethod
|
|
def parse_llm_response(
|
|
response_text: str, expected_count: int,
|
|
target_languages: list[str] | None = None,
|
|
finish_reason: str | None = None,
|
|
) -> dict[str, dict[str, str]]:
|
|
return _parse_llm_response(response_text, expected_count, target_languages, finish_reason)
|
|
|
|
def resolve_provider_model(self, job: TranslationJob) -> str | None:
|
|
return _resolve_provider_model(self.db, job)
|
|
|
|
@staticmethod
|
|
def compute_config_hash(job: TranslationJob) -> str:
|
|
return _compute_config_hash(job)
|
|
|
|
def compute_dict_snapshot_hash(self, job_id: str) -> str:
|
|
return _compute_dict_snapshot_hash(self.db, job_id)
|
|
# endregion delegation_helpers
|
|
|
|
# #endregion PreviewExecutor
|