API Reference: CrossClient

Source code in src/crosscontract/crossclient/crossclient.py

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

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

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

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

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

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

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

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

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

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.to_server()
        response = self._client.post(self._route, json=json_payload)
        raise_from_response(response)

        # 2. Build the resource from the response payload
        resource = ContractResource.from_response(self, response.json())

        # 3. Activate the contract if requested
        if activate:
            resource.change_status("Active")

        return resource

    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, contract_type: list[ContractType] | None = None
    ) -> dict[str, ContractResource]:
        """
        Lists all available contracts as ContractResource objects.

        Args:
            contract_type (list[str] | None): Optional filter restricting the
                result to one or more contract types (e.g.
                `["General", "Dimension"]`). If None, contracts of every type
                are returned.

        Returns:
            dict[str, ContractResource]: Dictionary of contract resources keyed
                by contract name.
        """
        endpoint = self._route
        params: dict[str, Any] = {}
        if contract_type:
            params["contract_type"] = contract_type
        response = self._client.get(endpoint, params=params or None)
        raise_from_response(response)
        json_body = response.json()
        return {
            item["name"]: ContractResource.from_response(self, item)
            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)
        return ContractResource.from_response(self, response.json())

    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: ContractStatus,
    ) -> 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 (ContractStatus): 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 is a decommissioning operation: it discards the data of **every**
        project that submitted under the contract, not only the caller's, and
        requires the contract to be `Retired`. It is restricted to
        administrators and is irreversible. `_delete_data` is the narrower,
        separate operation: it removes only the rows one project owns and
        requires the contract to be `Active`.

        Args:
            name (str): The name of the contract whose data table to drop.

        Raises:
            CrossClientError: If the request fails. Raised via
                `raise_from_response` as a more specific client exception
                such as `ResourceNotFoundError` or `ConflictError`.
        """
        # 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, *, project_name: str | None = None
    ) -> 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.
            project_name (str | None): Optional project name under which the data
                are submitted. If None, the CROSS platform infers the project from the
                caller's memberships, which succeeds only when there is exactly one.

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

        params = {"project_name": project_name} if project_name is not None else None

        # construct the payload as parquet for type safety and efficiency.
        with io.BytesIO() as buffer:
            data.to_parquet(buffer, index=False)
            buffer.seek(0)
            files = {
                "file": (f"{name}.parquet", buffer, "application/vnd.apache.parquet")
            }
            res = self._client.post(endpoint, files=files, params=params)
        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)
        df = pd.read_parquet(io.BytesIO(response.content))
        return df

    def _delete_data(
        self,
        name: str,
        filters: dict[str, FilterValue | list[FilterValue]],
        *,
        project_name: str | None = None,
        confirm_delete_all: bool = False,
    ) -> None:
        """Delete rows for the contract matching the given equality filters.

        Filters must be non-empty unless `confirm_delete_all` is set, which
        removes every row the resolved project owns under this contract while
        leaving the table in place. `_drop_data_table` is the wider, separate
        operation: it drops the whole table across every project and requires
        the contract to be `Retired`.

        Values may be str/int/float/bool (or lists thereof for multi-value
        equality) and are stringified before being sent as query parameters.
        List values produce repeated query params (e.g. `?col=a&col=b`),
        which the CROSS platform interprets as an equality match against any
        of the listed values.

        Args:
            name (str): The name of the contract whose rows to delete.
            filters (dict): Mapping of column name to value (or list of values)
                to match. Must be non-empty unless `confirm_delete_all` is
                set.
            project_name (str | None): Optional project name for which project the
                data are deleted. If None, the CROSS platform infers the project
                from the caller's memberships, which succeeds only when there is
                exactly one.
            confirm_delete_all (bool): Confirms that an unfiltered delete is
                intended, removing every row the resolved project owns under
                this contract. Required when `filters` is empty, so that a
                filter mapping which collapsed to empty cannot wipe the
                project's rows by accident. Ignored when `filters` is
                non-empty — the flag is then not sent at all, so a filtered
                delete stays filtered regardless of how the CROSS platform
                orders the two. Defaults to `False`.

        Raises:
            ValueError: If `filters` is empty and `confirm_delete_all` is
                False.
            CrossClientError: If the request fails. Raised via
                `raise_from_response` as a more specific client exception
                such as `ResourceNotFoundError`, `ConflictError`, or
                `ServerError`.
        """
        if not filters and not confirm_delete_all:
            raise ValueError(
                "Filters must be non-empty unless confirm_delete_all=True is specified"
            )
        params: dict[str, str | list[str]] = {
            k: [str(x) for x in v] if isinstance(v, list) else str(v)
            for k, v in filters.items()
        }
        if project_name is not None:
            params["project_name"] = project_name
        # Only an unfiltered delete carries the flag: sending it alongside
        # filters would make the scope of the deletion depend on how the
        # platform prioritises the two, and the fallout of guessing wrong is
        # every row the project owns.
        if confirm_delete_all and not filters:
            params["delete_all"] = "True"
        endpoint = f"{self._route}{name}/data"
        response = self._client.delete(endpoint, params=params)
        raise_from_response(response)

__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

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	ContractStatus	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

def change_status(
    self,
    name: str,
    status: ContractStatus,
) -> 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 (ContractStatus): 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

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.to_server()
    response = self._client.post(self._route, json=json_payload)
    raise_from_response(response)

    # 2. Build the resource from the response payload
    resource = ContractResource.from_response(self, response.json())

    # 3. Activate the contract if requested
    if activate:
        resource.change_status("Active")

    return resource

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

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

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)
    return ContractResource.from_response(self, response.json())

get_list(contract_type=None)

Lists all available contracts as ContractResource objects.

Parameters:

Name	Type	Description	Default
contract_type	list[str] | None	Optional filter restricting the result to one or more contract types (e.g. ["General", "Dimension"]). If None, contracts of every type are returned.	None

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

def get_list(
    self, contract_type: list[ContractType] | None = None
) -> dict[str, ContractResource]:
    """
    Lists all available contracts as ContractResource objects.

    Args:
        contract_type (list[str] | None): Optional filter restricting the
            result to one or more contract types (e.g.
            `["General", "Dimension"]`). If None, contracts of every type
            are returned.

    Returns:
        dict[str, ContractResource]: Dictionary of contract resources keyed
            by contract name.
    """
    endpoint = self._route
    params: dict[str, Any] = {}
    if contract_type:
        params["contract_type"] = contract_type
    response = self._client.get(endpoint, params=params or None)
    raise_from_response(response)
    json_body = response.json()
    return {
        item["name"]: ContractResource.from_response(self, item)
        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

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 handle to a contract that exists on the CROSS platform.

ContractResources are read-only wrappers around contract data fetched from the CROSS platform. They are produced exclusively by ContractService methods (create, get, get_list); end users do not construct them directly.

Attributes:

Name	Type	Description
name	str	The name of the contract.
status	str	The status of the contract.
contract	CrossContract	The full contract details.
contract_type	str	The type of the contract, e.g. "General".
service	ContractService	The owning ContractService.

Source code in src/crosscontract/crossclient/services/contract_resource.py

class ContractResource:
    """A handle to a contract that exists on the CROSS platform.

    ContractResources are read-only wrappers around contract data fetched from
    the CROSS platform. They are produced exclusively by `ContractService`
    methods (`create`, `get`, `get_list`); end users do not construct them
    directly.

    Attributes:
        name (str): The name of the contract.
        status (str): The status of the contract.
        contract (CrossContract): The full contract details.
        contract_type (str): The type of the contract, e.g. `"General"`.
        service (ContractService): The owning `ContractService`.
    """

    def __init__(
        self,
        service: "ContractService",
        payload: _ContractEntryPayload,
    ):
        """Initialise from a parsed server payload.

        Most callers should use :meth:`from_response` to parse a raw JSON dict;
        this constructor takes a pre-validated payload to keep tests direct.
        """
        self._service = service
        self._name = payload.name
        self._status = payload.status
        self._contract_type = payload.contract_type
        self._contract = CrossContract.from_server(payload.contract)

    @classmethod
    def from_response(
        cls,
        service: "ContractService",
        response_json: dict[str, Any],
    ) -> "ContractResource":
        """Build a ContractResource from a raw server response dict."""
        payload = _ContractEntryPayload.model_validate(response_json)
        return cls(service, payload)

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

    @property
    def status(self) -> ContractStatus:
        return self._status

    @property
    def contract_type(self) -> str:
        return self._contract_type

    @property
    def contract(self) -> CrossContract:
        return self._contract

    @property
    def is_dimension(self) -> bool:
        """True if the contract's tableschema is a dimension schema."""
        return isinstance(self._contract.tableschema, BaseDimensionSchema)

    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: ContractStatus) -> None:
        """Change the status of the contract.

        Args:
            status (ContractStatus): The new status for the contract.
        """
        self._service.change_status(self.name, status)
        self._status = status

    def refresh(self) -> None:
        """Re-fetch the contract details from the CROSS platform."""
        remote = self._service.get(self.name)
        if remote.name != self.name:
            raise ValueError(
                f"Fetched contract name '{remote.name}' does not match "
                f"resource name '{self.name}'."
            )
        self._contract = remote.contract
        self._status = remote.status
        self._contract_type = remote.contract_type

    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,
        *,
        project_name: str | None = None,
    ) -> None:
        """Add data for the contract on the CROSS platform.

        The rows are stored as owned by the resolved project.

        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.
            project_name (str | None): Optional project name under which the data
                are submitted. If None, the CROSS platform infers the project from the
                caller's memberships, which succeeds only when there is exactly one.

        Raises:
            ValidationError: If the data does not conform to the contract schema.
            CrossClientError: If the upload fails. Raised via
                `raise_from_response` as a more specific client exception such
                as `ResourceNotFoundError` or `ConflictError`.
        """
        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, project_name=project_name)

    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,
        check_existing_primary_key: bool = False,
        check_existing_foreign_key: bool = False,
        lazy: bool = True,
    ) -> None:
        """Validate a DataFrame against the schema of the contract.

        By default nothing is read from the CROSS platform and the data is
        validated on its own. Setting a check flag fetches the values already
        stored: the primary key is then checked against the union of the stored
        keys and the DataFrame's own, and the foreign keys against the stored
        values of the contracts they reference — plus the DataFrame's own rows
        in the case of a self-referencing foreign key.

        Args:
            df (pd.DataFrame): The DataFrame to validate.
            check_existing_primary_key (bool): If True, also check the primary
                key against the values already stored for this contract.
                Default is False.
            check_existing_foreign_key (bool): If True, also check the foreign
                keys against the values already stored for the contracts they
                reference. 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.
            CrossClientError: If fetching the stored values fails. Raised via
                `raise_from_response` as a more specific client exception such
                as `ResourceNotFoundError` when the contract has no stored data
                yet.
        """
        resolver = CrossContractResolver(self._service)
        try:
            self.contract.validate_data(
                df,
                resolver=resolver,
                check_existing_primary_key=check_existing_primary_key,
                check_existing_foreign_key=check_existing_foreign_key,
                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 drop_data(self) -> None:
        """Drop the storage table backing the contract on the CROSS platform.

        This is a decommissioning operation: it discards the data of **every**
        project that submitted under this contract, not only the caller's, and
        requires the contract to be `Retired`. It is restricted to
        administrators. To remove only the rows owned by one project, use
        `delete_data()` — with `confirm_delete_all=True` to clear that
        project's rows entirely.

        Raises:
            CrossClientError: If the request fails. Raised via
                `raise_from_response` as a more specific client exception such
                as `ResourceNotFoundError` or `ConflictError`.
        """
        self._service._drop_data_table(self.name)

    def delete_data(
        self,
        filters: "dict[str, FilterValue | list[FilterValue]]",
        *,
        project_name: str | None = None,
        confirm_delete_all: bool = False,
    ) -> None:
        """Delete rows from the contract's data matching the given equality filters.

        Only rows owned by the resolved project are removed, and the contract
        must be `Active`. `drop_data()` is the wider, separate operation: it
        drops the whole storage table across every project and requires the
        contract to be `Retired`.

        Args:
            filters (dict): Mapping of column name to value (or list of values)
                to match. Values may be str/int/float/bool. Must be non-empty
                unless `confirm_delete_all` is set.
            project_name (str | None): Optional project name for which project the
                data are deleted. If None, the CROSS platform infers the project
                from the caller's memberships, which succeeds only when there is
                exactly one.
            confirm_delete_all (bool): Confirms that an unfiltered delete is
                intended, removing every row the resolved project owns under
                this contract. Required when `filters` is empty, so that a
                filter mapping which collapsed to empty cannot wipe the
                project's rows by accident. Ignored when `filters` is
                non-empty — a filtered delete stays filtered, and the
                confirmation never reaches the CROSS platform. Defaults to
                `False`.

        Raises:
            ValueError: If the contract's cached status is not `"Active"`, or
                if `filters` is empty and `confirm_delete_all` is False. The
                status check is local and uses the cached status; call
                `refresh()` first if the status may have changed on the CROSS
                platform.
            CrossClientError: Propagated from the underlying service/HTTP
                request if the deletion request fails due to client, server,
                or network-related errors.
        """
        if self._status != "Active":
            raise ValueError(
                f"Cannot delete data from contract '{self.name}': status is "
                f"'{self._status}', must be 'Active'. Call refresh() if the "
                "status may have changed on the server."
            )
        self._service._delete_data(
            self.name,
            filters,
            project_name=project_name,
            confirm_delete_all=confirm_delete_all,
        )

is_dimension property

True if the contract's tableschema is a dimension schema.

__init__(service, payload)

Initialise from a parsed server payload.

Most callers should use :meth:from_response to parse a raw JSON dict; this constructor takes a pre-validated payload to keep tests direct.

Source code in src/crosscontract/crossclient/services/contract_resource.py

def __init__(
    self,
    service: "ContractService",
    payload: _ContractEntryPayload,
):
    """Initialise from a parsed server payload.

    Most callers should use :meth:`from_response` to parse a raw JSON dict;
    this constructor takes a pre-validated payload to keep tests direct.
    """
    self._service = service
    self._name = payload.name
    self._status = payload.status
    self._contract_type = payload.contract_type
    self._contract = CrossContract.from_server(payload.contract)

add_data(data, validate=True, *, project_name=None)

Add data for the contract on the CROSS platform.

The rows are stored as owned by the resolved project.

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
project_name	str | None	Optional project name under which the data are submitted. If None, the CROSS platform infers the project from the caller's memberships, which succeeds only when there is exactly one.	None

Raises:

Type	Description
ValidationError	If the data does not conform to the contract schema.
CrossClientError	If the upload fails. Raised via raise_from_response as a more specific client exception such as ResourceNotFoundError or ConflictError.

Source code in src/crosscontract/crossclient/services/contract_resource.py

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

    The rows are stored as owned by the resolved project.

    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.
        project_name (str | None): Optional project name under which the data
            are submitted. If None, the CROSS platform infers the project from the
            caller's memberships, which succeeds only when there is exactly one.

    Raises:
        ValidationError: If the data does not conform to the contract schema.
        CrossClientError: If the upload fails. Raised via
            `raise_from_response` as a more specific client exception such
            as `ResourceNotFoundError` or `ConflictError`.
    """
    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, project_name=project_name)

change_status(status)

Change the status of the contract.

Parameters:

Name	Type	Description	Default
status	ContractStatus	The new status for the contract.	required

Source code in src/crosscontract/crossclient/services/contract_resource.py

def change_status(self, status: ContractStatus) -> None:
    """Change the status of the contract.

    Args:
        status (ContractStatus): The new status for the contract.
    """
    self._service.change_status(self.name, status)
    self._status = status

delete_data(filters, *, project_name=None, confirm_delete_all=False)

Delete rows from the contract's data matching the given equality filters.

Only rows owned by the resolved project are removed, and the contract must be Active. drop_data() is the wider, separate operation: it drops the whole storage table across every project and requires the contract to be Retired.

Parameters:

Name	Type	Description	Default
filters	dict	Mapping of column name to value (or list of values) to match. Values may be str/int/float/bool. Must be non-empty unless confirm_delete_all is set.	required
project_name	str | None	Optional project name for which project the data are deleted. If None, the CROSS platform infers the project from the caller's memberships, which succeeds only when there is exactly one.	None
confirm_delete_all	bool	Confirms that an unfiltered delete is intended, removing every row the resolved project owns under this contract. Required when filters is empty, so that a filter mapping which collapsed to empty cannot wipe the project's rows by accident. Ignored when filters is non-empty — a filtered delete stays filtered, and the confirmation never reaches the CROSS platform. Defaults to False.	False

Raises:

Type	Description
ValueError	If the contract's cached status is not "Active", or if filters is empty and confirm_delete_all is False. The status check is local and uses the cached status; call refresh() first if the status may have changed on the CROSS platform.
CrossClientError	Propagated from the underlying service/HTTP request if the deletion request fails due to client, server, or network-related errors.

Source code in src/crosscontract/crossclient/services/contract_resource.py

def delete_data(
    self,
    filters: "dict[str, FilterValue | list[FilterValue]]",
    *,
    project_name: str | None = None,
    confirm_delete_all: bool = False,
) -> None:
    """Delete rows from the contract's data matching the given equality filters.

    Only rows owned by the resolved project are removed, and the contract
    must be `Active`. `drop_data()` is the wider, separate operation: it
    drops the whole storage table across every project and requires the
    contract to be `Retired`.

    Args:
        filters (dict): Mapping of column name to value (or list of values)
            to match. Values may be str/int/float/bool. Must be non-empty
            unless `confirm_delete_all` is set.
        project_name (str | None): Optional project name for which project the
            data are deleted. If None, the CROSS platform infers the project
            from the caller's memberships, which succeeds only when there is
            exactly one.
        confirm_delete_all (bool): Confirms that an unfiltered delete is
            intended, removing every row the resolved project owns under
            this contract. Required when `filters` is empty, so that a
            filter mapping which collapsed to empty cannot wipe the
            project's rows by accident. Ignored when `filters` is
            non-empty — a filtered delete stays filtered, and the
            confirmation never reaches the CROSS platform. Defaults to
            `False`.

    Raises:
        ValueError: If the contract's cached status is not `"Active"`, or
            if `filters` is empty and `confirm_delete_all` is False. The
            status check is local and uses the cached status; call
            `refresh()` first if the status may have changed on the CROSS
            platform.
        CrossClientError: Propagated from the underlying service/HTTP
            request if the deletion request fails due to client, server,
            or network-related errors.
    """
    if self._status != "Active":
        raise ValueError(
            f"Cannot delete data from contract '{self.name}': status is "
            f"'{self._status}', must be 'Active'. Call refresh() if the "
            "status may have changed on the server."
        )
    self._service._delete_data(
        self.name,
        filters,
        project_name=project_name,
        confirm_delete_all=confirm_delete_all,
    )

drop_data()

Drop the storage table backing the contract on the CROSS platform.

This is a decommissioning operation: it discards the data of every project that submitted under this contract, not only the caller's, and requires the contract to be Retired. It is restricted to administrators. To remove only the rows owned by one project, use delete_data() — with confirm_delete_all=True to clear that project's rows entirely.

Raises:

Type	Description
CrossClientError	If the request fails. Raised via raise_from_response as a more specific client exception such as ResourceNotFoundError or ConflictError.

Source code in src/crosscontract/crossclient/services/contract_resource.py

def drop_data(self) -> None:
    """Drop the storage table backing the contract on the CROSS platform.

    This is a decommissioning operation: it discards the data of **every**
    project that submitted under this contract, not only the caller's, and
    requires the contract to be `Retired`. It is restricted to
    administrators. To remove only the rows owned by one project, use
    `delete_data()` — with `confirm_delete_all=True` to clear that
    project's rows entirely.

    Raises:
        CrossClientError: If the request fails. Raised via
            `raise_from_response` as a more specific client exception such
            as `ResourceNotFoundError` or `ConflictError`.
    """
    self._service._drop_data_table(self.name)

from_response(service, response_json) classmethod

Build a ContractResource from a raw server response dict.

Source code in src/crosscontract/crossclient/services/contract_resource.py

@classmethod
def from_response(
    cls,
    service: "ContractService",
    response_json: dict[str, Any],
) -> "ContractResource":
    """Build a ContractResource from a raw server response dict."""
    payload = _ContractEntryPayload.model_validate(response_json)
    return cls(service, payload)

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

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
    )

refresh()

Re-fetch the contract details from the CROSS platform.

Source code in src/crosscontract/crossclient/services/contract_resource.py

def refresh(self) -> None:
    """Re-fetch the contract details from the CROSS platform."""
    remote = self._service.get(self.name)
    if remote.name != self.name:
        raise ValueError(
            f"Fetched contract name '{remote.name}' does not match "
            f"resource name '{self.name}'."
        )
    self._contract = remote.contract
    self._status = remote.status
    self._contract_type = remote.contract_type

validate_dataframe(df, check_existing_primary_key=False, check_existing_foreign_key=False, lazy=True)

Validate a DataFrame against the schema of the contract.

By default nothing is read from the CROSS platform and the data is validated on its own. Setting a check flag fetches the values already stored: the primary key is then checked against the union of the stored keys and the DataFrame's own, and the foreign keys against the stored values of the contracts they reference — plus the DataFrame's own rows in the case of a self-referencing foreign key.

Parameters:

Name	Type	Description	Default
df	DataFrame	The DataFrame to validate.	required
check_existing_primary_key	bool	If True, also check the primary key against the values already stored for this contract. Default is False.	False
check_existing_foreign_key	bool	If True, also check the foreign keys against the values already stored for the contracts they reference. Default is False.	False
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.
CrossClientError	If fetching the stored values fails. Raised via raise_from_response as a more specific client exception such as ResourceNotFoundError when the contract has no stored data yet.

Source code in src/crosscontract/crossclient/services/contract_resource.py

def validate_dataframe(
    self,
    df: pd.DataFrame,
    check_existing_primary_key: bool = False,
    check_existing_foreign_key: bool = False,
    lazy: bool = True,
) -> None:
    """Validate a DataFrame against the schema of the contract.

    By default nothing is read from the CROSS platform and the data is
    validated on its own. Setting a check flag fetches the values already
    stored: the primary key is then checked against the union of the stored
    keys and the DataFrame's own, and the foreign keys against the stored
    values of the contracts they reference — plus the DataFrame's own rows
    in the case of a self-referencing foreign key.

    Args:
        df (pd.DataFrame): The DataFrame to validate.
        check_existing_primary_key (bool): If True, also check the primary
            key against the values already stored for this contract.
            Default is False.
        check_existing_foreign_key (bool): If True, also check the foreign
            keys against the values already stored for the contracts they
            reference. 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.
        CrossClientError: If fetching the stored values fails. Raised via
            `raise_from_response` as a more specific client exception such
            as `ResourceNotFoundError` when the contract has no stored data
            yet.
    """
    resolver = CrossContractResolver(self._service)
    try:
        self.contract.validate_data(
            df,
            resolver=resolver,
            check_existing_primary_key=check_existing_primary_key,
            check_existing_foreign_key=check_existing_foreign_key,
            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

Bases: ContractResolver

Reads contracts and their data from the CROSS platform.

Answers the two questions a contract cannot answer on its own: what another contract looks like, and which values are already stored under it. Used when validating data against a contract that references other contracts.

Reaches the platform over HTTP through a ContractService, so it sees whatever the authenticated caller is allowed to read.

Source code in src/crosscontract/crossclient/services/resolver.py

class CrossContractResolver(ContractResolver):
    """Reads contracts and their data from the CROSS platform.

    Answers the two questions a contract cannot answer on its own: what another
    contract looks like, and which values are already stored under it. Used when
    validating data against a contract that references other contracts.

    Reaches the platform over HTTP through a `ContractService`, so it sees
    whatever the authenticated caller is allowed to read.
    """

    def __init__(self, service: ContractService):
        """Initialise the resolver with the service it reads through.

        Args:
            service (ContractService): The service used to reach the platform.
        """
        self._service = service

    def resolve(self, name: str) -> CrossContract | None:
        """Get a contract by name.

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

        Returns:
            CrossContract | None: The contract, or `None` if the platform has no
            contract with that name.

        Raises:
            CrossClientError: If the read fails for any reason other than the
                contract being absent. Only a missing contract becomes `None`; a
                permission error or a server failure propagates as the more
                specific client exception.
        """
        try:
            return self._service.get(name).contract
        except ResourceNotFoundError:
            return None

    def get_data(
        self, name: str, columns: list[str], *, unique: bool = True
    ) -> pd.DataFrame:
        """Get the stored values of the given columns for a contract.

        Args:
            name (str): The name of the contract to read from.
            columns (list[str]): The columns to retrieve.
            unique (bool): Whether to return only unique rows. Defaults to
                `True`, which keeps the response small; the result is the same
                either way.

        Returns:
            pd.DataFrame: The requested columns of the named contract.

        Raises:
            CrossClientError: If the read fails. Raised via
                `raise_from_response` as a more specific client exception such
                as `ResourceNotFoundError`.
        """
        return self._service._get_data(name, columns=columns, unique=unique)

__init__(service)

Initialise the resolver with the service it reads through.

Parameters:

Name	Type	Description	Default
service	ContractService	The service used to reach the platform.	required

Source code in src/crosscontract/crossclient/services/resolver.py

def __init__(self, service: ContractService):
    """Initialise the resolver with the service it reads through.

    Args:
        service (ContractService): The service used to reach the platform.
    """
    self._service = service

get_data(name, columns, *, unique=True)

Get the stored values of the given columns for a contract.

Parameters:

Name	Type	Description	Default
name	str	The name of the contract to read from.	required
columns	list[str]	The columns to retrieve.	required
unique	bool	Whether to return only unique rows. Defaults to True, which keeps the response small; the result is the same either way.	True

Returns:

Type	Description
DataFrame	pd.DataFrame: The requested columns of the named contract.

Raises:

Type	Description
CrossClientError	If the read fails. Raised via raise_from_response as a more specific client exception such as ResourceNotFoundError.

Source code in src/crosscontract/crossclient/services/resolver.py

def get_data(
    self, name: str, columns: list[str], *, unique: bool = True
) -> pd.DataFrame:
    """Get the stored values of the given columns for a contract.

    Args:
        name (str): The name of the contract to read from.
        columns (list[str]): The columns to retrieve.
        unique (bool): Whether to return only unique rows. Defaults to
            `True`, which keeps the response small; the result is the same
            either way.

    Returns:
        pd.DataFrame: The requested columns of the named contract.

    Raises:
        CrossClientError: If the read fails. Raised via
            `raise_from_response` as a more specific client exception such
            as `ResourceNotFoundError`.
    """
    return self._service._get_data(name, columns=columns, unique=unique)

resolve(name)

Get a contract by name.

Parameters:

Name	Type	Description	Default
name	str	The name of the contract.	required

Returns:

Type	Description
CrossContract | None	CrossContract | None: The contract, or None if the platform has no
CrossContract | None	contract with that name.

Raises:

Type	Description
CrossClientError	If the read fails for any reason other than the contract being absent. Only a missing contract becomes None; a permission error or a server failure propagates as the more specific client exception.

Source code in src/crosscontract/crossclient/services/resolver.py

def resolve(self, name: str) -> CrossContract | None:
    """Get a contract by name.

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

    Returns:
        CrossContract | None: The contract, or `None` if the platform has no
        contract with that name.

    Raises:
        CrossClientError: If the read fails for any reason other than the
            contract being absent. Only a missing contract becomes `None`; a
            permission error or a server failure propagates as the more
            specific client exception.
    """
    try:
        return self._service.get(name).contract
    except ResourceNotFoundError:
        return None