import asyncio import concurrent.futures import json import random import threading import time import uuid from dataclasses import dataclass, field from datetime import datetime, UTC from typing import ( Any, Final, ) from collections.abc import Callable, MutableMapping import anyio import demistomock as demisto import httpx from pydantic import BaseModel, Field # pylint: disable=E0611 from CommonServerPython import * # noqa: F401,F403 from CommonServerUserPython import * # noqa: F401,F403 # Type aliases for better readability HeadersType = dict[str, str] ParamsType = dict[str, Any] JsonType = dict[str, Any] StatusCodesType = tuple[int, ...] # Constants DEFAULT_USER_AGENT: Final[str] = "ContentClient/1.0" def _now() -> float: return time.monotonic() def _ensure_dict(value: MutableMapping[str, Any] | None) -> dict[str, Any]: if value is None: return {} return dict(value) def _get_value_by_path(obj: Any, path: str) -> Any: """Retrieve a value from a nested dictionary using dot notation. Args: obj: The dictionary or object to search. path: The dot-separated path (e.g., "data.items"). Returns: The value at the path, or None if not found. """ if not path or not isinstance(path, str): return obj current = obj for part in path.split("."): if not part: # Skip empty parts from ".." or leading/trailing dots continue if isinstance(current, dict): current = current.get(part) elif isinstance(current, list): try: idx = int(part) if 0 <= idx < len(current): current = current[idx] else: return None except ValueError: return None else: return None if current is None: return None return current def _extract_list(data: Any, path: str | None) -> list[Any]: extracted = _get_value_by_path(data, path) if path else data if extracted is None: return [] if isinstance(extracted, list): return extracted if isinstance(extracted, dict): return [extracted] # Handle primitive types explicitly if isinstance(extracted, str | int | float | bool): return [extracted] return [extracted] def _parse_retry_after(response: httpx.Response | None) -> float | None: if not response: return None retry_after = response.headers.get("Retry-After") if not retry_after: return None try: return float(retry_after) except ValueError: try: parsed = datetime.strptime(retry_after, "%a, %d %b %Y %H:%M:%S GMT").replace(tzinfo=UTC) return max(0.0, (parsed - datetime.now(UTC)).total_seconds()) except ValueError: return None # Error Classes class ContentClientError(DemistoException): """Base error for all content client failures.""" def __init__(self, message: str, response: httpx.Response | None = None): super().__init__(message) self.response = response class ContentClientAuthenticationError(ContentClientError): """Raised when authentication cannot be completed.""" class ContentClientRateLimitError(ContentClientError): """Raised when the client hits a user-defined rate limit.""" class ContentClientTimeoutError(ContentClientError): """Raised when the execution timeout window is about to be exceeded.""" class ContentClientCircuitOpenError(ContentClientError): """Raised when the circuit breaker prevents additional requests.""" class ContentClientRetryError(ContentClientError): """Raised when requests exhaust all retry attempts.""" class ContentClientConfigurationError(ContentClientError): """Raised when the blueprint or request definition is invalid.""" @dataclass class ClientExecutionMetrics: """Metrics for content client execution.""" success: int = 0 retry_error: int = 0 quota_error: int = 0 auth_error: int = 0 service_error: int = 0 general_error: int = 0 # Helper Classes class RetryPolicy(BaseModel): """Retry policy for handling transient API failures. Configures exponential backoff with jitter for retrying failed requests. Attributes: max_attempts: Maximum number of retry attempts (default: 5). initial_delay: Initial delay in seconds before first retry (default: 1.0). multiplier: Multiplier for exponential backoff (default: 2.0). max_delay: Maximum delay between retries in seconds (default: 60.0). jitter: Random jitter factor (0.0-1.0) to prevent thundering herd (default: 0.2). retryable_status_codes: HTTP status codes that trigger a retry. retryable_exceptions: Exception types that trigger a retry. respect_retry_after: Whether to honor Retry-After headers (default: True). """ class Config: extra = "forbid" arbitrary_types_allowed = True # Required for Tuple[Type[Exception], ...] max_attempts: int = Field(5, ge=1) initial_delay: float = Field(1.0, ge=0) multiplier: float = Field(2.0, ge=1.0) max_delay: float = Field(60.0, gt=0) jitter: float = Field(0.2, ge=0.0, le=1.0) retryable_status_codes: tuple[int, ...] = (408, 413, 425, 429, 500, 502, 503, 504) retryable_exceptions: tuple[type[Exception], ...] = ( httpx.ConnectError, httpx.ReadTimeout, httpx.WriteTimeout, httpx.RemoteProtocolError, httpx.PoolTimeout, ) respect_retry_after: bool = True def next_delay(self, attempt: int, retry_after: float | None = None) -> float: """Calculate the next delay for retry with exponential backoff and jitter. Args: attempt: The current attempt number (1-based). retry_after: Optional server-provided retry-after value in seconds. Returns: The delay in seconds before the next retry attempt. """ if retry_after is not None and self.respect_retry_after: return retry_after delay = min(self.max_delay, self.initial_delay * (self.multiplier ** (attempt - 1))) jitter_value = delay * self.jitter return max(0.0, delay + random.uniform(-jitter_value, jitter_value)) class CircuitBreakerPolicy(BaseModel): """Circuit breaker policy for preventing cascading failures.""" class Config: extra = "forbid" failure_threshold: int = Field(5, ge=1) recovery_timeout: float = Field(60.0, gt=0) class CircuitBreaker: """Circuit breaker implementation for preventing cascading failures. Tracks failures and opens the circuit when the failure threshold is reached, preventing further requests until the recovery timeout expires. This implementation uses a half-open state with probe request support: - CLOSED: Normal operation, requests are allowed - OPEN: Circuit is tripped, requests are blocked - HALF_OPEN: Recovery timeout expired, one probe request is allowed The circuit only fully closes after a successful probe request. This implementation is thread-safe using a threading lock. Attributes: policy: The circuit breaker policy configuration. """ def __init__(self, policy: CircuitBreakerPolicy) -> None: self.policy: CircuitBreakerPolicy = policy self._failure_count: int = 0 self._opened_at: float | None = None self._half_open: bool = False self._lock: threading.Lock = threading.Lock() def can_execute(self) -> bool: """Check if the circuit allows execution. Returns: True if the circuit is closed or half-open (allowing probe), False if open. """ with self._lock: if self._opened_at is None: return True elapsed: float = _now() - self._opened_at if elapsed >= self.policy.recovery_timeout: # Enter half-open state - allow one probe request if not self._half_open: self._half_open = True return True # Already in half-open and probe is in progress, block additional requests return False return False def record_success(self) -> None: """Record a successful execution, resetting the failure count and closing the circuit.""" with self._lock: self._failure_count = 0 self._opened_at = None self._half_open = False def record_failure(self) -> None: """Record a failed execution, potentially opening the circuit.""" with self._lock: self._failure_count += 1 if self._half_open: # Probe failed, re-open the circuit self._opened_at = _now() self._half_open = False elif self._failure_count >= self.policy.failure_threshold: self._opened_at = _now() class RateLimitPolicy(BaseModel): """Rate limiting policy using token bucket algorithm.""" class Config: extra = "forbid" rate_per_second: float = Field(0.0, ge=0) burst: int = Field(1, ge=1) respect_retry_after: bool = True @property def enabled(self) -> bool: return self.rate_per_second > 0 class TokenBucketRateLimiter: """Token bucket rate limiter for controlling request rate. Uses the token bucket algorithm to limit the rate of requests. Tokens are refilled at a constant rate up to a maximum capacity (burst). This implementation uses threading.Lock for thread safety, which works correctly across different event loops and in synchronous contexts. Attributes: policy: The rate limit policy configuration. """ def __init__(self, policy: RateLimitPolicy) -> None: self.policy: RateLimitPolicy = policy self._capacity: int = max(1, policy.burst) self._tokens: float = float(self._capacity) self._updated: float = _now() self._lock: threading.Lock = threading.Lock() # Thread-safe lock async def acquire(self) -> None: """Acquire a token, waiting if necessary until one is available.""" while True: with self._lock: self._refill_locked() if self._tokens >= 1: self._tokens -= 1 return needed = 1 - self._tokens # Protect against division by zero - this should not happen in normal usage # since the rate limiter is only created when rate_per_second > 0 rate = self.policy.rate_per_second if rate <= 0: raise ContentClientConfigurationError("rate_per_second must be positive when rate limiting is enabled") wait_seconds = needed / rate await anyio.sleep(wait_seconds) def _refill_locked(self) -> None: """Refill tokens based on elapsed time (must be called with lock held).""" now = _now() delta = now - self._updated if delta <= 0: return refill = delta * self.policy.rate_per_second self._tokens = min(self._capacity, self._tokens + refill) self._updated = now class TimeoutSettings(BaseModel): """Timeout configuration for HTTP requests and execution deadline.""" class Config: extra = "forbid" connect: float = Field(10.0, gt=0) read: float = Field(60.0, gt=0) write: float = Field(60.0, gt=0) pool: float = Field(60.0, gt=0) execution: float | None = Field(None, gt=0) safety_buffer: float = Field(30.0, gt=0) def as_httpx(self) -> httpx.Timeout: return httpx.Timeout(connect=self.connect, read=self.read, write=self.write, pool=self.pool) @dataclass class ContentClientState: """Pagination and collection state for resuming after timeouts. Stores pagination position (cursor, page, offset) and metadata to allow seamless resumption of collection after timeouts or interruptions. Attributes: cursor: Current cursor value for cursor-based pagination. page: Current page number for page-based pagination. offset: Current offset value for offset-based pagination. last_event_id: Last processed event ID (for deduplication). partial_results: Events from incomplete pages (preserved on timeout). metadata: Custom metadata dictionary for storing additional state. """ cursor: str | None = None page: int | None = None offset: int | None = None last_event_id: str | None = None partial_results: list[Any] = field(default_factory=list) metadata: dict[str, Any] = field(default_factory=dict) def to_dict(self) -> dict[str, Any]: """Convert state to a dictionary for serialization. Returns: Dictionary representation of the state. """ return { "cursor": self.cursor, "page": self.page, "offset": self.offset, "last_event_id": self.last_event_id, "partial_results": self.partial_results, "metadata": self.metadata, } @classmethod def from_dict(cls, raw: dict[str, Any] | None) -> "ContentClientState": """Create a ContentClientState from a dictionary. Args: raw: Dictionary containing state data. Returns: A new ContentClientState instance. """ if not raw: return cls() return cls( cursor=raw.get("cursor"), page=raw.get("page"), offset=raw.get("offset"), last_event_id=raw.get("last_event_id"), partial_results=raw.get("partial_results", []), metadata=raw.get("metadata", {}), ) class ContentClientContextStore: """Store for persisting state to Demisto integration context. Provides read/write access to the integration context with retry logic for handling transient failures. Uses thread-safe locking to prevent race conditions during concurrent read-modify-write operations. Attributes: namespace: A namespace prefix for storing data (e.g., client name). max_retries: Maximum number of retry attempts for write operations. """ def __init__(self, namespace: str, max_retries: int = 3) -> None: self.namespace: str = namespace self.max_retries: int = max_retries self._lock: threading.Lock = threading.Lock() def read(self) -> dict[str, Any]: """Read the current integration context. Returns: The integration context dictionary. """ with self._lock: context: dict[str, Any] = demisto.getIntegrationContext() or {} return context def write(self, data: dict[str, Any]) -> None: """Write data to the integration context with retry logic. Uses thread-safe locking to prevent race conditions. Args: data: The data dictionary to write to the context. Raises: Exception: If all retry attempts fail. """ last_error: Exception | None = None with self._lock: for attempt in range(self.max_retries): try: demisto.setIntegrationContext(data) return except Exception as e: last_error = e time.sleep(0.1 * (attempt + 1)) # Simple backoff if last_error: raise last_error # Auth Handlers class AuthHandler: """Abstract base class for authentication handlers.""" name: str = "auth" async def on_request(self, client: "ContentClient", request: httpx.Request) -> None: """Modify the request to add authentication credentials.""" raise NotImplementedError("Subclasses must implement on_request()") async def on_auth_failure(self, client: "ContentClient", response: httpx.Response) -> bool: """Handle authentication failure response.""" return False class APIKeyAuthHandler(AuthHandler): """Authentication handler for API key-based authentication. Supports adding the API key either as a header or as a query parameter. At least one of header_name or query_param must be provided. Args: key: The API key value. header_name: Optional header name to add the key to (e.g., 'X-API-Key'). query_param: Optional query parameter name to add the key to (e.g., 'api_key'). Raises: ContentClientConfigurationError: If neither header_name nor query_param is provided, or if the key is empty. """ def __init__(self, key: str, header_name: str | None = None, query_param: str | None = None): super().__init__() if not key: raise ContentClientConfigurationError("APIKeyAuthHandler requires a non-empty key") if not header_name and not query_param: raise ContentClientConfigurationError("APIKeyAuthHandler requires header_name or query_param") self.key = key self.header_name = header_name self.query_param = query_param self.name = "api_key" async def on_request(self, client: "ContentClient", request: httpx.Request) -> None: if self.header_name: request.headers[self.header_name] = self.key if self.query_param: # Use copy_add_param for proper URL parameter handling request.url = request.url.copy_add_param(self.query_param, self.key) class BearerTokenAuthHandler(AuthHandler): """Authentication handler for Bearer token authentication. Args: token: The bearer token value. Raises: ContentClientConfigurationError: If the token is empty. """ def __init__(self, token: str): super().__init__() if not token: raise ContentClientConfigurationError("BearerTokenAuthHandler requires a non-empty token") self.token = token self.name = "bearer" async def on_request(self, client: "ContentClient", request: httpx.Request) -> None: request.headers["Authorization"] = f"Bearer {self.token}" class BasicAuthHandler(AuthHandler): """Authentication handler for HTTP Basic Authentication. Args: username: The username for authentication. password: The password for authentication. Raises: ContentClientConfigurationError: If username is empty. """ def __init__(self, username: str, password: str): super().__init__() if not username: raise ContentClientConfigurationError("BasicAuthHandler requires a non-empty username") credentials = f"{username}:{password}" self._encoded = b64_encode(credentials) self.name = "basic" async def on_request(self, client: "ContentClient", request: httpx.Request) -> None: request.headers["Authorization"] = f"Basic {self._encoded}" class OAuth2ClientCredentialsHandler(AuthHandler): """Authentication handler for OAuth2 Client Credentials flow. Implements the OAuth2 client credentials grant type for machine-to-machine authentication. Automatically handles token refresh when tokens expire. Uses thread-safe locking to prevent race conditions during token refresh, which is important when the handler is used across different event loops or in synchronous contexts. Args: token_url: The OAuth2 token endpoint URL. client_id: The client ID for authentication. client_secret: The client secret for authentication. scope: Optional OAuth2 scope(s) to request. audience: Optional audience parameter for the token request. auth_params: Optional additional parameters to include in token requests. context_store: Optional context store for persisting tokens across executions. token_timeout: Timeout in seconds for token refresh requests (default: 30). Raises: ContentClientConfigurationError: If required parameters are missing or invalid. """ def __init__( self, token_url: str, client_id: str, client_secret: str, scope: str | None = None, audience: str | None = None, auth_params: dict[str, str] | None = None, context_store: Any | None = None, token_timeout: float = 30.0, ): super().__init__() if not token_url: raise ContentClientConfigurationError("OAuth2ClientCredentialsHandler requires a non-empty token_url") if not client_id: raise ContentClientConfigurationError("OAuth2ClientCredentialsHandler requires a non-empty client_id") if not client_secret: raise ContentClientConfigurationError("OAuth2ClientCredentialsHandler requires a non-empty client_secret") self.token_url = token_url self.client_id = client_id self.client_secret = client_secret self.scope = scope self.audience = audience self.auth_params = auth_params or {} self.context_store = context_store self.token_timeout = token_timeout self.name = "oauth2_client_credentials" self._access_token: str | None = None self._expires_at: float = 0 self._lock: threading.Lock = threading.Lock() # Thread-safe lock # Try to load cached token from context store if self.context_store: try: context = self.context_store.read() cached_token = context.get("oauth2_token", {}) if cached_token.get("access_token") and cached_token.get("expires_at", 0) > _now(): self._access_token = cached_token["access_token"] self._expires_at = cached_token["expires_at"] except Exception as e: # Log but don't fail if cache read fails demisto.debug(f"Failed to load cached OAuth2 token: {e}") async def on_request(self, client: "ContentClient", request: httpx.Request) -> None: # Check if refresh is needed (thread-safe read) if self._should_refresh(): # Perform refresh outside the lock to avoid blocking async operations await self._refresh_token_if_needed(client) # Read token under lock to ensure visibility with self._lock: if self._access_token: request.headers["Authorization"] = f"Bearer {self._access_token}" async def on_auth_failure(self, client: "ContentClient", response: httpx.Response) -> bool: # If we get 401, force refresh token await self._force_refresh_token(client) return True def _should_refresh(self) -> bool: """Check if token needs refresh. Thread-safe read.""" with self._lock: return not self._access_token or _now() >= self._expires_at - 60 # Refresh 60s before expiry async def _refresh_token_if_needed(self, client: "ContentClient") -> None: """Refresh token if needed, with proper locking that doesn't block async operations.""" # Double-check pattern: check again after we're ready to refresh # We use a flag to track if we should refresh, then release the lock before awaiting should_refresh = False with self._lock: if self._should_refresh_unlocked(): should_refresh = True if should_refresh: await self._refresh_token(client) async def _force_refresh_token(self, client: "ContentClient") -> None: """Force refresh the token (used after auth failure).""" await self._refresh_token(client) def _should_refresh_unlocked(self) -> bool: """Check if token needs refresh. Must be called with lock held.""" return not self._access_token or _now() >= self._expires_at - 60 async def _refresh_token(self, client: "ContentClient") -> None: """Refresh the OAuth2 access token. Uses a separate HTTP client for token refresh to avoid recursion/deadlocks and to not share state with the main client. This method should be called with the lock held. Args: client: The ContentClient instance (used for SSL verification settings). Raises: ContentClientAuthenticationError: If token refresh fails. """ # Use a separate client for token refresh to avoid recursion/deadlocks # and to not share state with the main client async with httpx.AsyncClient( verify=client._verify, timeout=httpx.Timeout(self.token_timeout), follow_redirects=True ) as token_client: data = { "grant_type": "client_credentials", "client_id": self.client_id, "client_secret": self.client_secret, } if self.scope: data["scope"] = self.scope if self.audience: data["audience"] = self.audience if self.auth_params: data.update(self.auth_params) try: response = await token_client.post(self.token_url, data=data) response.raise_for_status() token_data = response.json() self._access_token = token_data.get("access_token") if not self._access_token: raise ContentClientAuthenticationError("No access_token in response") expires_in = token_data.get("expires_in", 3600) self._expires_at = _now() + expires_in # Persist token if context store is available if self.context_store: try: current_context = self.context_store.read() current_context["oauth2_token"] = {"access_token": self._access_token, "expires_at": self._expires_at} self.context_store.write(current_context) except Exception as e: # Log but don't fail auth if persistence fails demisto.debug(f"Failed to persist OAuth2 token: {e}") except httpx.HTTPStatusError as e: raise ContentClientAuthenticationError( f"Token refresh failed with status {e.response.status_code}: {e.response.text}" ) from e except httpx.TimeoutException as e: raise ContentClientAuthenticationError(f"Token refresh timed out: {str(e)}") from e except Exception as e: raise ContentClientAuthenticationError(f"Failed to refresh token: {str(e)}") from e # Structured Logging for Google Cloud class StructuredLogEntry: """Structured log entry optimized for Google Cloud Logs Explorer and BigQuery. Creates JSON-formatted log entries with consistent fields that can be easily queried in Google Cloud Logs Explorer and exported to BigQuery for analysis. Fields follow Google Cloud Logging conventions: - severity: Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL) - message: Human-readable log message - timestamp: ISO 8601 formatted timestamp - labels: Key-value pairs for filtering (client_name, request_id, etc.) - httpRequest: HTTP request details (method, url, status, latency) - Additional custom fields for debugging """ def __init__(self, severity: str, message: str, client_name: str, request_id: str | None = None, **kwargs: Any) -> None: self.entry: dict[str, Any] = { "severity": severity.upper(), "message": message, "timestamp": datetime.now(UTC).isoformat(), "labels": { "client_name": client_name, "component": "ContentClient", }, } if request_id: self.entry["labels"]["request_id"] = request_id # Add any additional fields for key, value in kwargs.items(): if key == "http_request": self.entry["httpRequest"] = value elif key == "error": self.entry["error"] = value elif key == "labels": self.entry["labels"].update(value) else: self.entry[key] = value def to_json(self) -> str: """Convert to JSON string for logging.""" return json.dumps(self.entry, default=str) def to_dict(self) -> dict[str, Any]: """Convert to dictionary.""" return self.entry def create_http_request_log( method: str, url: str, status: int | None = None, latency_ms: float | None = None, request_size: int | None = None, response_size: int | None = None, user_agent: str | None = None, ) -> dict[str, Any]: """Create an httpRequest object following Google Cloud Logging format. This format is recognized by Google Cloud Logs Explorer and enables automatic HTTP request analysis and filtering. Args: method: HTTP method (GET, POST, etc.) url: Request URL status: HTTP response status code latency_ms: Request latency in milliseconds request_size: Size of request body in bytes response_size: Size of response body in bytes user_agent: User-Agent header value Returns: Dictionary in Google Cloud httpRequest format """ http_request: dict[str, Any] = { "requestMethod": method.upper(), "requestUrl": url, } if status is not None: http_request["status"] = status if latency_ms is not None: # Google Cloud expects latency as a string with 's' suffix http_request["latency"] = f"{latency_ms / 1000:.3f}s" if request_size is not None: http_request["requestSize"] = str(request_size) if response_size is not None: http_request["responseSize"] = str(response_size) if user_agent: http_request["userAgent"] = user_agent return http_request def create_error_log( error_type: str, error_message: str, stack_trace: str | None = None, error_code: str | None = None, ) -> dict[str, Any]: """Create an error object for structured logging. Args: error_type: Type/class of the error error_message: Error message stack_trace: Optional stack trace error_code: Optional error code for categorization Returns: Dictionary with error details """ error: dict[str, Any] = { "type": error_type, "message": error_message, } if stack_trace: error["stackTrace"] = stack_trace if error_code: error["code"] = error_code return error # Diagnostics @dataclass class RequestTrace: """Trace information for a single HTTP request.""" method: str url: str headers: dict[str, str] params: dict[str, Any] body: Any | None timestamp: float response_status: int | None = None response_headers: dict[str, str] | None = None response_body: Any | None = None elapsed_ms: float | None = None error: str | None = None retry_attempt: int = 0 @dataclass class DiagnosticReport: """Comprehensive diagnostic report for troubleshooting.""" content_item_name: str configuration: dict[str, Any] request_traces: list[RequestTrace] state_snapshots: list[dict[str, Any]] performance_metrics: dict[str, Any] errors: list[dict[str, Any]] recommendations: list[str] timestamp: float class ContentClientLogger: """Enhanced logger with diagnostic capabilities and structured logging. Provides structured JSON logging optimized for Google Cloud Logs Explorer and BigQuery analysis. Supports request tracing, performance metrics, and correlation IDs for distributed tracing. Attributes: client_name: Name of the client for log identification. diagnostic_mode: Whether to enable detailed diagnostic logging. structured_logging: Whether to use JSON structured logging (default: True). Example BigQuery queries: -- Find all errors for a specific client SELECT * FROM `project.dataset.logs` WHERE JSON_VALUE(jsonPayload.labels.client_name) = 'MyClient' AND severity = 'ERROR' -- Analyze request latencies SELECT JSON_VALUE(jsonPayload.httpRequest.requestUrl) as url, AVG(CAST(REPLACE(JSON_VALUE(jsonPayload.httpRequest.latency), 's', '') AS FLOAT64) * 1000) as avg_latency_ms FROM `project.dataset.logs` WHERE JSON_VALUE(jsonPayload.labels.component) = 'ContentClient' GROUP BY url -- Find requests by correlation ID SELECT * FROM `project.dataset.logs` WHERE JSON_VALUE(jsonPayload.labels.request_id) = 'abc-123' ORDER BY timestamp """ def __init__(self, client_name: str, diagnostic_mode: bool = False, structured_logging: bool = True) -> None: self.client_name: str = client_name self.diagnostic_mode: bool = diagnostic_mode self.structured_logging: bool = structured_logging self._traces: list[RequestTrace] = [] self._errors: list[dict[str, Any]] = [] self._performance: dict[str, list[float]] = { "request_times": [], "pagination_times": [], "auth_times": [], } self._current_request_id: str | None = None def new_request_id(self) -> str: """Generate a new request ID for correlation. Returns: A unique request ID string. """ self._current_request_id = str(uuid.uuid4())[:8] return self._current_request_id def get_request_id(self) -> str | None: """Get the current request ID. Returns: The current request ID or None if not set. """ return self._current_request_id def debug(self, message: str, extra: dict[str, Any] | None = None) -> None: """Log a debug message (only in diagnostic mode).""" if self.diagnostic_mode: if self.structured_logging: log_entry = StructuredLogEntry( severity="DEBUG", message=message, client_name=self.client_name, request_id=self._current_request_id, **(extra or {}), ) demisto.debug(log_entry.to_json()) else: demisto.debug(self._format("DEBUG", message, extra)) def info(self, message: str, extra: dict[str, Any] | None = None) -> None: """Log an info message.""" if self.structured_logging: log_entry = StructuredLogEntry( severity="INFO", message=message, client_name=self.client_name, request_id=self._current_request_id, **(extra or {}), ) demisto.info(log_entry.to_json()) else: demisto.info(self._format("INFO", message, extra)) def warning(self, message: str, extra: dict[str, Any] | None = None) -> None: """Log a warning message.""" if self.structured_logging: log_entry = StructuredLogEntry( severity="WARNING", message=message, client_name=self.client_name, request_id=self._current_request_id, **(extra or {}), ) demisto.debug(log_entry.to_json()) else: demisto.debug(self._format("WARNING", message, extra)) def error(self, message: str, extra: dict[str, Any] | None = None) -> None: """Log an error message.""" if self.structured_logging: error_info = None # Create a copy of extra to avoid modifying the original extra_copy = dict(extra) if extra else {} if extra_copy and "error_type" in extra_copy: error_info = create_error_log( error_type=extra_copy.get("error_type", "unknown"), error_message=extra_copy.get("error", message), ) # Remove keys that are handled separately to avoid conflicts extra_copy.pop("error_type", None) extra_copy.pop("error", None) log_entry = StructuredLogEntry( severity="ERROR", message=message, client_name=self.client_name, request_id=self._current_request_id, error=error_info, **extra_copy, ) demisto.error(log_entry.to_json()) else: demisto.error(self._format("ERROR", message, extra)) if self.diagnostic_mode and extra: self._errors.append({"message": message, "context": extra}) def log_http_request( self, method: str, url: str, status: int | None = None, latency_ms: float | None = None, request_size: int | None = None, response_size: int | None = None, retry_attempt: int = 0, error: str | None = None, ) -> None: """Log an HTTP request with structured format for Google Cloud. This creates a log entry that can be easily queried in Logs Explorer and analyzed in BigQuery. Args: method: HTTP method url: Request URL status: HTTP response status code latency_ms: Request latency in milliseconds request_size: Size of request body in bytes response_size: Size of response body in bytes retry_attempt: Current retry attempt number error: Optional error message """ http_request = create_http_request_log( method=method, url=url, status=status, latency_ms=latency_ms, request_size=request_size, response_size=response_size, ) severity = "INFO" message = f"HTTP {method} {url}" if error: severity = "ERROR" message = f"HTTP {method} {url} failed: {error}" elif status and status >= 400: severity = "WARNING" if status < 500 else "ERROR" message = f"HTTP {method} {url} returned {status}" extra: dict[str, Any] = { "http_request": http_request, "retry_attempt": retry_attempt, } if error: extra["error"] = create_error_log( error_type="HTTPError", error_message=error, ) log_entry = StructuredLogEntry( severity=severity, message=message, client_name=self.client_name, request_id=self._current_request_id, **extra ) if severity == "ERROR": demisto.error(log_entry.to_json()) elif severity == "WARNING": demisto.debug(log_entry.to_json()) else: if self.diagnostic_mode: demisto.debug(log_entry.to_json()) def log_metrics_summary(self) -> None: """Log a summary of execution metrics for analysis. This creates a structured log entry with aggregated metrics that can be used for monitoring and alerting in Google Cloud. """ if not self._performance["request_times"]: return times = self._performance["request_times"] metrics = { "total_requests": len(times), "avg_latency_ms": sum(times) / len(times), "min_latency_ms": min(times), "max_latency_ms": max(times), "p50_latency_ms": sorted(times)[len(times) // 2] if times else 0, "p95_latency_ms": sorted(times)[int(len(times) * 0.95)] if len(times) >= 20 else max(times), "error_count": len(self._errors), "retry_count": sum(1 for t in self._traces if t.retry_attempt > 0), } log_entry = StructuredLogEntry( severity="INFO", message="Request metrics summary", client_name=self.client_name, request_id=self._current_request_id, metrics=metrics, labels={"metric_type": "summary"}, ) demisto.info(log_entry.to_json()) def trace_request( self, method: str, url: str, headers: dict[str, str], params: dict[str, Any], body: Any | None = None, retry_attempt: int = 0, ) -> RequestTrace: """Create a trace record for an HTTP request. Args: method: HTTP method. url: Request URL. headers: Request headers (sensitive values will be redacted). params: Query parameters. body: Request body. retry_attempt: Current retry attempt number. Returns: A RequestTrace instance for tracking the request. """ # Generate a new request ID for this request if retry_attempt == 0: self.new_request_id() # Redact sensitive headers for security safe_headers = headers.copy() sensitive_keys = ["Authorization", "X-API-Key", "X-Auth-Token", "API-Key", "Api-Key", "apikey"] for key in list(safe_headers.keys()): if key.lower() in [k.lower() for k in sensitive_keys]: safe_headers[key] = "***REDACTED***" trace = RequestTrace( method=method, url=url, headers=safe_headers, params=params.copy(), body=body, timestamp=_now(), retry_attempt=retry_attempt, ) if self.diagnostic_mode: self._traces.append(trace) # Limit trace history to prevent memory issues if len(self._traces) > 1000: self._traces.pop(0) return trace def trace_response( self, trace: RequestTrace, status: int, headers: dict[str, str], body: Any, elapsed_ms: float, ) -> None: trace.response_status = status trace.response_headers = headers.copy() trace.response_body = body trace.elapsed_ms = elapsed_ms if self.diagnostic_mode: self._performance["request_times"].append(elapsed_ms) # Log with structured format for Google Cloud self.log_http_request( method=trace.method, url=trace.url, status=status, latency_ms=elapsed_ms, response_size=len(str(body)) if body else 0, retry_attempt=trace.retry_attempt, ) def trace_error( self, trace: RequestTrace, error: str, elapsed_ms: float | None = None, ) -> None: trace.error = error if elapsed_ms: trace.elapsed_ms = elapsed_ms if self.diagnostic_mode: # Log with structured format for Google Cloud self.log_http_request( method=trace.method, url=trace.url, status=trace.response_status, latency_ms=elapsed_ms, retry_attempt=trace.retry_attempt, error=error, ) def get_diagnostic_report( self, configuration: dict[str, Any], state_snapshots: list[dict[str, Any]] | None = None, ) -> DiagnosticReport: # Calculate performance metrics perf_metrics: dict[str, Any] = {} if self._performance["request_times"]: times = self._performance["request_times"] perf_metrics["avg_request_time_ms"] = sum(times) / len(times) perf_metrics["min_request_time_ms"] = min(times) perf_metrics["max_request_time_ms"] = max(times) perf_metrics["total_requests"] = len(times) # Generate recommendations recommendations: list[str] = [] if self._errors: error_types: dict[str, int] = {} for err in self._errors: err_type = err.get("context", {}).get("error_type", "unknown") error_types[err_type] = error_types.get(err_type, 0) + 1 if error_types.get("auth", 0) > 0: recommendations.append("Authentication errors detected. Check credentials and token expiration.") if error_types.get("rate_limit", 0) > 0: recommendations.append("Rate limit errors detected. Consider increasing rate_limit.rate_per_second.") if error_types.get("timeout", 0) > 0: recommendations.append("Timeout errors detected. Consider increasing timeout settings.") if error_types.get("network", 0) > 0: recommendations.append("Network errors detected. Check connectivity and proxy settings.") if perf_metrics.get("avg_request_time_ms", 0) > 5000: recommendations.append("Slow request times detected. Consider optimizing API calls or increasing timeouts.") if self._traces: retry_count = sum(1 for t in self._traces if t.retry_attempt > 0) if retry_count > len(self._traces) * 0.5: recommendations.append("High retry rate detected. Check API health and retry policy configuration.") return DiagnosticReport( content_item_name=self.client_name, configuration=configuration, request_traces=self._traces.copy(), state_snapshots=state_snapshots or [], performance_metrics=perf_metrics, errors=self._errors.copy(), recommendations=recommendations, timestamp=_now(), ) def _format(self, level: str, message: str, extra: dict[str, Any] | None) -> str: if not extra: return f"[ContentClient:{self.client_name}:{level}] {message}" try: extra_str = json.dumps(extra) except (TypeError, ValueError): extra_str = str(extra) return f"[ContentClient:{self.client_name}:{level}] {message} | extra={extra_str}" def _create_rate_limiter(policy: RateLimitPolicy | None) -> TokenBucketRateLimiter | None: """Create a rate limiter from policy if enabled. Factory function that encapsulates the rate limiter creation logic, improving readability and making the initialization more maintainable. Args: policy: Optional rate limit policy configuration. Returns: TokenBucketRateLimiter instance if policy is provided and enabled, None otherwise. """ if policy is None: return None if not policy.enabled: return None return TokenBucketRateLimiter(policy) class ContentClient: """Drop-in replacement for BaseClient with enhanced features. Fully compatible with BaseClient constructor and _http_request() method. Existing integrations can switch from BaseClient to ContentClient with zero code changes. Attributes: timeout: Default timeout for HTTP requests in seconds. execution_metrics: Metrics tracking for request outcomes. logger: Logger instance for diagnostic output. """ def __init__( self, base_url: str, verify: bool = True, proxy: bool = False, ok_codes: StatusCodesType = (), headers: HeadersType | None = None, auth: tuple[str, str] | None = None, timeout: float = 60.0, # New optional parameters (backward compatible): auth_handler: AuthHandler | None = None, retry_policy: RetryPolicy | None = None, rate_limiter: RateLimitPolicy | None = None, circuit_breaker: CircuitBreakerPolicy | None = None, diagnostic_mode: bool = False, client_name: str = "ContentClient", is_multithreaded: bool = True, reuse_client: bool = True, ) -> None: """Initialize ContentClient with BaseClient-compatible parameters. Args: base_url: The base URL for all API requests. verify: Whether to verify SSL certificates. proxy: Whether to use system proxy settings. ok_codes: Tuple of HTTP status codes considered successful. headers: Default headers to include in all requests. auth: Authentication credentials (tuple of username/password or AuthBase). timeout: Default timeout for requests in seconds. auth_handler: Optional custom authentication handler. retry_policy: Optional retry policy configuration. rate_limiter: Optional rate limiting policy. circuit_breaker: Optional circuit breaker policy. diagnostic_mode: Whether to enable diagnostic logging. WARNING: intended for short, supervised debugging sessions only. Request traces retain full response bodies in memory, so leaving this enabled on high-volume or long-running integrations may exhaust container memory. client_name: Name for logging identification. is_multithreaded: Whether to enable multithreading support. reuse_client: Whether to reuse the HTTP client across sync requests (default: True). When True, the client is kept open for better performance. When False, the client is closed after each sync request to avoid event loop issues. """ # Store BaseClient-compatible parameters self._base_url: str = base_url self._verify: bool = verify self._ok_codes: StatusCodesType = ok_codes self._headers: HeadersType = headers or {} self._auth: tuple[str, str] | None = auth self.timeout: float = timeout self._closed: bool = False self._is_multithreaded: bool = is_multithreaded self._reuse_client: bool = reuse_client # Handle proxy exactly like BaseClient if proxy: ensure_proxy_has_http_prefix() else: skip_proxy() if not verify: skip_cert_verification() # Enhanced features (optional, backward compatible) self._auth_handler: AuthHandler | None = auth_handler self._retry_policy: RetryPolicy = retry_policy if retry_policy is not None else RetryPolicy() # type: ignore[call-arg] self._rate_limiter: TokenBucketRateLimiter | None = _create_rate_limiter(rate_limiter) self._circuit_breaker: CircuitBreaker = CircuitBreaker( circuit_breaker if circuit_breaker is not None else CircuitBreakerPolicy() # type: ignore[call-arg] ) self._diagnostic_mode = diagnostic_mode # Execution metrics (like BaseClient) self.execution_metrics = ClientExecutionMetrics() # Logger self.logger = ContentClientLogger(client_name, diagnostic_mode=diagnostic_mode) # httpx client for async operations - created lazily per event loop self._local_storage = threading.local() self._http2_available: bool = True # Will be set to False if HTTP/2 fails if self._is_multithreaded: support_multithreading() def _get_async_client(self) -> httpx.AsyncClient: """Get or create an async client for the current event loop. This method ensures that the httpx.AsyncClient is created for and used within the same event loop, preventing issues when request_sync creates new event loops. Returns: An httpx.AsyncClient instance bound to the current event loop. """ try: current_loop = asyncio.get_running_loop() except RuntimeError: current_loop = None # Check if we need to create a new client if ( not hasattr(self._local_storage, "client") or self._local_storage.client is None or getattr(self._local_storage, "client_event_loop", None) != current_loop ): # Close existing client if any if getattr(self._local_storage, "client", None) is not None: try: # Schedule close on the old loop if possible old_loop = getattr(self._local_storage, "client_event_loop", None) old_client = self._local_storage.client if old_loop is not None and not old_loop.is_closed(): old_loop.call_soon_threadsafe(lambda c=old_client: asyncio.create_task(c.aclose())) except Exception: pass # Best effort cleanup # Create new client for current loop try: if self._http2_available: self._local_storage.client = httpx.AsyncClient( base_url=self._base_url.rstrip("/"), timeout=self.timeout, headers={"User-Agent": DEFAULT_USER_AGENT}, verify=self._verify, http2=True, follow_redirects=True, ) except ImportError: self._http2_available = False self.logger.info("HTTP/2 dependencies missing, falling back to HTTP/1.1 transport") if not self._http2_available or getattr(self._local_storage, "client", None) is None: self._local_storage.client = httpx.AsyncClient( base_url=self._base_url.rstrip("/"), timeout=self.timeout, headers={"User-Agent": DEFAULT_USER_AGENT}, verify=self._verify, http2=False, follow_redirects=True, ) self._local_storage.client_event_loop = current_loop return self._local_storage.client async def aclose(self) -> None: """Close the async HTTP client and release resources.""" if self._closed: return self._closed = True if hasattr(self._local_storage, "client") and self._local_storage.client is not None: try: await self._local_storage.client.aclose() except Exception as e: self.logger.error(f"Error closing async client: {e}") finally: self._local_storage.client = None self._local_storage.client_event_loop = None def __enter__(self) -> "ContentClient": """Enter context manager.""" return self def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: """Exit context manager and close resources.""" self.close() async def __aenter__(self) -> "ContentClient": """Enter async context manager.""" return self async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: """Exit async context manager and close resources.""" await self.aclose() def close(self) -> None: """Close the client synchronously. Note: If called from within an async context, use aclose() instead. This method handles various event loop scenarios gracefully. """ if self._closed: return try: # Try to get the current event loop try: loop = asyncio.get_running_loop() # We're in an async context - schedule the close self.logger.warning("close() called from async context. Use 'await client.aclose()' instead.") # Schedule the close but don't wait - the caller should use aclose() loop.call_soon(lambda: asyncio.create_task(self.aclose())) return except RuntimeError: # No running loop - we can safely run synchronously pass # Try to use existing event loop or create new one try: loop = asyncio.get_event_loop() if loop.is_closed(): raise RuntimeError("Event loop is closed") loop.run_until_complete(self.aclose()) except RuntimeError: # No event loop or it's closed, create a new one asyncio.run(self.aclose()) except Exception as e: self.logger.error(f"Error closing client: {e}") # Mark as closed even if there was an error to prevent repeated attempts self._closed = True async def _request( self, method: str, url_suffix: str = "", full_url: str | None = None, headers: HeadersType | None = None, auth: tuple[str, str] | None = None, json_data: JsonType | None = None, params: ParamsType | None = None, data: Any | None = None, files: dict[str, Any] | None = None, timeout: float | None = None, resp_type: str = "json", ok_codes: StatusCodesType | None = None, return_empty_response: bool = False, retries: int = 0, status_list_to_retry: list[int] | None = None, backoff_factor: float = 5.0, backoff_jitter: float = 0.0, raise_on_redirect: bool = False, raise_on_status: bool = False, error_handler: Callable[[httpx.Response], None] | None = None, empty_valid_codes: list[int] | None = None, params_parser: Callable[[ParamsType], ParamsType] | None = None, with_metrics: bool = False, **kwargs: Any, ) -> httpx.Response: """Execute an async HTTP request with retry and error handling. Args: method: HTTP method (GET, POST, PUT, PATCH, DELETE). url_suffix: URL path to append to base_url. full_url: Complete URL (overrides base_url + url_suffix). headers: Additional headers for this request. auth: Authentication tuple (username, password) for this request. json_data: JSON body data. params: Query parameters. data: Form data or raw body. files: Files to upload. timeout: Request timeout in seconds. resp_type: Expected response type ('json', 'text', 'content', 'response', 'xml'). ok_codes: HTTP status codes considered successful. return_empty_response: Whether to return empty dict for empty responses. retries: Number of retry attempts (BaseClient compatibility). status_list_to_retry: Status codes that trigger retry. backoff_factor: Backoff multiplier for retries. backoff_jitter: Jitter factor for retry delays. raise_on_redirect: Whether to raise on redirect responses. raise_on_status: Whether to raise on non-2xx responses. error_handler: Custom error handler callback. empty_valid_codes: Status codes that return empty response. params_parser: Custom parameter parser function. with_metrics: Whether to include metrics in response. **kwargs: Additional arguments. Returns: The HTTP response object. Raises: ContentClientCircuitOpenError: If circuit breaker is open. ContentClientRateLimitError: If rate limit is exceeded. ContentClientAuthenticationError: If authentication fails. ContentClientRetryError: If all retry attempts are exhausted. ContentClientError: For other request failures. """ if not self._circuit_breaker.can_execute(): raise ContentClientCircuitOpenError("Circuit breaker is open, refusing to send request") if self._rate_limiter: await self._rate_limiter.acquire() # Prepare request url = full_url if full_url else urljoin(self._base_url, url_suffix) req_headers = self._headers.copy() if headers: req_headers.update(headers) if "User-Agent" not in req_headers: req_headers["User-Agent"] = DEFAULT_USER_AGENT req_params = _ensure_dict(params) attempt = 0 last_error: Exception = Exception("Unknown error") trace: RequestTrace | None = None # Determine max attempts # If retries param is passed (BaseClient style), use it. Otherwise use retry_policy. max_attempts = retries + 1 if retries > 0 else self._retry_policy.max_attempts while attempt < max_attempts: attempt += 1 start = _now() try: client = self._get_async_client() http_request = client.build_request( method.upper(), url=url, params=req_params, json=json_data, data=data, files=files, headers=req_headers, timeout=timeout or self.timeout, ) if self._auth_handler: await self._auth_handler.on_request(self, http_request) elif auth: # Manual auth override if isinstance(auth, tuple): http_request.headers["Authorization"] = f"Basic {b64_encode(f'{auth[0]}:{auth[1]}')}" elif self._auth and isinstance(self._auth, tuple): http_request.headers["Authorization"] = f"Basic {b64_encode(f'{self._auth[0]}:{self._auth[1]}')}" # Trace request if in diagnostic mode if self._diagnostic_mode: full_req_url = str(http_request.url) trace = self.logger.trace_request( method=method.upper(), url=full_req_url, headers=dict(http_request.headers), params=req_params, body=json_data or data, retry_attempt=attempt - 1, ) response = await client.send(http_request) elapsed_ms = (_now() - start) * 1000 if self._diagnostic_mode and trace: try: response_body = response.json() if response.content else None except Exception: response_body = response.text[:1000] self.logger.trace_response( trace, status=response.status_code, headers=dict(response.headers), body=response_body, elapsed_ms=elapsed_ms, ) if response.status_code == 401 and self._auth_handler: should_retry = await self._auth_handler.on_auth_failure(self, response) if should_retry: continue # Check for retryable status codes # Use status_list_to_retry if provided, else use policy retry_codes = status_list_to_retry if status_list_to_retry else self._retry_policy.retryable_status_codes if response.status_code in retry_codes: raise httpx.HTTPStatusError("Retryable status", request=http_request, response=response) # Check for ok_codes # BaseClient logic: if ok_codes provided, check against it. # If not, check _ok_codes instance variable. Else check response.ok is_ok = False effective_ok_codes = ok_codes or self._ok_codes if effective_ok_codes: is_ok = response.status_code in effective_ok_codes else: is_ok = response.is_success if not is_ok: # Raise exception to trigger error handling response.raise_for_status() self.execution_metrics.success += 1 self._circuit_breaker.record_success() self.logger.debug( # noqa: PLE1205 "HTTP request completed", {"status": response.status_code, "elapsed": elapsed_ms, "endpoint": url}, ) return response except ContentClientError: # Re-raise ContentClientError (including subclasses) without wrapping raise except ( httpx.ConnectError, httpx.ReadTimeout, httpx.WriteTimeout, httpx.RemoteProtocolError, httpx.PoolTimeout, ) as exc: last_error = exc elapsed_ms = (_now() - start) * 1000 if self._diagnostic_mode and trace: self.logger.trace_error(trace, str(exc), elapsed_ms) should_retry = attempt < max_attempts if not should_retry: break retry_after = _parse_retry_after(getattr(exc, "response", None)) delay = self._retry_policy.next_delay(attempt, retry_after) self.execution_metrics.retry_error += 1 self.logger.debug( # noqa: PLE1205 "Retryable exception occurred", {"attempt": attempt, "delay": delay, "error": str(exc), "error_type": type(exc).__name__}, ) await anyio.sleep(delay) except httpx.HTTPStatusError as exc: last_error = exc elapsed_ms = (_now() - start) * 1000 if self._diagnostic_mode and trace: try: response_body = exc.response.json() if exc.response.content else None except Exception: response_body = exc.response.text[:1000] self.logger.trace_response( trace, status=exc.response.status_code, headers=dict(exc.response.headers), body=response_body, elapsed_ms=elapsed_ms, ) self.logger.trace_error(trace, f"HTTP {exc.response.status_code}: {exc.response.text[:200]}") # Map HTTP status codes to specific exception types if exc.response.status_code == 429: self.execution_metrics.quota_error += 1 self.logger.error("Rate limit error", {"status": 429, "error_type": "rate_limit"}) # noqa: PLE1205 should_retry = exc.response.status_code in (status_list_to_retry or self._retry_policy.retryable_status_codes) if should_retry and attempt < max_attempts: retry_after = _parse_retry_after(exc.response) delay = self._retry_policy.next_delay(attempt, retry_after) self.execution_metrics.retry_error += 1 await anyio.sleep(delay) continue self._circuit_breaker.record_failure() raise ContentClientRateLimitError(f"Rate limit exceeded: {exc.response.text}", response=exc.response) from exc elif exc.response.status_code in (401, 403): self.execution_metrics.auth_error += 1 self.logger.error( # noqa: PLE1205 "Authentication error", {"status": exc.response.status_code, "error_type": "auth"}, ) should_retry = exc.response.status_code in (status_list_to_retry or self._retry_policy.retryable_status_codes) if should_retry and attempt < max_attempts: retry_after = _parse_retry_after(exc.response) delay = self._retry_policy.next_delay(attempt, retry_after) self.execution_metrics.retry_error += 1 await anyio.sleep(delay) continue self._circuit_breaker.record_failure() raise ContentClientAuthenticationError( f"Authentication failed: {exc.response.text}", response=exc.response ) from exc else: self.execution_metrics.service_error += 1 self.logger.error( # noqa: PLE1205 "Service error", {"status": exc.response.status_code, "error_type": "service"}, ) should_retry = exc.response.status_code in (status_list_to_retry or self._retry_policy.retryable_status_codes) if should_retry and attempt < max_attempts: retry_after = _parse_retry_after(exc.response) delay = self._retry_policy.next_delay(attempt, retry_after) self.execution_metrics.retry_error += 1 await anyio.sleep(delay) continue self._circuit_breaker.record_failure() raise ContentClientError(f"Request failed: {exc.response.text}", response=exc.response) from exc except Exception as exc: elapsed_ms = (_now() - start) * 1000 if self._diagnostic_mode and trace: self.logger.trace_error(trace, str(exc), elapsed_ms) self.execution_metrics.general_error += 1 self._circuit_breaker.record_failure() self.logger.error( # noqa: PLE1205 "Non-retryable exception occurred", {"error": str(exc), "error_type": type(exc).__name__}, ) raise self._circuit_breaker.record_failure() last_response = getattr(last_error, "response", None) raise ContentClientRetryError(f"Exceeded retry attempts: {last_error}", response=last_response) def _http_request( self, method: str, url_suffix: str = "", full_url: str | None = None, headers: HeadersType | None = None, auth: tuple[str, str] | None = None, json_data: JsonType | None = None, params: ParamsType | None = None, data: Any | None = None, files: dict[str, Any] | None = None, timeout: float | None = None, resp_type: str = "json", ok_codes: StatusCodesType | None = None, return_empty_response: bool = False, retries: int = 0, status_list_to_retry: list[int] | None = None, backoff_factor: float = 5.0, backoff_jitter: float = 0.0, raise_on_redirect: bool = False, raise_on_status: bool = False, error_handler: Callable[[httpx.Response], None] | None = None, empty_valid_codes: list[int] | None = None, params_parser: Callable[[ParamsType], ParamsType] | None = None, with_metrics: bool = False, **kwargs: Any, ) -> Any: """Synchronous wrapper for _request to maintain BaseClient compatibility. This method provides the same interface as BaseClient._http_request(). See _request() for full parameter documentation. Returns: The processed response based on resp_type parameter. """ return self.request_sync( method=method, url_suffix=url_suffix, full_url=full_url, headers=headers, auth=auth, json_data=json_data, params=params, data=data, files=files, timeout=timeout, resp_type=resp_type, ok_codes=ok_codes, return_empty_response=return_empty_response, retries=retries, status_list_to_retry=status_list_to_retry, backoff_factor=backoff_factor, backoff_jitter=backoff_jitter, raise_on_redirect=raise_on_redirect, raise_on_status=raise_on_status, error_handler=error_handler, empty_valid_codes=empty_valid_codes, params_parser=params_parser, with_metrics=with_metrics, **kwargs, ) def request_sync(self, *args: Any, **kwargs: Any) -> Any: """Execute a request synchronously. This method wraps the async _request method for synchronous usage. It handles response processing based on the resp_type parameter. The implementation creates a new event loop for each synchronous request to avoid issues with event loop reuse. The httpx.AsyncClient is automatically recreated for each new event loop. Args: *args: Positional arguments passed to _request. **kwargs: Keyword arguments passed to _request. Returns: The processed response based on resp_type (json, text, content, response, xml). """ async def _do_request() -> Any: try: response = await self._request(*args, **kwargs) # Handle response processing (json, text, content) similar to BaseClient resp_type = kwargs.get("resp_type", "json") return_empty_response = kwargs.get("return_empty_response", False) empty_valid_codes = kwargs.get("empty_valid_codes") if return_empty_response and empty_valid_codes and response.status_code in empty_valid_codes: return {} if resp_type == "json": try: return response.json() except json.JSONDecodeError: if not response.content: return {} raise elif resp_type == "text": return response.text elif resp_type == "content": return response.content elif resp_type == "response": return response elif resp_type == "xml": return response.text # Default fallback - try JSON try: return response.json() except json.JSONDecodeError: # If JSON parsing fails, return the response object return response finally: # Clean up the client after each sync request only if reuse_client is False # When reuse_client is True (default), the client is kept open for better performance if not self._reuse_client and hasattr(self._local_storage, "client") and self._local_storage.client is not None: try: await self._local_storage.client.aclose() except Exception: pass self._local_storage.client = None self._local_storage.client_event_loop = None # Check if we're already in an async context try: asyncio.get_running_loop() # We're in an async context, need to use a thread pool with concurrent.futures.ThreadPoolExecutor() as executor: future = executor.submit(asyncio.run, _do_request()) return future.result() except RuntimeError: # No running event loop, safe to use asyncio.run return asyncio.run(_do_request()) # Standard HTTP verb helpers def get(self, url_suffix: str, params: ParamsType | None = None, **kwargs: Any) -> Any: """Execute a GET request. Args: url_suffix: URL path to append to base_url. params: Query parameters. **kwargs: Additional arguments passed to _http_request. Returns: The response object or processed response data. """ if "resp_type" not in kwargs: kwargs["resp_type"] = "response" return self._http_request("GET", url_suffix, params=params, **kwargs) def post(self, url_suffix: str, json_data: JsonType | None = None, **kwargs: Any) -> Any: """Execute a POST request. Args: url_suffix: URL path to append to base_url. json_data: JSON body data. **kwargs: Additional arguments passed to _http_request. Returns: The response object or processed response data. """ if "resp_type" not in kwargs: kwargs["resp_type"] = "response" return self._http_request("POST", url_suffix, json_data=json_data, **kwargs) def put(self, url_suffix: str, json_data: JsonType | None = None, **kwargs: Any) -> Any: """Execute a PUT request. Args: url_suffix: URL path to append to base_url. json_data: JSON body data. **kwargs: Additional arguments passed to _http_request. Returns: The response object or processed response data. """ if "resp_type" not in kwargs: kwargs["resp_type"] = "response" return self._http_request("PUT", url_suffix, json_data=json_data, **kwargs) def patch(self, url_suffix: str, json_data: JsonType | None = None, **kwargs: Any) -> Any: """Execute a PATCH request. Args: url_suffix: URL path to append to base_url. json_data: JSON body data. **kwargs: Additional arguments passed to _http_request. Returns: The response object or processed response data. """ if "resp_type" not in kwargs: kwargs["resp_type"] = "response" return self._http_request("PATCH", url_suffix, json_data=json_data, **kwargs) def delete(self, url_suffix: str, **kwargs: Any) -> Any: """Execute a DELETE request. Args: url_suffix: URL path to append to base_url. **kwargs: Additional arguments passed to _http_request. Returns: The response object or processed response data. """ if "resp_type" not in kwargs: kwargs["resp_type"] = "response" return self._http_request("DELETE", url_suffix, **kwargs) @property def metrics(self) -> ClientExecutionMetrics: """Get the execution metrics for this client. Returns: The ClientExecutionMetrics instance tracking request outcomes. """ return self.execution_metrics def get_diagnostic_report(self) -> DiagnosticReport: """Generate a diagnostic report for troubleshooting. Returns: A DiagnosticReport containing traces, metrics, and recommendations. """ return self.logger.get_diagnostic_report( configuration={"name": self.logger.client_name, "base_url": self._base_url, "timeout": self.timeout} ) def diagnose_error(self, error: Exception) -> dict[str, str]: """Diagnose an error and provide a solution recommendation. Args: error: The exception to diagnose. Returns: A dictionary with 'issue' and 'solution' keys. """ if isinstance(error, ContentClientAuthenticationError): return {"issue": "Authentication failed", "solution": "Check credentials and token expiration."} if isinstance(error, ContentClientRateLimitError): return {"issue": "Rate limit exceeded", "solution": "Increase retry delay or request quota."} if isinstance(error, ContentClientTimeoutError): return {"issue": "Execution timeout", "solution": "Increase timeout settings or reduce batch size."} if isinstance(error, ContentClientCircuitOpenError): return {"issue": "Circuit breaker is open", "solution": "Wait for recovery timeout."} if isinstance(error, ContentClientRetryError): return {"issue": "All retry attempts exhausted", "solution": "Check API availability and retry policy."} if isinstance(error, ContentClientConfigurationError): return {"issue": "Configuration error", "solution": "Check integration parameters."} return {"issue": "Unexpected error", "solution": f"Check logs for details: {str(error)}"} def health_check(self) -> dict[str, Any]: """Perform a health check on the client. Returns: A dictionary with 'status', 'configuration_valid', 'warnings', and 'metrics'. """ status: str = "healthy" warnings: list[str] = [] if self.execution_metrics.auth_error > 0: status = "degraded" warnings.append("Authentication errors detected") if self.execution_metrics.general_error > 0: status = "degraded" warnings.append("General errors detected") if self.execution_metrics.quota_error > 0: status = "degraded" warnings.append("Rate limit errors detected") return {"status": status, "configuration_valid": True, "warnings": warnings, "metrics": self.execution_metrics}