from __future__ import annotations from dataclasses import dataclass from threading import Event from time import sleep from typing import Any from urllib.parse import urljoin, urlsplit, urlunsplit from uuid import uuid4 import httpx from .config import DEFAULT_USER_AGENT, get_default_qf_version, get_timeout_seconds, normalize_base_url from .errors import QingflowApiError from .json_types import JSONObject, JSONScalar, JSONValue @dataclass(slots=True) class BackendRequestContext: base_url: str token: str ws_id: int | None qf_request_id: str | None = None qf_version: str | None = None qf_version_source: str | None = None @dataclass(slots=True) class BackendResponse: data: JSONValue headers: dict[str, str] request_id: str http_status: int qf_response_version: str | None = None @dataclass(slots=True) class BackendBinaryResponse: content: bytes headers: dict[str, str] http_status: int final_url: str redirected: bool class BackendClient: def __init__(self, timeout: float | None = None, client: httpx.Client | None = None) -> None: self._owns_client = client is None self._default_qf_version = get_default_qf_version() self._client = client or httpx.Client( timeout=timeout or get_timeout_seconds(), follow_redirects=True, trust_env=False, ) def close(self) -> None: if self._owns_client: self._client.close() def public_request( self, method: str, base_url: str, path: str, *, params: JSONObject | None = None, json_body: JSONValue = None, unwrap: bool = True, qf_version: str | None = None, ) -> JSONValue: return self.public_request_with_meta( method, base_url, path, params=params, json_body=json_body, unwrap=unwrap, qf_version=qf_version, ).data def public_request_with_meta( self, method: str, base_url: str, path: str, *, params: JSONObject | None = None, json_body: JSONValue = None, unwrap: bool = True, qf_version: str | None = None, ) -> BackendResponse: return self._request_with_meta( method, self._build_url(base_url, path), params=params, json_body=json_body, headers=self._base_headers(None, None, qf_version=qf_version), unwrap=unwrap, ) def public_request_with_headers( self, method: str, base_url: str, path: str, *, params: JSONObject | None = None, json_body: JSONValue = None, unwrap: bool = True, qf_version: str | None = None, headers: dict[str, str] | None = None, ) -> BackendResponse: request_headers = self._base_headers(None, None, qf_version=qf_version) if headers: request_headers.update({key: value for key, value in headers.items() if value is not None}) return self._request_with_meta( method, self._build_url(base_url, path), params=params, json_body=json_body, headers=request_headers, unwrap=unwrap, ) def request( self, method: str, context: BackendRequestContext, path: str, *, params: JSONObject | None = None, json_body: JSONValue = None, unwrap: bool = True, ) -> JSONValue: return self.request_with_meta( method, context, path, params=params, json_body=json_body, unwrap=unwrap, ).data def request_with_meta( self, method: str, context: BackendRequestContext, path: str, *, params: JSONObject | None = None, json_body: JSONValue = None, unwrap: bool = True, ) -> BackendResponse: return self._request_with_meta( method, self._build_url(context.base_url, path), params=params, json_body=json_body, headers=self._base_headers( context.token, context.ws_id, context.qf_request_id, qf_version=context.qf_version, ), unwrap=unwrap, ) def stream_request( self, method: str, context: BackendRequestContext, path: str, *, params: JSONObject | None = None, json_body: JSONValue = None, headers: dict[str, str] | None = None, ) -> list[str]: request_headers = self._base_headers( context.token, context.ws_id, context.qf_request_id, qf_version=context.qf_version, ) if headers: request_headers.update({key: value for key, value in headers.items() if value is not None}) return self._stream_lines( method, self._build_url(context.base_url, path), params=params, json_body=json_body, headers=request_headers, ) def public_stream_request( self, method: str, base_url: str, path: str, *, params: JSONObject | None = None, json_body: JSONValue = None, headers: dict[str, str] | None = None, qf_version: str | None = None, ) -> list[str]: request_headers = self._base_headers(None, None, qf_version=qf_version) if headers: request_headers.update({key: value for key, value in headers.items() if value is not None}) return self._stream_lines( method, self._build_url(base_url, path), params=params, json_body=json_body, headers=request_headers, ) def describe_route(self, context: BackendRequestContext) -> JSONObject: qf_version, source = self._resolve_qf_version(context.qf_version) if context.qf_version is not None and context.qf_version_source: source = context.qf_version_source return { "base_url": normalize_base_url(context.base_url) or context.base_url, "qf_version": qf_version, "qf_version_source": source, } def upload_binary( self, url: str, content: bytes, *, content_type: str | None = None, headers: dict[str, str] | None = None, ) -> JSONObject: request_headers = dict(headers or {}) if content_type: request_headers.setdefault("Content-Type", content_type) try: response = self._client.put(url, content=content, headers=request_headers or None) except httpx.RequestError as exc: raise QingflowApiError(category="network", message=str(exc)) if response.status_code >= 400: raise QingflowApiError( category="http", message=self._extract_message(response.text) or f"HTTP {response.status_code}", http_status=response.status_code, ) return { "status_code": response.status_code, "headers": dict(response.headers), } def upload_form_file( self, url: str, *, form_fields: dict[str, str], file_field: str, file_name: str, content: bytes, content_type: str | None = None, headers: dict[str, str] | None = None, ) -> JSONObject: try: response = self._client.post( url, data=form_fields, files={file_field: (file_name, content, content_type or "application/octet-stream")}, headers=headers or None, ) except httpx.RequestError as exc: raise QingflowApiError(category="network", message=str(exc)) if response.status_code >= 400: raise QingflowApiError( category="http", message=self._extract_message(response.text) or f"HTTP {response.status_code}", http_status=response.status_code, ) body: JSONValue = None if response.content: try: body = response.json() except ValueError: body = response.text return { "status_code": response.status_code, "headers": dict(response.headers), "body": body, } def download_binary(self, url: str, *, headers: dict[str, str] | None = None) -> bytes: try: response = self._client.get(url, headers=headers or None) except httpx.RequestError as exc: raise QingflowApiError(category="network", message=str(exc)) if response.status_code >= 400: raise QingflowApiError( category="http", message=self._extract_message(response.text) or f"HTTP {response.status_code}", http_status=response.status_code, ) return response.content def download_binary_with_cookie( self, context: BackendRequestContext, url: str, *, cookie_name: str, headers: dict[str, str] | None = None, ) -> BackendBinaryResponse: qf_version, _source = self._resolve_qf_version(context.qf_version) request_headers = { "User-Agent": DEFAULT_USER_AGENT, "Qf-Request-Id": context.qf_request_id or str(uuid4()), } if headers: request_headers.update({key: value for key, value in headers.items() if value is not None}) cookie_parts = [f"{cookie_name}={context.token}"] if qf_version: cookie_parts.append(f"qfVersion={qf_version}") request_headers["Cookie"] = "; ".join(cookie_parts) try: response = self._client.get(url, headers=request_headers, follow_redirects=False) except httpx.RequestError as exc: raise QingflowApiError(category="network", message=str(exc)) redirected = 300 <= response.status_code < 400 and bool(response.headers.get("location")) if redirected: redirect_headers = {key: value for key, value in request_headers.items() if key.lower() != "cookie"} redirect_url = urljoin(str(response.url), response.headers["location"]) try: response = self._client.get(redirect_url, headers=redirect_headers) except httpx.RequestError as exc: raise QingflowApiError(category="network", message=str(exc)) if response.status_code >= 400: raise QingflowApiError( category="http", message=self._extract_message(response.text) or f"HTTP {response.status_code}", http_status=response.status_code, ) return BackendBinaryResponse( content=response.content, headers=dict(response.headers), http_status=response.status_code, final_url=str(response.url), redirected=redirected or bool(response.history) or str(response.url) != url, ) def request_multipart( self, method: str, context: BackendRequestContext, path: str, *, data: dict[str, Any] | None = None, files: dict[str, tuple[str, bytes, str | None]] | None = None, unwrap: bool = True, ) -> JSONValue: return self.request_multipart_with_meta( method, context, path, data=data, files=files, unwrap=unwrap, ).data def request_multipart_with_meta( self, method: str, context: BackendRequestContext, path: str, *, data: dict[str, Any] | None = None, files: dict[str, tuple[str, bytes, str | None]] | None = None, unwrap: bool = True, ) -> BackendResponse: headers = self._base_headers( context.token, context.ws_id, context.qf_request_id, qf_version=context.qf_version, ) request_files: dict[str, tuple[str, bytes, str | None]] | None = None if files: request_files = {key: value for key, value in files.items()} try: response = self._client.request( method.upper(), self._build_url(context.base_url, path), data=data, files=request_files, headers=headers, ) except httpx.RequestError as exc: raise QingflowApiError(category="network", message=str(exc), request_id=headers["Qf-Request-Id"]) parsed = self._parse_response(response, headers["Qf-Request-Id"], unwrap=unwrap) return BackendResponse( data=parsed, headers=dict(response.headers), request_id=headers["Qf-Request-Id"], http_status=response.status_code, qf_response_version=self._extract_response_qf_version(response.headers), ) def start_socket_data_import( self, context: BackendRequestContext, *, app_key: str, being_enter_auditing: bool, view_key: str | None, excel_url: str, excel_name: str, ack_timeout_seconds: float = 8.0, initial_wait_seconds: float = 4.0, ) -> dict[str, Any]: try: import socketio # type: ignore[import-not-found] except ImportError as exc: raise QingflowApiError( category="config", message=f"socket.io client dependency is missing: {exc}", ) socket_base_url = self._build_socket_base_url(context.base_url) import_result: dict[str, Any] = { "import_id": None, "process_id_str": None, "status": "accepted", "warnings": [], "initial_event": None, "failure_event": None, } initial_event_received = Event() failure_event_received = Event() sio = socketio.Client(reconnection=False, logger=False, engineio_logger=False) def _unwrap_socket_event_payload(payload: Any) -> Any: if isinstance(payload, list) and payload: return payload[0] return payload def _handle_initial(payload: Any) -> None: normalized_payload = _unwrap_socket_event_payload(payload) import_result["initial_event"] = normalized_payload if isinstance(normalized_payload, dict): process_id = normalized_payload.get("processIdStr") or normalized_payload.get("process_id_str") or normalized_payload.get("processId") if process_id is not None: import_result["process_id_str"] = str(process_id) initial_event_received.set() def _handle_failure(payload: Any) -> None: normalized_payload = _unwrap_socket_event_payload(payload) import_result["failure_event"] = normalized_payload if isinstance(normalized_payload, dict): process_id = normalized_payload.get("processIdStr") or normalized_payload.get("process_id_str") or normalized_payload.get("processId") if process_id is not None: import_result["process_id_str"] = str(process_id) failure_event_received.set() try: sio.connect( socket_base_url, transports=["websocket"], socketio_path="socket.io", headers=self._base_headers( context.token, context.ws_id, qf_version=context.qf_version, ), wait_timeout=ack_timeout_seconds, ) sio.emit("token", context.token) sleep(0.2) ack = sio.call( "dataImport", ( context.token, app_key, bool(being_enter_auditing), view_key, False, excel_url, excel_name, ), timeout=ack_timeout_seconds, ) ack_payload = ack[0] if isinstance(ack, list) and ack else ack if isinstance(ack_payload, dict): error_code = ack_payload.get("error") ack_message = ack_payload.get("message") import_id = ack_payload.get("data") if error_code not in (None, 0): raise QingflowApiError( category="backend", message=str(ack_message or f"socket import rejected with error {error_code}"), details={"socket_error_code": error_code, "import_id": import_id}, ) else: import_id = ack_payload if not import_id: raise QingflowApiError(category="backend", message="socket import ack did not return import_id") import_result["import_id"] = str(import_id) sio.on(f"dataImportRes_{import_result['import_id']}", _handle_initial) sio.on(f"dataImportFail_{import_result['import_id']}", _handle_failure) if not initial_event_received.wait(timeout=initial_wait_seconds) and not failure_event_received.wait(timeout=0.1): import_result["warnings"].append( { "code": "IMPORT_SOCKET_INITIAL_EVENT_PENDING", "message": "Import ack received, but no initial progress payload arrived within the initial wait window.", } ) except Exception as exc: message = str(exc) if "timeout" in message.lower(): raise QingflowApiError( category="network", message="socket import ack timed out", details={"error_code": "IMPORT_SOCKET_ACK_TIMEOUT"}, ) if isinstance(exc, QingflowApiError): raise raise QingflowApiError(category="network", message=message or "socket import failed") finally: try: if sio.connected: sio.disconnect() except Exception: pass return import_result def start_socket_record_export( self, context: BackendRequestContext, *, app_key: str, view_id: str, filter_bean: JSONObject, export_config: JSONObject, view_key: str | None = None, result_amount: int = 0, ack_timeout_seconds: float = 8.0, ) -> dict[str, Any]: try: import socketio # type: ignore[import-not-found] except ImportError as exc: raise QingflowApiError( category="config", message=f"socket.io client dependency is missing: {exc}", ) socket_base_url = self._build_socket_base_url(context.base_url) export_result: dict[str, Any] = { "backend_export_id": None, "warnings": [], } sio = socketio.Client(reconnection=False, logger=False, engineio_logger=False) event_name = "excelViewgraph" if view_key else "excel" event_args: tuple[Any, ...] if view_key: event_args = ( context.token, None, view_key, filter_bean, export_config, int(result_amount), ) else: event_args = ( context.token, app_key, filter_bean, export_config, int(result_amount), ) try: sio.connect( socket_base_url, transports=["websocket"], socketio_path="socket.io", headers=self._base_headers( context.token, context.ws_id, qf_version=context.qf_version, ), wait_timeout=ack_timeout_seconds, ) sio.emit("token", context.token) sleep(0.2) ack = sio.call( event_name, event_args, timeout=ack_timeout_seconds, ) ack_payload = ack[0] if isinstance(ack, list) and ack else ack export_id: Any = ack_payload if isinstance(ack_payload, dict): error_code = ack_payload.get("error") ack_message = ack_payload.get("message") export_id = ack_payload.get("data") if isinstance(export_id, dict): export_id = ( export_id.get("exportId") or export_id.get("export_id") or export_id.get("id") ) if error_code not in (None, 0): raise QingflowApiError( category="backend", message=str(ack_message or f"socket export rejected with error {error_code}"), details={ "socket_error_code": error_code, "app_key": app_key, "view_id": view_id, "view_key": view_key, }, ) if not export_id: raise QingflowApiError(category="backend", message="socket export ack did not return export_id") export_result["backend_export_id"] = str(export_id) except Exception as exc: message = str(exc) if "timeout" in message.lower(): raise QingflowApiError( category="network", message="socket export ack timed out", details={"error_code": "EXPORT_SOCKET_ACK_TIMEOUT"}, ) if isinstance(exc, QingflowApiError): raise raise QingflowApiError(category="network", message=message or "socket export failed") finally: try: if sio.connected: sio.disconnect() except Exception: pass return export_result def _request_with_meta( self, method: str, url: str, *, params: JSONObject | None, json_body: JSONValue, headers: dict[str, str], unwrap: bool, ) -> BackendResponse: attempts = 2 if method.upper() == "GET" else 1 last_error: QingflowApiError | None = None for _ in range(attempts): try: response = self._client.request(method.upper(), url, params=params, json=json_body, headers=headers) parsed = self._parse_response(response, headers["Qf-Request-Id"], unwrap=unwrap) return BackendResponse( data=parsed, headers=dict(response.headers), request_id=headers["Qf-Request-Id"], http_status=response.status_code, qf_response_version=self._extract_response_qf_version(response.headers), ) except httpx.RequestError as exc: last_error = QingflowApiError(category="network", message=str(exc), request_id=headers["Qf-Request-Id"]) assert last_error is not None raise last_error def _stream_lines( self, method: str, url: str, *, params: JSONObject | None, json_body: JSONValue, headers: dict[str, str], ) -> list[str]: try: with self._client.stream(method.upper(), url, params=params, json=json_body, headers=headers) as response: request_id = headers["Qf-Request-Id"] if response.status_code >= 400: raw_bytes = response.read() payload: JSONValue try: payload = response.json() except ValueError: payload = raw_bytes.decode("utf-8", errors="replace") raise QingflowApiError( category="http", message=self._extract_message(payload) or f"HTTP {response.status_code}", backend_code=self._extract_code(payload), request_id=request_id, http_status=response.status_code, ) return [line for line in response.iter_lines()] except httpx.RequestError as exc: raise QingflowApiError(category="network", message=str(exc), request_id=headers["Qf-Request-Id"]) def _parse_response(self, response: httpx.Response, request_id: str, *, unwrap: bool) -> JSONValue: payload: JSONValue try: payload = response.json() except ValueError: payload = response.text if response.status_code >= 400: raise QingflowApiError( category="http", message=self._extract_message(payload) or f"HTTP {response.status_code}", backend_code=self._extract_code(payload), request_id=request_id, http_status=response.status_code, ) if not unwrap: return payload return self._unwrap_payload(payload, request_id, response.status_code) def _unwrap_payload(self, payload: JSONValue, request_id: str, http_status: int) -> JSONValue: if not isinstance(payload, dict): return payload if "success" in payload: if not bool(payload.get("success")): raise QingflowApiError( category="backend", message=self._extract_message(payload) or "Qingflow request failed", backend_code=self._extract_code(payload), request_id=request_id, http_status=http_status, ) return self._extract_success_data(payload) code = self._extract_code(payload) if code not in (None, 0, "0"): raise QingflowApiError( category="backend", message=self._extract_message(payload) or "Qingflow request failed", backend_code=code, request_id=request_id, http_status=http_status, ) if code in (0, "0"): return self._extract_success_data(payload) return payload def _extract_success_data(self, payload: JSONObject) -> JSONValue: for key in ("result", "data", "page", "obj"): if key in payload: return payload[key] return payload def _extract_message(self, payload: JSONValue) -> str | None: if not isinstance(payload, dict): return str(payload) if payload else None for key in ("message", "msg", "errMsg", "error", "detail"): value = payload.get(key) if value: return str(value) return None def _extract_code(self, payload: JSONValue) -> JSONScalar: if isinstance(payload, dict): return payload.get("code", payload.get("errCode")) return None def _base_headers( self, token: str | None, ws_id: int | None, request_id: str | None = None, *, qf_version: str | None = None, ) -> dict[str, str]: headers = { "User-Agent": DEFAULT_USER_AGENT, "Qf-Request-Id": request_id or str(uuid4()), } resolved_qf_version, _ = self._resolve_qf_version(qf_version) if resolved_qf_version: headers["Cookie"] = f"qfVersion={resolved_qf_version}" if token: headers["token"] = token if ws_id is not None: headers["wsId"] = str(ws_id) return headers def _resolve_qf_version(self, explicit_qf_version: str | None) -> tuple[str | None, str]: if explicit_qf_version is not None: normalized = str(explicit_qf_version).strip() or None return normalized, "context" if self._default_qf_version: return self._default_qf_version, "default_config" return None, "unset" def _extract_response_qf_version(self, headers: httpx.Headers) -> str | None: value = headers.get("x-q-response-version") if value is None: return None normalized = str(value).strip() return normalized or None def _build_url(self, base_url: str, path: str) -> str: normalized = normalize_base_url(base_url) if not normalized: raise QingflowApiError.config_error("base_url is required") return f"{normalized}/{path.lstrip('/')}" def _build_socket_base_url(self, base_url: str) -> str: normalized = normalize_base_url(base_url) if not normalized: raise QingflowApiError.config_error("base_url is required") parsed = urlsplit(normalized) path = parsed.path.rstrip("/") if path.endswith("/api"): path = path[:-4] return urlunsplit((parsed.scheme, parsed.netloc, path or "", "", ""))