Skip to content

Failed upload

FailedUpload dataclass #

FailedUpload(
    item: T,
    error_message: str,
    error_type: str,
    timestamp: Optional[datetime] = now(),
    exception: Optional[Exception] = None,
    trace_id: Optional[str] = None,
    stage: Optional[str] = None,
    http_status: Optional[int] = None,
)

Bases: Generic[T]

Represents a failed upload with the item and error details.

Attributes:

Name Type Description
item T

The item that failed to upload.

error_message str

The error message describing the failure reason.

error_type str

The type of the exception (e.g., "RapidataError").

timestamp Optional[datetime]

Optional timestamp when the failure occurred.

exception Optional[Exception]

Optional original exception for richer error context.

trace_id Optional[str]

Optional backend trace ID, when the failure originated from a RapidataError whose response carried one — either as a traceId body member or an x-trace-id header. Used to correlate an SDK-side failure with the backend trace that produced it.

stage Optional[str]

Optional ingestion stage that failed, for remote-URL assets (e.g. "download", "content_type", "timeout", "size", "internal"). Only "internal" indicates a Rapidata-side fault; every other stage is caller-actionable.

http_status Optional[int]

Optional origin-server HTTP status, when the failure was an HTTP response from the asset host (e.g. a 403 on the URL). Distinct from any status of the Rapidata API call itself.

from_exception classmethod #

from_exception(
    item: T, exception: Exception | None
) -> FailedUpload[T]

Create a FailedUpload from an item and exception.

For RapidataError exceptions, extracts the clean API error reason, the backend trace ID, and the structured ingestion stage / origin HTTP status (when present in the error response). For other exceptions, uses the string representation.

Parameters:

Name Type Description Default
item T

The item that failed to upload.

required
exception Exception | None

The exception that occurred.

required

Returns:

Type Description
FailedUpload[T]

FailedUpload instance with error details extracted from the exception.

Source code in src/rapidata/rapidata_client/exceptions/failed_upload.py
@classmethod
def from_exception(cls, item: T, exception: Exception | None) -> FailedUpload[T]:
    """
    Create a FailedUpload from an item and exception.

    For RapidataError exceptions, extracts the clean API error reason, the
    backend trace ID, and the structured ingestion stage / origin HTTP
    status (when present in the error response). For other exceptions, uses
    the string representation.

    Args:
        item: The item that failed to upload.
        exception: The exception that occurred.

    Returns:
        FailedUpload instance with error details extracted from the exception.
    """
    if exception is None:
        return cls(
            item=item,
            error_message="Unknown error",
            error_type="Unknown",
            exception=None,
        )

    from rapidata.rapidata_client.exceptions.rapidata_error import RapidataError

    error_type = type(exception).__name__
    trace_id: Optional[str] = None
    stage: Optional[str] = None
    http_status: Optional[int] = None

    if isinstance(exception, RapidataError):
        error_message = exception.get_reason()
        trace_id = exception.trace_id
        if isinstance(exception.details, dict):
            # stage / upstreamHttpStatus are ProblemDetails extension
            # members the asset service adds for remote-URL ingestion
            # failures; they serialize flat alongside title / traceId.
            raw_stage = exception.details.get("stage")
            if isinstance(raw_stage, str) and raw_stage:
                stage = raw_stage

            raw_http_status = exception.details.get("upstreamHttpStatus")
            if isinstance(raw_http_status, int):
                http_status = raw_http_status
    else:
        error_message = str(exception)

    return cls(
        item=item,
        error_message=error_message,
        error_type=error_type,
        exception=exception,
        trace_id=trace_id,
        stage=stage,
        http_status=http_status,
    )

format_error_details #

format_error_details() -> str

Format error details for logging or display.

Returns:

Type Description
str

Formatted string with all error details including timestamp.

Source code in src/rapidata/rapidata_client/exceptions/failed_upload.py
def format_error_details(self) -> str:
    """
    Format error details for logging or display.

    Returns:
        Formatted string with all error details including timestamp.
    """
    details = [
        f"Item: {self.item}",
        f"Error Type: {self.error_type}",
        f"Error Message: {self.error_message}",
    ]

    if self.stage:
        details.append(f"Stage: {self.stage}")

    if self.http_status is not None:
        details.append(f"HTTP Status: {self.http_status}")

    if self.trace_id:
        details.append(f"Trace Id: {self.trace_id}")

    if self.timestamp:
        details.append(f"Timestamp: {self.timestamp.isoformat()}")

    return "\n".join(details)