Rapidata api client

RapidataApiClient

RapidataApiClient(*args, **kwargs)

Bases: ApiClient

Custom API client that wraps errors in RapidataError.

Source code in src/rapidata/rapidata_client/api/rapidata_api_client.py

def __init__(self, *args, **kwargs):
    super().__init__(*args, **kwargs)
    self.id_generator = RandomIdGenerator()

response_deserialize

response_deserialize(
    response_data: RESTResponse,
    response_types_map: Optional[
        dict[str, ApiResponseT]
    ] = None,
) -> ApiResponse[ApiResponseT]

Override the response_deserialize method to catch and convert exceptions.

Source code in src/rapidata/rapidata_client/api/rapidata_api_client.py

def response_deserialize(
    self,
    response_data: rest.RESTResponse,
    response_types_map: Optional[dict[str, ApiResponseT]] = None,
) -> ApiResponse[ApiResponseT]:
    """Override the response_deserialize method to catch and convert exceptions."""
    try:
        return super().response_deserialize(response_data, response_types_map)
    except ApiException as e:
        status_code = getattr(e, "status", None)
        message = str(e)
        details = None

        # Extract more detailed error message from response body if available
        if hasattr(e, "body") and e.body:
            try:
                body_json = json.loads(e.body)
                if isinstance(body_json, dict):
                    if "message" in body_json:
                        message = body_json["message"]
                    elif "error" in body_json:
                        message = body_json["error"]

                    # Store the full error details for debugging
                    details = body_json
            except (json.JSONDecodeError, AttributeError):
                # If we can't parse the body as JSON, use the original message
                pass

        error_formatted = RapidataError(
            status_code=status_code,
            message=message,
            original_exception=e,
            details=details,
            trace_id=_trace_id_from_headers(getattr(e, "headers", None)),
        )

        # Only log error if not suppressed
        if not _should_suppress_error_logging():
            logger.error("Error: %s", error_formatted)
        else:
            logger.debug("Suppressed Error: %s", error_formatted)

        raise error_formatted from None

mark_sdk_outdated

mark_sdk_outdated(
    current_version: str, latest_version: str
) -> None

Record that the installed SDK is behind the latest release.

Source code in src/rapidata/rapidata_client/api/rapidata_api_client.py

def mark_sdk_outdated(current_version: str, latest_version: str) -> None:
    """Record that the installed SDK is behind the latest release."""
    global _sdk_outdated_info
    _sdk_outdated_info = {
        "current": current_version,
        "latest": latest_version,
    }

format_outdated_sdk_note

format_outdated_sdk_note() -> Optional[str]

Build the human-readable outdated-SDK note, or None if not outdated.

Used by RapidataError and LazyValidatedModel to append the same hint to their error messages when the installed SDK is behind the latest release.

Source code in src/rapidata/rapidata_client/api/rapidata_api_client.py

def format_outdated_sdk_note() -> Optional[str]:
    """Build the human-readable outdated-SDK note, or None if not outdated.

    Used by RapidataError and LazyValidatedModel to append the same hint to
    their error messages when the installed SDK is behind the latest release.
    """
    info = _sdk_outdated_info
    if not info:
        return None
    current = info.get("current")
    latest = info.get("latest")
    return (
        f"Note: Your Rapidata SDK is outdated (installed: {current}, "
        f"latest: {latest}). This error may be caused by the SDK being "
        f"out of sync with the API - please upgrade and try again."
    )

suppress_rapidata_error_logging

suppress_rapidata_error_logging()

Context manager to suppress error logging for RapidataApiClient calls.

Source code in src/rapidata/rapidata_client/api/rapidata_api_client.py

@contextmanager
def suppress_rapidata_error_logging():
    """Context manager to suppress error logging for RapidataApiClient calls."""
    old_value = getattr(_thread_local, "suppress_error_logging", False)
    _thread_local.suppress_error_logging = True
    try:
        yield
    finally:
        _thread_local.suppress_error_logging = old_value

optional_api_call

optional_api_call(description: str)

Mark a block as non-critical / best-effort.

Inside the block

• RapidataApiClient errors are logged at DEBUG instead of ERROR (via suppress_rapidata_error_logging).
• Any exception that escapes the block is caught and logged at DEBUG as well. The caller never sees it.

Use for best-effort calls like version checks, telemetry, or feature-flag lookups where failure must not impact the user.

Source code in src/rapidata/rapidata_client/api/rapidata_api_client.py

@contextmanager
def optional_api_call(description: str):
    """Mark a block as non-critical / best-effort.

    Inside the block:
      - RapidataApiClient errors are logged at DEBUG instead of ERROR
        (via suppress_rapidata_error_logging).
      - Any exception that escapes the block is caught and logged at
        DEBUG as well. The caller never sees it.

    Use for best-effort calls like version checks, telemetry, or
    feature-flag lookups where failure must not impact the user.
    """
    with suppress_rapidata_error_logging():
        try:
            yield
        except Exception as e:
            logger.debug("Optional call '%s' failed: %s", description, e)