Skip to content

API Reference: CrossClient

Source code in src/crosscontract/crossclient/crossclient.py 
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
class CrossClient:
    def __init__(
        self,
        username: str,
        password: str,
        base_url: str = DEFAULT_URL,
        verify: bool = True,
    ) -> None:
        """Initialize the client with authentication.

        Args:
            username (str): The username for authentication.
            password (str): The password for authentication.
            base_url (str): If provided, use this domain instead of the default
                DEFAULT_URL.
                The domain must include the protocol (e.g., http:// or https://).
                Example: "http://example.com".
                Defaults to DEFAULT_URL: "https://backend.sweet-cross.ch".
                Trailing slashes are stripped internally.
            verify (bool): Whether to verify SSL certificates.
                Defaults to True.

        Returns:
            CrossClient: An instance of the authenticated client.
        """
        self._base_url = base_url.rstrip("/")  # Ensure no trailing slash
        self._username = username
        self._password = password
        self._verify = verify
        self._token = None

        # Create the client
        timeout = httpx.Timeout(10.0, connect=30.0, read=60.0, write=None)
        limits = httpx.Limits(max_connections=5, max_keepalive_connections=5)
        self._client = httpx.Client(
            base_url=self._base_url,
            verify=verify,
            timeout=timeout,
            limits=limits,
        )
        self._is_closed = False

        # ---- include services ----
        self.contracts: ContractService = ContractService(client=self)

        # authenticate upon initialization
        self.authenticate()

        # Register cleanup on interpreter shutdown
        atexit.register(self.close)

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.close()

    def __del__(self):
        # Best-effort cleanup during garbage collection
        try:
            self.close()
        except Exception:
            pass

    def __repr__(self):
        return f"CrossClient(base_url={self._base_url}, username={self._username})"

    def close(self):
        """Close the HTTPX client."""
        self._client.close()
        self._is_closed = True

    def authenticate(self) -> str:
        """Authenticate with the server and retrieve an access token.

        Returns:
            str: The authentication token.
        """
        response = self._client.post(
            "/user/auth/login",
            data={"username": self._username, "password": self._password},
        )
        response.raise_for_status()  # Raise an error for bad responses
        token = response.json().get("access_token", "")
        self._token = token
        self._client.headers["Authorization"] = f"Bearer {self._token}"
        return token

    def request(self, method: str, endpoint: str, **kwargs: Any) -> httpx.Response:
        """Send an HTTP request to the specified endpoint.

        Args:
            method (str): The HTTP method (e.g., 'GET', 'POST').
            endpoint (str): The API endpoint to send the request to.
            **kwargs: Additional arguments to pass to the request.

        Returns:
            httpx.Response: The response from the server.
        """
        if self._is_closed:
            raise RuntimeError(
                "Attempted to make a request with a closed CrossClient. Ensure you "
                "are performing all operations within the 'with' context block."
            )
        if not self._token:
            self.authenticate()
        response = self._client.request(method, endpoint, **kwargs)

        # try to get a new token if unauthorized
        if response.status_code == 401:
            # Token expired: Refresh and retry
            self.authenticate()

            # Re-issue the request with the new header (handled by self._client update)
            # We must recreate the request to pick up the new headers from the
            # client state
            response = self._client.request(method, endpoint, **kwargs)
        return response

    def post(self, endpoint: str, json: dict | None = None, **kwargs) -> httpx.Response:
        """Send a POST request to the specified endpoint."""
        return self.request("POST", endpoint, json=json, **kwargs)  # pragma: no cover

    def delete(self, endpoint: str, **kwargs) -> httpx.Response:
        """Send a DELETE request to the specified endpoint."""
        return self.request("DELETE", endpoint, **kwargs)  # pragma: no cover

    def get(self, endpoint: str, **kwargs) -> httpx.Response:
        """Send a GET request to the specified endpoint."""
        return self.request("GET", endpoint, **kwargs)  # pragma: no cover

    def patch(
        self, endpoint: str, json: dict | None = None, **kwargs
    ) -> httpx.Response:
        """Send a PATCH request to the specified endpoint."""
        return self.request("PATCH", endpoint, json=json, **kwargs)  # pragma: no cover

__init__(username, password, base_url=DEFAULT_URL, verify=True)

Initialize the client with authentication.

Parameters:

Name Type Description Default
username str

The username for authentication.

 required
password str

The password for authentication.

 required
base_url str

