Skip to content

exceptions

exceptions

CloudAmplifier_CRUD_Error

CloudAmplifier_CRUD_Error(
    operation: str,
    entity_id: str | None = None,
    res: ResponseGetData | None = None,
    message: str | None = None,
    **kwargs
)

Bases: RouteError

Raised when Cloud Amplifier integration create, update, or delete operations fail.

This exception is used for failures during integration creation, modification, deletion, warehouse updates, and federated dataset conversion operations.

Source code in src/crew_dcs/routes/cloud_amplifier/exceptions.py
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def __init__(
    self,
    operation: str,
    entity_id: str | None = None,
    res: rgd.ResponseGetData | None = None,
    message: str | None = None,
    **kwargs,
):
    if not message:
        if entity_id:
            message = (
                f"Cloud Amplifier {operation} failed for integration {entity_id}"
            )
        else:
            message = f"Cloud Amplifier {operation} operation failed"

    super().__init__(
        message=message,
        entity_id=entity_id,
        res=res,
        **kwargs,
    )

CloudAmplifier_GET_Error

CloudAmplifier_GET_Error(
    entity_id: str | None = None,
    res: ResponseGetData | None = None,
    message: str | None = None,
    **kwargs
)

Bases: RouteError

Raised when Cloud Amplifier integration retrieval operations fail.

This exception is used for failures during GET operations on integrations, warehouses, databases, schemas, tables, and metadata retrieval.

Source code in src/crew_dcs/routes/cloud_amplifier/exceptions.py
27
28
29
30
31
32
33
34
35
def __init__(
    self,
    entity_id: str | None = None,
    res: rgd.ResponseGetData | None = None,
    message: str | None = None,
    **kwargs,
):
    if not message and entity_id:
        message = f"Failed to retrieve Cloud Amplifier integration {entity_id}"

SearchCloudAmplifierNotFoundError

SearchCloudAmplifierNotFoundError(
    search_criteria: str,
    res: ResponseGetData | None = None,
    **kwargs
)

Bases: RouteError

Raised when Cloud Amplifier integration search operations return no results.

Source code in src/crew_dcs/routes/cloud_amplifier/exceptions.py
42
43
44
45
46
47
48
49
50
51
52
53
def __init__(
    self,
    search_criteria: str,
    res: rgd.ResponseGetData | None = None,
    **kwargs,
):
    message = f"No Cloud Amplifier resources found matching: {search_criteria}"
    super().__init__(
        message=message,
        res=res,
        **kwargs,
    )

SearchCloudAmplifierNotFoundError

SearchCloudAmplifierNotFoundError(
    search_criteria: str,
    res: ResponseGetData | None = None,
    **kwargs
)

Bases: RouteError

Raised when Cloud Amplifier integration search operations return no results.

This exception is used when searching for specific integrations, warehouses, databases, schemas, or tables that don't exist or when search criteria match nothing.

Source code in src/crew_dcs/routes/cloud_amplifier/exceptions.py
64
65
66
67
68
69
70
71
72
73
74
75
def __init__(
    self,
    search_criteria: str,
    res: rgd.ResponseGetData | None = None,
    **kwargs,
):
    message = f"No Cloud Amplifier resources found matching: {search_criteria}"
    super().__init__(
        message=message,
        res=res,
        **kwargs,
    )