Rapidata api client

RapidataError

RapidataError(
    status_code: Optional[int] = None,
    message: str | None = None,
    original_exception: Exception | None = None,
    details: Any = None,
)

Bases: Exception

Custom error class for Rapidata API errors.

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

def __init__(
    self,
    status_code: Optional[int] = None,
    message: str | None = None,
    original_exception: Exception | None = None,
    details: Any = None,
):
    self.status_code = status_code
    self.message = message
    self.original_exception = original_exception
    self.details = details

    # Create a nice error message
    error_msg = "Rapidata API Error"
    if status_code:
        error_msg += f" ({status_code})"
    if message:
        error_msg += f": {message}"

    super().__init__(error_msg)

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,
        )

        # 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

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