If provided, use this domain instead of the default DEFAULT_URL. The domain must include the protocol (e.g., http:// or https://). Example: "http://example.com". Defaults to DEFAULT_URL: "https://backend.sweet-cross.ch". Trailing slashes are stripped internally.

 DEFAULT_URL
verify bool

Whether to verify SSL certificates. Defaults to True.

 True

Returns:

Name Type Description
CrossClient None

An instance of the authenticated client.
Source code in src/crosscontract/crossclient/crossclient.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def __init__(
    self,
    username: str,
    password: str,
    base_url: str = DEFAULT_URL,
    verify: bool = True,
) -> None:
    """Initialize the client with authentication.

    Args:
        username (str): The username for authentication.
        password (str): The password for authentication.
        base_url (str): If provided, use this domain instead of the default
            DEFAULT_URL.
            The domain must include the protocol (e.g., http:// or https://).
            Example: "http://example.com".
            Defaults to DEFAULT_URL: "https://backend.sweet-cross.ch".
            Trailing slashes are stripped internally.
        verify (bool): Whether to verify SSL certificates.
            Defaults to True.

    Returns:
        CrossClient: An instance of the authenticated client.
    """
    self._base_url = base_url.rstrip("/")  # Ensure no trailing slash
    self._username = username
    self._password = password
    self._verify = verify
    self._token = None

    # Create the client
    timeout = httpx.Timeout(10.0, connect=30.0, read=60.0, write=None)
    limits = httpx.Limits(max_connections=5, max_keepalive_connections=5)
    self._client = httpx.Client(
        base_url=self._base_url,
        verify=verify,
        timeout=timeout,
        limits=limits,
    )
    self._is_closed = False

    # ---- include services ----
    self.contracts: ContractService = ContractService(client=self)

    # authenticate upon initialization
    self.authenticate()

    # Register cleanup on interpreter shutdown
    atexit.register(self.close)

authenticate()

Authenticate with the server and retrieve an access token.

Returns:

Name Type Description
str str

The authentication token.
Source code in src/crosscontract/crossclient/crossclient.py 
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def authenticate(self) -> str:
    """Authenticate with the server and retrieve an access token.

    Returns:
        str: The authentication token.
    """
    response = self._client.post(
        "/user/auth/login",
        data={"username": self._username, "password": self._password},
    )
    response.raise_for_status()  # Raise an error for bad responses
    token = response.json().get("access_token", "")
    self._token = token
    self._client.headers["Authorization"] = f"Bearer {self._token}"
    return token

close()

Close the HTTPX client.

Source code in src/crosscontract/crossclient/crossclient.py 
78
79
80
81
def close(self):
    """Close the HTTPX client."""
    self._client.close()
    self._is_closed = True

delete(endpoint, **kwargs)

Send a DELETE request to the specified endpoint.

Source code in src/crosscontract/crossclient/crossclient.py 
134
135
136
def delete(self, endpoint: str, **kwargs) -> httpx.Response:
    """Send a DELETE request to the specified endpoint."""
    return self.request("DELETE", endpoint, **kwargs)  # pragma: no cover

get(endpoint, **kwargs)

Send a GET request to the specified endpoint.

Source code in src/crosscontract/crossclient/crossclient.py 
138
139
140
def get(self, endpoint: str, **kwargs) -> httpx.Response:
    """Send a GET request to the specified endpoint."""
    return self.request("GET", endpoint, **kwargs)  # pragma: no cover

patch(endpoint, json=None, **kwargs)

Send a PATCH request to the specified endpoint.

Source code in src/crosscontract/crossclient/crossclient.py 
142
143
144
145
146
def patch(
    self, endpoint: str, json: dict | None = None, **kwargs
) -> httpx.Response:
    """Send a PATCH request to the specified endpoint."""
    return self.request("PATCH", endpoint, json=json, **kwargs)  # pragma: no cover

post(endpoint, json=None, **kwargs)

Send a POST request to the specified endpoint.

Source code in src/crosscontract/crossclient/crossclient.py 
130
131
132
def post(self, endpoint: str, json: dict | None = None, **kwargs) -> httpx.Response:
    """Send a POST request to the specified endpoint."""
    return self.request("POST", endpoint, json=json, **kwargs)  # pragma: no cover

request(method, endpoint, **kwargs)

Send an HTTP request to the specified endpoint.

Parameters:

Name Type Description Default
method str

The HTTP method (e.g., 'GET', 'POST').

 required
endpoint str

The API endpoint to send the request to.

 required
**kwargs Any

Additional arguments to pass to the request.

 {}

Returns:

Type Description
Response

httpx.Response: The response from the server.
Source code in src/crosscontract/crossclient/crossclient.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def request(self, method: str, endpoint: str, **kwargs: Any) -> httpx.Response:
    """Send an HTTP request to the specified endpoint.

    Args:
        method (str): The HTTP method (e.g., 'GET', 'POST').
        endpoint (str): The API endpoint to send the request to.
        **kwargs: Additional arguments to pass to the request.

    Returns:
        httpx.Response: The response from the server.
    """
    if self._is_closed:
        raise RuntimeError(
            "Attempted to make a request with a closed CrossClient. Ensure you "
            "are performing all operations within the 'with' context block."
        )
    if not self._token:
        self.authenticate()
    response = self._client.request(method, endpoint, **kwargs)

    # try to get a new token if unauthorized
    if response.status_code == 401:
        # Token expired: Refresh and retry
        self.authenticate()

        # Re-issue the request with the new header (handled by self._client update)
        # We must recreate the request to pick up the new headers from the
        # client state
        response = self._client.request(method, endpoint, **kwargs)
    return response

Entry point for operations on the collection of contracts.

Source code in src/crosscontract/crossclient/services/contract_service.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
class ContractService:
    """
    Entry point for operations on the collection of contracts.
    """

    _api_version_prefix = "/api/v1"

    def __init__(self, client: "CrossClient"):
        """Initialize the ContractService. The ContractService is responsible for
        managing contracts on the CROSS platform. It provides methods to create,
        retrieve, list, and delete contracts.

        Args:
            client (CrossClient): The CrossClient instance to use for API calls.
        """
        self._client = client
        self._route = f"{self._client._base_url}{self._api_version_prefix}/contract/"

    def create(
        self, contract: CrossContract, activate: bool = False
    ) -> ContractResource:
        """Create a new contract on the CROSS platform

        Args:
            contract (CrossContract): The contract data to create.
            activate (bool): Whether to activate the contract upon creation.
                Defaults to False.

        Raises:
            httpx.HTTPStatusError: If the request fails.

        Returns:
            ContractResource: The created contract object.
        """
        # 1. Create the contract on the platform
        json_payload = contract.model_dump(mode="json")
        response = self._client.post(self._route, json=json_payload)
        raise_from_response(response)

        # 2. Extract info from response
        resp = response.json()
        contract = CrossContract.model_validate(resp["contract"])
        status = resp["status"]

        # 3. Activate the contract if requested
        if activate:
            status = self._client.contracts.change_status(contract.name, "Active")

        # 4. Return the ContractResource
        return ContractResource(self, contract=contract, status=status)

    def overview(self) -> pd.DataFrame:
        """Get a DataFrame with an overview of all contracts, their status, and
        metadata.

        Returns:
            pd.DataFrame: DataFrame containing contract overviews.
        """
        endpoint = f"{self._route}metadata"
        response = self._client.get(endpoint)
        raise_from_response(response)
        df = pd.DataFrame(response.json())
        return df

    def get_list(self) -> dict[str, ContractResource]:
        """
        Lists all available contracts as ContractResource objects.

        Returns:
            dict[str, ContractResource]: Dictionary of contract resources keyed
                by contract name.
        """
        endpoint = self._route
        response = self._client.get(endpoint)
        raise_from_response(response)
        json_body = response.json()
        return {
            item["name"]: ContractResource(
                self,
                contract=CrossContract.model_validate(item["contract"]),
                status=item["status"],
            )
            for item in json_body
        }

    def get(self, name: str) -> ContractResource:
        """Get contract from the CROSS platform by name.

        Args:
            name (str): The name of the contract.

        Raises:
            httpx.HTTPStatusError: If the request fails.

        Returns:
            ContractResource: The contract resource object.
        """
        endpoint = f"{self._route}{name}"
        response = self._client.get(endpoint)
        raise_from_response(response)
        json_body = response.json()
        contract = CrossContract.model_validate(json_body["contract"])
        return ContractResource(self, contract=contract, status=json_body.get("status"))

    def delete(self, name: str, hard: bool = False) -> None:
        """Delete a contract by name if it exists. A contract can only be deleted
        if:
        1. Contract is in "Draft" status
        2. Contract status is "Retired" and the data associated with the contract
            is deleted

        If `hard` is set to True, the contract and all associated data will be deleted.
        Note: This is a dangerous operation and should be used with caution. Usually,
            admin rights are required.

        Args:
            name (str): The name of the contract to delete.
            hard (bool): Whether to perform a hard delete (including data).
                Note: This is a dangerous operation and should be used with caution.
                    it will delete all data associated with the contract. Usually,
                    admin rights are required.
        """
        if hard:
            # if in active or suspended status, change to retired first
            # if in draft status, this will raise an error, which is fine
            try:
                self.change_status(name, "Retired")
            except Exception:
                pass
            # delete all associated data
            try:
                self._drop_data_table(name)
            except Exception:
                pass
        # delete the contract
        try:
            res = self._client.delete(f"{self._route}{name}")
            raise_from_response(res)
        except ResourceNotFoundError:
            # be silent if the contract does not exist
            return

    def change_status(
        self,
        name: str,
        status: Literal["Draft", "Active", "Suspended", "Retired"],
    ) -> str:
        """Change the status of a contract. Allowable status transitions are enforced
        by the CROSS platform. The allowable statuses are:
            1. Draft
            2. Active
            3. Suspended
            4. Retired
        Allowed transitions:
            - Draft -> Active
            - Active -> Suspended
            - Suspended -> Active
            - Active -> Retired
            - Suspended -> Retired

        Args:
            name (str): The name of the contract to change status.
            status (Literal["Draft", "Active", "Suspended", "Retired"]):
                The new status for the contract.

        Raises:
            httpx.HTTPStatusError: If the request fails.

        Returns:
            str: The updated status of the contract.
        """
        payload = {"status": status}
        res = self._client.patch(f"{self._route}{name}/state", json=payload)
        raise_from_response(res)
        return res.json()

    def _drop_data_table(self, name: str) -> None:
        """Drop the table storing the data for the given contract. This deletes all
        data associated with the contract. Dropping the data table is irreversible.
        It can only be performed if the contract is in "Retired" status.

        Note: This operation is ireversible and will delete all data associated with the
        contract.

        Args:
            name (str): The name of the contract whose data to delete.
        """
        # delete the contract
        res = self._client.delete(f"{self._route}{name}/storage")
        raise_from_response(res)

    def _add_data(self, name: str, data: pd.DataFrame) -> None:
        """Add data for the contract on the CROSS platform. Note that this method
        does not perform schema validation. Use ContractResource.add_data() to
        validate data against the contract schema before uploading. I.e., it is
        better to use:
            contract = client.contracts.get(name)
            contract.add_data(data)  # <-- performs validation

        Args:
            name (str): The name of the contract to add data to.
            data (pd.DataFrame): The data to be added.

        Raises:
            httpx.HTTPStatusError: If the request fails.
        """
        endpoint = f"{self._route}{name}/data"

        # construct the payload
        with io.BytesIO(data.to_csv(index=False).encode("utf-8")) as csv_buffer:
            files = {"file": (f"{name}.csv", csv_buffer, "text/csv")}
            res = self._client.post(endpoint, files=files)
        raise_from_response(res)
        return

    def _get_data(
        self,
        name: str,
        columns: list[str] | None = None,
        filters: dict[str, str] | None = None,
        unique: bool = False,
    ) -> pd.DataFrame:
        """Get data for the contract from the CROSS platform.

        Args:
            name (str): The name of the contract to get data for.
            columns (list[str] | None): Optional list of columns to retrieve.
                If None, all columns are retrieved.
            filters (dict[str, str] | None): Optional dictionary of filters to apply.
                The keys are column names and the values are the filter values.
                Currently, only equality filters are supported and only one value per
                filter.
            unique (bool): Whether to return only unique rows.

        Returns:
            pd.DataFrame: The data associated with the contract.
        """
        endpoint = f"{self._route}{name}/data"
        params: dict[str, Any] = {}
        if columns:
            params["columns"] = ",".join(columns)
        if filters:
            for key, value in filters.items():
                params[key] = value
        if unique:
            params["unique"] = "true"

        # perform the request using parquet as data format for efficiency
        params["format"] = "parquet"
        response = self._client.get(endpoint, params=params)
        raise_from_response(response)
        # read the CSV data into a DataFrame
        df = pd.read_parquet(io.BytesIO(response.content))
        return df

__init__(client)

Initialize the ContractService. The ContractService is responsible for managing contracts on the CROSS platform. It provides methods to create, retrieve, list, and delete contracts.

Parameters:

Name Type Description Default
client CrossClient

The CrossClient instance to use for API calls.

 required
Source code in src/crosscontract/crossclient/services/contract_service.py 
22
23
24
25
26
27
28
29
30
31
def __init__(self, client: "CrossClient"):
    """Initialize the ContractService. The ContractService is responsible for
    managing contracts on the CROSS platform. It provides methods to create,
    retrieve, list, and delete contracts.

    Args:
        client (CrossClient): The CrossClient instance to use for API calls.
    """
    self._client = client
    self._route = f"{self._client._base_url}{self._api_version_prefix}/contract/"

change_status(name, status)

Change the status of a contract. Allowable status transitions are enforced by the CROSS platform. The allowable statuses are: 1. Draft 2. Active 3. Suspended 4. Retired Allowed transitions: - Draft -> Active - Active -> Suspended - Suspended -> Active - Active -> Retired - Suspended -> Retired

Parameters:

Name Type Description Default
name str

The name of the contract to change status.

 required
status Literal['Draft', 'Active', 'Suspended', 'Retired']

The new status for the contract.

 required

Raises:

Type Description
HTTPStatusError

If the request fails.

Returns:

Name Type Description
str str

The updated status of the contract.
Source code in src/crosscontract/crossclient/services/contract_service.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
def change_status(
    self,
    name: str,
    status: Literal["Draft", "Active", "Suspended", "Retired"],
) -> str:
    """Change the status of a contract. Allowable status transitions are enforced
    by the CROSS platform. The allowable statuses are:
        1. Draft
        2. Active
        3. Suspended
        4. Retired
    Allowed transitions:
        - Draft -> Active
        - Active -> Suspended
        - Suspended -> Active
        - Active -> Retired
        - Suspended -> Retired

    Args:
        name (str): The name of the contract to change status.
        status (Literal["Draft", "Active", "Suspended", "Retired"]):
            The new status for the contract.

    Raises:
        httpx.HTTPStatusError: If the request fails.

    Returns:
        str: The updated status of the contract.
    """
    payload = {"status": status}
    res = self._client.patch(f"{self._route}{name}/state", json=payload)
    raise_from_response(res)
    return res.json()

create(contract, activate=False)

Create a new contract on the CROSS platform

Parameters:

Name Type Description Default
contract CrossContract

The contract data to create.

 required
activate bool

Whether to activate the contract upon creation. Defaults to False.

 False

Raises:

Type Description
HTTPStatusError

If the request fails.

Returns:

Name Type Description
ContractResource ContractResource

The created contract object.
Source code in src/crosscontract/crossclient/services/contract_service.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def create(
    self, contract: CrossContract, activate: bool = False
) -> ContractResource:
    """Create a new contract on the CROSS platform

    Args:
        contract (CrossContract): The contract data to create.
        activate (bool): Whether to activate the contract upon creation.
            Defaults to False.

    Raises:
        httpx.HTTPStatusError: If the request fails.

    Returns:
        ContractResource: The created contract object.
    """
    # 1. Create the contract on the platform
    json_payload = contract.model_dump(mode="json")
    response = self._client.post(self._route, json=json_payload)
    raise_from_response(response)

    # 2. Extract info from response
    resp = response.json()
    contract = CrossContract.model_validate(resp["contract"])
    status = resp["status"]

    # 3. Activate the contract if requested
    if activate:
        status = self._client.contracts.change_status(contract.name, "Active")

    # 4. Return the ContractResource
    return ContractResource(self, contract=contract, status=status)

delete(name, hard=False)

Delete a contract by name if it exists. A contract can only be deleted if: 1. Contract is in "Draft" status 2. Contract status is "Retired" and the data associated with the contract is deleted

If hard is set to True, the contract and all associated data will be deleted. Note: This is a dangerous operation and should be used with caution. Usually, admin rights are required.

Parameters:

Name Type Description Default
name str

The name of the contract to delete.

 required
hard bool

Whether to perform a hard delete (including data). Note: This is a dangerous operation and should be used with caution. it will delete all data associated with the contract. Usually, admin rights are required.

 False
Source code in src/crosscontract/crossclient/services/contract_service.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def delete(self, name: str, hard: bool = False) -> None:
    """Delete a contract by name if it exists. A contract can only be deleted
    if:
    1. Contract is in "Draft" status
    2. Contract status is "Retired" and the data associated with the contract
        is deleted

    If `hard` is set to True, the contract and all associated data will be deleted.
    Note: This is a dangerous operation and should be used with caution. Usually,
        admin rights are required.

    Args:
        name (str): The name of the contract to delete.
        hard (bool): Whether to perform a hard delete (including data).
            Note: This is a dangerous operation and should be used with caution.
                it will delete all data associated with the contract. Usually,
                admin rights are required.
    """
    if hard:
        # if in active or suspended status, change to retired first
        # if in draft status, this will raise an error, which is fine
        try:
            self.change_status(name, "Retired")
        except Exception:
            pass
        # delete all associated data
        try:
            self._drop_data_table(name)
        except Exception:
            pass
    # delete the contract
    try:
        res = self._client.delete(f"{self._route}{name}")
        raise_from_response(res)
    except ResourceNotFoundError:
        # be silent if the contract does not exist
        return

get(name)

Get contract from the CROSS platform by name.

Parameters:

Name Type Description Default
name str

The name of the contract.

 required

Raises:

Type Description
HTTPStatusError

If the request fails.

Returns:

Name Type Description
ContractResource ContractResource

The contract resource object.
Source code in src/crosscontract/crossclient/services/contract_service.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def get(self, name: str) -> ContractResource:
    """Get contract from the CROSS platform by name.

    Args:
        name (str): The name of the contract.

    Raises:
        httpx.HTTPStatusError: If the request fails.

    Returns:
        ContractResource: The contract resource object.
    """
    endpoint = f"{self._route}{name}"
    response = self._client.get(endpoint)
    raise_from_response(response)
    json_body = response.json()
    contract = CrossContract.model_validate(json_body["contract"])
    return ContractResource(self, contract=contract, status=json_body.get("status"))

get_list()

Lists all available contracts as ContractResource objects.

Returns:

Type Description
dict[str, ContractResource]

dict[str, ContractResource]: Dictionary of contract resources keyed by contract name.
Source code in src/crosscontract/crossclient/services/contract_service.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def get_list(self) -> dict[str, ContractResource]:
    """
    Lists all available contracts as ContractResource objects.

    Returns:
        dict[str, ContractResource]: Dictionary of contract resources keyed
            by contract name.
    """
    endpoint = self._route
    response = self._client.get(endpoint)
    raise_from_response(response)
    json_body = response.json()
    return {
        item["name"]: ContractResource(
            self,
            contract=CrossContract.model_validate(item["contract"]),
            status=item["status"],
        )
        for item in json_body
    }

overview()

Get a DataFrame with an overview of all contracts, their status, and metadata.

Returns:

Type Description
DataFrame

pd.DataFrame: DataFrame containing contract overviews.
Source code in src/crosscontract/crossclient/services/contract_service.py 
66
67
68
69
70
71
72
73
74
75
76
77
def overview(self) -> pd.DataFrame:
    """Get a DataFrame with an overview of all contracts, their status, and
    metadata.

    Returns:
        pd.DataFrame: DataFrame containing contract overviews.
    """
    endpoint = f"{self._route}metadata"
    response = self._client.get(endpoint)
    raise_from_response(response)
    df = pd.DataFrame(response.json())
    return df

A contract that is related to contract on the CROSS platform.

ContractResources are read-only wrappers around the actual contract data that is stored on the CROSS platform. They provide lazy loading of the contract details and methods to interact with the contract, such as adding data.

Attributes:

Name Type Description
name str

The name of the contract.
status str

The status of the contract.
contract CrossContract

The full contract details.
service ContractService

The ContractService instance used for API calls.
Source code in src/crosscontract/crossclient/services/contract_resource.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
class ContractResource:
    """A contract that is related to contract on the CROSS platform.

    ContractResources are read-only wrappers around the actual contract data that
    is stored on the CROSS platform. They provide lazy loading of the contract
    details and methods to interact with the contract, such as adding data.

    Attributes:
        name (str): The name of the contract.
        status (str): The status of the contract.
        contract (CrossContract): The full contract details.
        service (ContractService): The ContractService instance used for API calls.
    """

    def __init__(
        self,
        service: "ContractService",
        status: str,
        name: str | None = None,
        contract: CrossContract | None = None,
    ):
        """Initialize the ContractResource.

        Args:
            service (ContractService): The ContractService instance to use for
                API calls.
            name (str | None): The name of the contract.
                Required if contract is not provided.
            contract (CrossContract | None): The CrossContract instance.
                If not provided, the contract details will be fetched lazily
                when accessed.
        """
        self._service = service

        # ensure consistence of the name and contract
        if contract and name and contract.name != name:
            raise ValueError(
                f"Name '{name}' does not match contract name '{contract.name}'."
            )
        elif not name and not contract:
            raise ValueError("Either name or contract must be provided.")
        self._name = name or contract.name  # type: ignore
        self._contract = contract
        self._status = status

    @property
    def name(self) -> str:
        return self._name

    @property
    def status(self) -> str | None:
        return self._status

    @property
    def contract(self) -> CrossContract:
        """The full contract details as a CrossContract object.

        This property uses lazy loading to fetch the contract details from the
        CROSS platform only when accessed for the first time.

        Returns:
            CrossContract: The full contract details.
        """
        if self._contract is None:
            self.refresh()
        return self._contract  # type: ignore

    def __setattr__(self, name, value):
        # 1. Access the class to find the attribute definition
        # We use type(self) to avoid triggering infinite recursion or property getters
        attr = getattr(type(self), name, None)

        # 2. Check if the attribute is a property and if it has no setter
        if isinstance(attr, property) and attr.fset is None:
            raise AttributeError(
                "ContractResource is read-only. Use the methods to update properties."
            )

        # 3. If it's not a read-only property, allow the default behavior
        # This allows setting private variables like self._x = 10
        super().__setattr__(name, value)

    def __repr__(self):
        return f"ContractResource(name={self.name}, status={self.status})"

    def change_status(
        self,
        status: Literal["Draft", "Active", "Suspended", "Retired"],
    ) -> None:
        """Change the status of the contract.

        Args:
            status (Literal["Draft", "Active", "Suspended", "Retired"]):
                The new status for the contract.
        """
        self._service.change_status(self.name, status)
        self._status = status

    def refresh(self):
        """Fetch the full contract details from the CROSS platform."""
        contract = self._service.get(self.name)
        if contract.name != self.name:
            raise ValueError(
                f"Fetched contract name '{contract.name}' does not match "
                f"resource name '{self.name}'."
            )
        self._contract = contract

    def _prepare_dataframe_csv_upload(self, df: pd.DataFrame) -> pd.DataFrame:
        """Prepare a DataFrame for CSV upload by formatting datetime columns.

        This method converts datetime-typed fields defined in the contract's
        table schema from pandas datetime dtypes to string values using the
        field's configured format. Columns of other data types are left
        unchanged.

        Args:
            df (pd.DataFrame): The input DataFrame to be prepared for CSV upload.

        Returns:
            pd.DataFrame: The prepared DataFrame ready for CSV upload.
        """
        # convert datetime fields to string with correct format
        dt_fields = [
            f
            for f in self.contract.tableschema.field_iterator()
            if f.type == "datetime" and f.name in df.columns
        ]
        if len(dt_fields) == 0:
            return df
        df_out = df.copy(deep=False)
        for field in dt_fields:
            if pd.api.types.is_datetime64_any_dtype(df[field.name]):
                df_out[field.name] = df_out[field.name].dt.strftime(field.format)
        return df_out

    def add_data(self, data: pd.DataFrame, validate: bool = True) -> None:
        """Add data for the contract on in the CROSS platform.

        Args:
            data (pd.DataFrame): The data to be added.
            validate (bool): Whether to validate the data against the contract
                schema before uploading.
                Defaults to True.

        Raises:
            validationError: If the data does not conform to the contract schema.
        """
        if validate:
            # validate data against contract schema at the client side
            self.validate_dataframe(data)
        data_out = self._prepare_dataframe_csv_upload(data)
        self._service._add_data(self.name, data_out)

    def get_data(
        self,
        columns: list[str] | None = None,
        filters: dict[str, str] | None = None,
        unique: bool = False,
    ) -> pd.DataFrame:
        """Get data for the contract from the CROSS platform.

        Args:
            columns (list[str] | None): Optional list of columns to retrieve.
                If None, all columns are retrieved.
            filters (dict[str, str] | None): Optional dictionary of filters to apply.
                The keys are column names and the values are the filter values.
                Currently, only equality filters are supported and only one value per
                filter.
            unique (bool): Whether to return only unique rows.

        Returns:
            pd.DataFrame: The data associated with the contract.
        """
        return self._service._get_data(
            name=self.name, columns=columns, filters=filters, unique=unique
        )

    def validate_dataframe(
        self,
        df: pd.DataFrame,
        skip_primary_key_validation: bool = True,
        skip_foreign_key_validation: bool = True,
        lazy: bool = True,
    ):
        """Validate a DataFrame against the schema of the contract.
        It allows to provide existing primary
        key and foreign key values for validation. If provided, the primary key
        uniqueness is checked against the union of the existing and the DataFrame
        values. Similarly, foreign key integrity is checked against the union of
        existing and DataFrame values in case of self-referencing foreign keys.

        The validation is performed including primary key and foreign key checks
        that may require fetching existing key values from the CROSS platform.

        Args:
            df (pd.DataFrame): The DataFrame to validate.
            skip_primary_key_validation (bool): If True, skip primary key validation.
                Default is False.
            skip_foreign_key_validation (bool): If True, skip foreign key validation.
                Default is False.
            lazy (bool): If True, collect all validation errors and raise them together.
                If False, raise the first validation error encountered.
                Default is True.

        Raises:
            ValidationError: If the DataFrame does not conform to the schema.
        """
        schema = self.contract.tableschema

        # get the existing primary key values from the platform if needed
        if skip_primary_key_validation:
            primary_key_values = None
        else:
            # fetch the existing primary key values from the platform
            primary_key_values = self.get_primary_key_values()

        # get the existing foreign key values from the platform if needed
        if skip_foreign_key_validation:
            foreign_key_values = None
        else:
            foreign_key_values = self.get_foreign_key_values()

        # validate the dataframe against the schema
        try:
            schema.validate_dataframe(
                df=df,
                primary_key_values=primary_key_values,
                foreign_key_values=foreign_key_values,
                skip_primary_key_validation=skip_primary_key_validation,
                skip_foreign_key_validation=skip_foreign_key_validation,
                lazy=lazy,
            )
        except SchemaValidationError as e:
            # convert to CrossClient ValidationError
            raise ValidationError(
                message=f"DataFrame validation against contract '{self.name}' "
                "schema failed.",
                validation_errors=e.to_list(),
            ) from e

    def get_primary_key_values(self) -> list[tuple] | None:
        """Get the existing primary key values for the contract from the CROSS platform.
        This is needed if you want to perform primary key validation including existing
        values, i.e., to ensure uniqueness of the primary key across both existing
        and new data.

        Returns:
            list[tuple]: A list of tuples representing the existing primary key values.

        Returns:
            list[tuple] | None: A list of tuples representing the existing primary key
                values. Returns None if the contract does not have a primary key defined
                or if there are no existing primary key values.
        """
        schema = self.contract.tableschema

        # if there is no primary key defined, return None
        if not schema.primaryKey:
            return None

        # get the existing primary key values from the platform
        df_primary_key_values = self.get_data(
            columns=schema.primaryKey.root,
            unique=True,
        )

        # if there are no existing primary key values, return None else return the
        # values as list of tuples
        if df_primary_key_values.empty:
            primary_key_values = None
        else:
            primary_key_values = [
                tuple(row)
                for row in df_primary_key_values.itertuples(index=False, name=None)
            ]
        return primary_key_values

    def get_foreign_key_values(self) -> dict[tuple, list[tuple]] | None:
        """Get the existing foreign key values for the contract from the CROSS platform.
        This is needed if you want to perform foreign key validation including existing
        values, i.e., to ensure referential integrity of the foreign keys across both
        existing and new data.

        Returns:
            dict[tuple, list[tuple]] | None: A dictionary where the keys are tuples
                representing the foreign key fields, and the values are lists of tuples
                representing the existing foreign key values. Returns None if the
                contract does not have foreign keys defined or if there are no existing
                foreign key values.
        """
        schema = self.contract.tableschema

        # if there are no foreign keys defined, return None
        if not schema.foreignKeys:
            return None

        foreign_key_values: dict[tuple, list[tuple]] = {}

        # for each foreign key, get the existing foreign key values from the platform
        for fk in schema.foreignKeys.root:
            fk_contract_name = fk.reference.resource or self.name
            fk_field_names = fk.reference.fields

            df_fk = self._service._get_data(
                name=fk_contract_name,
                columns=fk_field_names,
                unique=True,
            )

            foreign_key_values[tuple(fk.fields)] = [
                tuple(row) for row in df_fk.itertuples(index=False, name=None)
            ]

        return foreign_key_values

    def drop_data(self) -> None:
        """Delete all data associated with the contract on the CROSS platform."""
        self._service._drop_data_table(self.name)

contract property

The full contract details as a CrossContract object.

This property uses lazy loading to fetch the contract details from the CROSS platform only when accessed for the first time.

Returns:

Name Type Description
CrossContract CrossContract

The full contract details.

__init__(service, status, name=None, contract=None)

Initialize the ContractResource.

Parameters:

Name Type Description Default
service ContractService

The ContractService instance to use for API calls.

 required
name str | None

The name of the contract. Required if contract is not provided.

 None
contract CrossContract | None

The CrossContract instance. If not provided, the contract details will be fetched lazily when accessed.

 None
Source code in src/crosscontract/crossclient/services/contract_resource.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def __init__(
    self,
    service: "ContractService",
    status: str,
    name: str | None = None,
    contract: CrossContract | None = None,
):
    """Initialize the ContractResource.

    Args:
        service (ContractService): The ContractService instance to use for
            API calls.
        name (str | None): The name of the contract.
            Required if contract is not provided.
        contract (CrossContract | None): The CrossContract instance.
            If not provided, the contract details will be fetched lazily
            when accessed.
    """
    self._service = service

    # ensure consistence of the name and contract
    if contract and name and contract.name != name:
        raise ValueError(
            f"Name '{name}' does not match contract name '{contract.name}'."
        )
    elif not name and not contract:
        raise ValueError("Either name or contract must be provided.")
    self._name = name or contract.name  # type: ignore
    self._contract = contract
    self._status = status

add_data(data, validate=True)

Add data for the contract on in the CROSS platform.

Parameters:

Name Type Description Default
data DataFrame

The data to be added.

 required
validate bool

Whether to validate the data against the contract schema before uploading. Defaults to True.

 True

Raises:

Type Description
validationError

If the data does not conform to the contract schema.
Source code in src/crosscontract/crossclient/services/contract_resource.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def add_data(self, data: pd.DataFrame, validate: bool = True) -> None:
    """Add data for the contract on in the CROSS platform.

    Args:
        data (pd.DataFrame): The data to be added.
        validate (bool): Whether to validate the data against the contract
            schema before uploading.
            Defaults to True.

    Raises:
        validationError: If the data does not conform to the contract schema.
    """
    if validate:
        # validate data against contract schema at the client side
        self.validate_dataframe(data)
    data_out = self._prepare_dataframe_csv_upload(data)
    self._service._add_data(self.name, data_out)

change_status(status)

Change the status of the contract.

Parameters:

Name Type Description Default
status Literal['Draft', 'Active', 'Suspended', 'Retired']

The new status for the contract.

 required
Source code in src/crosscontract/crossclient/services/contract_resource.py 
 99
100
101
102
103
104
105
106
107
108
109
110
def change_status(
    self,
    status: Literal["Draft", "Active", "Suspended", "Retired"],
) -> None:
    """Change the status of the contract.

    Args:
        status (Literal["Draft", "Active", "Suspended", "Retired"]):
            The new status for the contract.
    """
    self._service.change_status(self.name, status)
    self._status = status

drop_data()

Delete all data associated with the contract on the CROSS platform.

Source code in src/crosscontract/crossclient/services/contract_resource.py 
330
331
332
def drop_data(self) -> None:
    """Delete all data associated with the contract on the CROSS platform."""
    self._service._drop_data_table(self.name)

get_data(columns=None, filters=None, unique=False)

Get data for the contract from the CROSS platform.

Parameters:

Name Type Description Default
columns list[str] | None

Optional list of columns to retrieve. If None, all columns are retrieved.

 None
filters dict[str, str] | None

Optional dictionary of filters to apply. The keys are column names and the values are the filter values. Currently, only equality filters are supported and only one value per filter.

 None
unique bool

Whether to return only unique rows.

 False

Returns:

Type Description
DataFrame

pd.DataFrame: The data associated with the contract.
Source code in src/crosscontract/crossclient/services/contract_resource.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
def get_data(
    self,
    columns: list[str] | None = None,
    filters: dict[str, str] | None = None,
    unique: bool = False,
) -> pd.DataFrame:
    """Get data for the contract from the CROSS platform.

    Args:
        columns (list[str] | None): Optional list of columns to retrieve.
            If None, all columns are retrieved.
        filters (dict[str, str] | None): Optional dictionary of filters to apply.
            The keys are column names and the values are the filter values.
            Currently, only equality filters are supported and only one value per
            filter.
        unique (bool): Whether to return only unique rows.

    Returns:
        pd.DataFrame: The data associated with the contract.
    """
    return self._service._get_data(
        name=self.name, columns=columns, filters=filters, unique=unique
    )

get_foreign_key_values()

Get the existing foreign key values for the contract from the CROSS platform. This is needed if you want to perform foreign key validation including existing values, i.e., to ensure referential integrity of the foreign keys across both existing and new data.

Returns:

Type Description
dict[tuple, list[tuple]] | None

dict[tuple, list[tuple]] | None: A dictionary where the keys are tuples representing the foreign key fields, and the values are lists of tuples representing the existing foreign key values. Returns None if the contract does not have foreign keys defined or if there are no existing foreign key values.
Source code in src/crosscontract/crossclient/services/contract_resource.py 
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
def get_foreign_key_values(self) -> dict[tuple, list[tuple]] | None:
    """Get the existing foreign key values for the contract from the CROSS platform.
    This is needed if you want to perform foreign key validation including existing
    values, i.e., to ensure referential integrity of the foreign keys across both
    existing and new data.

    Returns:
        dict[tuple, list[tuple]] | None: A dictionary where the keys are tuples
            representing the foreign key fields, and the values are lists of tuples
            representing the existing foreign key values. Returns None if the
            contract does not have foreign keys defined or if there are no existing
            foreign key values.
    """
    schema = self.contract.tableschema

    # if there are no foreign keys defined, return None
    if not schema.foreignKeys:
        return None

    foreign_key_values: dict[tuple, list[tuple]] = {}

    # for each foreign key, get the existing foreign key values from the platform
    for fk in schema.foreignKeys.root:
        fk_contract_name = fk.reference.resource or self.name
        fk_field_names = fk.reference.fields

        df_fk = self._service._get_data(
            name=fk_contract_name,
            columns=fk_field_names,
            unique=True,
        )

        foreign_key_values[tuple(fk.fields)] = [
            tuple(row) for row in df_fk.itertuples(index=False, name=None)
        ]

    return foreign_key_values

get_primary_key_values()

Get the existing primary key values for the contract from the CROSS platform. This is needed if you want to perform primary key validation including existing values, i.e., to ensure uniqueness of the primary key across both existing and new data.

Returns:

Type Description
list[tuple] | None

list[tuple]: A list of tuples representing the existing primary key values.

Returns:

Type Description
list[tuple] | None

list[tuple] | None: A list of tuples representing the existing primary key values. Returns None if the contract does not have a primary key defined or if there are no existing primary key values.
Source code in src/crosscontract/crossclient/services/contract_resource.py 
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
def get_primary_key_values(self) -> list[tuple] | None:
    """Get the existing primary key values for the contract from the CROSS platform.
    This is needed if you want to perform primary key validation including existing
    values, i.e., to ensure uniqueness of the primary key across both existing
    and new data.

    Returns:
        list[tuple]: A list of tuples representing the existing primary key values.

    Returns:
        list[tuple] | None: A list of tuples representing the existing primary key
            values. Returns None if the contract does not have a primary key defined
            or if there are no existing primary key values.
    """
    schema = self.contract.tableschema

    # if there is no primary key defined, return None
    if not schema.primaryKey:
        return None

    # get the existing primary key values from the platform
    df_primary_key_values = self.get_data(
        columns=schema.primaryKey.root,
        unique=True,
    )

    # if there are no existing primary key values, return None else return the
    # values as list of tuples
    if df_primary_key_values.empty:
        primary_key_values = None
    else:
        primary_key_values = [
            tuple(row)
            for row in df_primary_key_values.itertuples(index=False, name=None)
        ]
    return primary_key_values

refresh()

Fetch the full contract details from the CROSS platform.

Source code in src/crosscontract/crossclient/services/contract_resource.py 
112
113
114
115
116
117
118
119
120
def refresh(self):
    """Fetch the full contract details from the CROSS platform."""
    contract = self._service.get(self.name)
    if contract.name != self.name:
        raise ValueError(
            f"Fetched contract name '{contract.name}' does not match "
            f"resource name '{self.name}'."
        )
    self._contract = contract

validate_dataframe(df, skip_primary_key_validation=True, skip_foreign_key_validation=True, lazy=True)

Validate a DataFrame against the schema of the contract. It allows to provide existing primary key and foreign key values for validation. If provided, the primary key uniqueness is checked against the union of the existing and the DataFrame values. Similarly, foreign key integrity is checked against the union of existing and DataFrame values in case of self-referencing foreign keys.

The validation is performed including primary key and foreign key checks that may require fetching existing key values from the CROSS platform.

Parameters:

Name Type Description Default
df DataFrame

The DataFrame to validate.

 required
skip_primary_key_validation bool

If True, skip primary key validation. Default is False.

 True
skip_foreign_key_validation bool

If True, skip foreign key validation. Default is False.

 True
lazy bool

If True, collect all validation errors and raise them together. If False, raise the first validation error encountered. Default is True.

 True

Raises:

Type Description
ValidationError

If the DataFrame does not conform to the schema.
Source code in src/crosscontract/crossclient/services/contract_resource.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
def validate_dataframe(
    self,
    df: pd.DataFrame,
    skip_primary_key_validation: bool = True,
    skip_foreign_key_validation: bool = True,
    lazy: bool = True,
):
    """Validate a DataFrame against the schema of the contract.
    It allows to provide existing primary
    key and foreign key values for validation. If provided, the primary key
    uniqueness is checked against the union of the existing and the DataFrame
    values. Similarly, foreign key integrity is checked against the union of
    existing and DataFrame values in case of self-referencing foreign keys.

    The validation is performed including primary key and foreign key checks
    that may require fetching existing key values from the CROSS platform.

    Args:
        df (pd.DataFrame): The DataFrame to validate.
        skip_primary_key_validation (bool): If True, skip primary key validation.
            Default is False.
        skip_foreign_key_validation (bool): If True, skip foreign key validation.
            Default is False.
        lazy (bool): If True, collect all validation errors and raise them together.
            If False, raise the first validation error encountered.
            Default is True.

    Raises:
        ValidationError: If the DataFrame does not conform to the schema.
    """
    schema = self.contract.tableschema

    # get the existing primary key values from the platform if needed
    if skip_primary_key_validation:
        primary_key_values = None
    else:
        # fetch the existing primary key values from the platform
        primary_key_values = self.get_primary_key_values()

    # get the existing foreign key values from the platform if needed
    if skip_foreign_key_validation:
        foreign_key_values = None
    else:
        foreign_key_values = self.get_foreign_key_values()

    # validate the dataframe against the schema
    try:
        schema.validate_dataframe(
            df=df,
            primary_key_values=primary_key_values,
            foreign_key_values=foreign_key_values,
            skip_primary_key_validation=skip_primary_key_validation,
            skip_foreign_key_validation=skip_foreign_key_validation,
            lazy=lazy,
        )
    except SchemaValidationError as e:
        # convert to CrossClient ValidationError
        raise ValidationError(
            message=f"DataFrame validation against contract '{self.name}' "
            "schema failed.",
            validation_errors=e.to_list(),
        ) from e