API client implementation

This guide describes the conventions for implementing a Foundry DevTools APIClient.

The api_name attribute

Every APIClient implementation needs to set the class attribute api_name.

The api_request method

The api_request method is a special request function, that can be used to easily make API requests to Foundry.

    def api_request(
        self,
        method: str | bytes,
        api_path: str,
        params: _Params | None = None,
        data: _Data | None = None,
        headers: dict | None = None,
        cookies: RequestsCookieJar | _TextMapping | None = None,
        files: _Files | None = None,
        auth: _Auth | None = None,
        timeout: _Timeout | None = None,
        allow_redirects: bool = True,
        proxies: _TextMapping | None = None,
        hooks: _HooksInput | None = None,
        stream: bool | None = None,
        verify: _Verify | None = None,
        cert: _Cert | None = None,
        json: Incomplete | None = None,
        error_handling: ErrorHandlingConfig | Literal[False] | None = None,
    ) -> Response:
        """Make an authenticated request to the Foundry API.

        The `api_path` argument is only the api path and not the full URL.
        For https://foundry/example/api/method/... this would be only,
        "method/...".

        Args:
            method: see :py:meth:`requests.Session.request`
            api_path: **only** the api path
            params: see :py:meth:`requests.Session.request`
            data: see :py:meth:`requests.Session.request`
            headers: see :py:meth:`requests.Session.request`, content-type defaults to application/json if not set
            cookies: see :py:meth:`requests.Session.request`
            files: see :py:meth:`requests.Session.request`
            auth: see :py:meth:`foundry_dev_tools.clients.context_client.ContextHTTPClient.auth_handler`
            timeout: see :py:meth:`requests.Session.request`
            allow_redirects: see :py:meth:`requests.Session.request`
            proxies: see :py:meth:`requests.Session.request`
            hooks: see :py:meth:`requests.Session.request`
            stream: see :py:meth:`requests.Session.request`
            verify: see :py:meth:`requests.Session.request`
            cert: see :py:meth:`requests.Session.request`
            json: see :py:meth:`requests.Session.request`
            error_handling: error handling config; if set to False, errors won't be automatically handled
        """
        if headers:
            headers["content-type"] = headers.get("content-type") or headers.get("Content-Type") or "application/json"
        else:
            headers = {"content-type": "application/json"}

        return self.context.client.request(
            method=method,
            url=self.api_url(api_path),
            params=params,
            data=data,
            headers=headers,
            cookies=cookies,
            files=files,
            auth=auth,
            timeout=timeout,
            allow_redirects=allow_redirects,
            proxies=proxies,
            hooks=hooks,
            stream=stream,
            verify=verify,
            cert=cert,
            json=json,
            error_handling=error_handling,
        )

The method parameter gets passed to the ContextHTTPClient of the FoundryContext associated with the APIClient. The api_path parameter builds the correct URL for that API with the build_api_url method.

@cache
def build_api_url(url: str, api_name: str, api_path: str) -> str:
    """Cached function for building the api URLs."""
    return url + "/" + api_name + "/api/" + api_path

This convenience function just appends the Foundry url to the api_name of the client and appends the api_path.

Warning

The ‘Content-Type’ is per default ‘application/json’, if you want another ‘Content-Type’ you need to set it in the headers parameter.

Naming conventions

Classes

The APIClient implementations should (not a must) be called <Api_name>Client, omitting “Foundry” from the class name is okay. For example the API client for “foundry-catalog” is called CatalogClient instead of FoundryCatalogClient.

Methods

Direct API calls

If it is just a wrapper around an API and does nothing else than calling the api_request method. The method’s name should begin with ‘api_’ followed by the name of the api.

For example the DataProxyClient.api_get_file method:

    def api_get_file(
        self,
        dataset_rid: DatasetRid,
        transaction_rid: TransactionRid,
        logical_path: PathInDataset,
        range_header: str | None = None,
        requests_stream: bool = True,
        **kwargs,
    ) -> requests.Response:
        """Returns a file from the specified dataset and transaction.

        Args:
            dataset_rid: dataset rid
            transaction_rid: transaction rid
            logical_path: path in dataset
            range_header: HTTP range header
            requests_stream: passed to :py:meth:`requests.Session.request` as `stream`
            **kwargs: gets passed to :py:meth:`APIClient.api_request`
        """
        return self.api_request(
            "GET",
            f"dataproxy/datasets/{dataset_rid}/transactions/{transaction_rid}/{quote(logical_path)}",
            headers={"Range": range_header} if range_header else None,
            stream=requests_stream,
            **kwargs,
        )

The parameters of the method should be very similar to the original parameters name. The **kwargs should be passed to the api_request method, for custom error handling. These methods always return the requests.Response of the API call.

Higher level methods

Methods which do more than just calling the API like CatalogClient.list_dataset_files:

    def list_dataset_files(
        self,
        dataset_rid: api_types.DatasetRid,
        end_ref: api_types.View = "master",
        page_size: int = 1000,
        logical_path: api_types.PathInDataset | None = None,
        page_start_logical_path: api_types.PathInDataset | None = None,
        include_open_exclusive_transaction: bool = False,
        exclude_hidden_files: bool = False,
        temporary_credentials_auth_token: str | None = None,
    ) -> list:
        """Same as :py:meth:`CatalogClient.api_get_dataset_view_files3`, but iterates through all pages.

        Args:
            dataset_rid: the dataset rid
            end_ref: branch or transaction rid of the dataset
            page_size: the maximum page size returned
            logical_path: If logical_path is absent, returns all files in the view.
                If logical_path matches a file exactly, returns just that file.
                Otherwise, returns all files in the "directory" of logical_path:
                (a slash is added to the end of logicalPath if necessary and a prefix-match is performed)
            page_start_logical_path: if specified page starts at the given path,
                otherwise at the beginning of the file list
            include_open_exclusive_transaction: if files added in open transaction should be returned
                as well in the response
            exclude_hidden_files: if hidden files should be excluded (e.g. _log files)
            temporary_credentials_auth_token: to generate temporary credentials for presigned URLs

        Returns:
            list[FileResourcesPage]:
                .. code-block:: python

                    [
                        {
                            "logicalPath": "..",
                            "pageStartLogicalPath": "..",
                            "includeOpenExclusiveTransaction": "..",
                            "excludeHiddenFiles": "..",
                        },
                    ]
        """

        def _inner_get(next_page_token: str | None = None) -> dict:
            return self.api_get_dataset_view_files3(
                dataset_rid=dataset_rid,
                end_ref=end_ref,
                page_size=page_size,
                logical_path=logical_path,
                page_start_logical_path=page_start_logical_path,
                include_open_exclusive_transaction=include_open_exclusive_transaction,
                exclude_hidden_files=exclude_hidden_files,
                start_transaction_rid=next_page_token,
                temporary_credentials_auth_token=temporary_credentials_auth_token,
            ).json()

        result: list[dict] = []
        batch_result = {"nextPageToken": ""}
        while batch_result["nextPageToken"] is not None:
            batch_result = _inner_get(next_page_token=batch_result["nextPageToken"])
            result.extend(batch_result["values"])  # type: ignore[arg-type]
        return result

These methods don’t have specific naming convention and can return anything they want.

API typing

As you may have seen in the examples above, there are types like DatasetRid, PathInDataset, and more. These are not “real” types. They are all exactly the same as the str class. These are encouraged to use, for better readability.

Here dataset could be anything:

def api_xyz(self, dataset:str): ...

Now we know, it is a path on foundry:

def api_xyz(self,dataset:FoundryPath): ...

Both of these methods can be called via api_xyz("/xyz"). You don’t need to convert strings to this type, calling e.g. PathInDataset("/xyz") is exactly the same as str("/xyz"), which just returns "/xyz" again.

Example

Let’s imagine there is an api called “info”. This does not exist and is completely for illustrative purposes, to show what goes into writing an API client for Foundry with Foundry DevTools.

We start of by creating the class InfoClient, and set the api_name attribute to info.

from __future__ import annotations  # this way types don't get 'executed' at runtime
from foundry_dev_tools.clients.api_client import APIClient
from typing import TYPE_CHECKING  # special variable that is only `True` if a type checker is 'looking' at the code
from foundry_dev_tools.errors.handling import ErrorHandlingConfig  # we will need this later for error handling
from foundry_dev_tools.errors.info import (
    UserNotFoundError,
    UserAlreadyExistsError,
    InsufficientPermissionsError,
)  # the errors we will create in the next step

if TYPE_CHECKING:
    # will only be imported if type checking, and not at runtime
    import requests


class InfoClient(APIClient):
    api_name = "info"

Errors

First of we will define the Exceptions that the API provides, in a seperate file.

UserNotFound

from foundry_dev_tools.errors.meta import FoundryAPIError

class UserNotFoundError(FoundryAPIError):
    message = "The username provided does not exist."

UserAlreadyExists

class UserNotFoundError(FoundryAPIError):
    message = "The username already exists."

InsufficientPermissions

class InsufficientPermissionsError(FoundryAPIError):
    message = "You don't have sufficient permissions to use this API."

API Methods

Each API method on the info endpoint, will be its own low level method.

Get User Info

URL: /user

HTTP Method: GET

Query Parameters:

• username: for which user to get information

    def api_get_user_info(self, username: str, **kwargs) -> requests.Response:
        """Returns information about the user.

        Args:
            username: for which user to get information
        """
        return self.api_request(
            "GET",
            "user",
            params={"username": username},
            error_handling=ErrorHandlingConfig(
                {
                    "Info:UserNotFoundError": UserNotFoundError,
                    "Info:InsufficientPermissions": InsufficientPermissionsError,
                }
            ),
            **kwargs,
        )

Change User Info

URL: /user

HTTP Method: POST

Query Parameters:

• username: the name of the user where the information should be changed

Body:

JSON Body with the following definition:

Key Value object, where the key is the information name, and the value is the information value.

Following information names exist:

• org: organization name

• isAdmin: controls if the user is an admin

• enabled: controls if the users account is enabled

If a key is not in the post body it will not be changed.

    def api_change_user_info(
        self,
        username: str,
        org: str | None = None,
        is_admin: bool | None = None,
        enabled: bool | None = None,
        **kwargs,
    ) -> requests.Response:
        """Change user information.

        Args:
            username: the name of the user where the information should be changed
            org: organization name
            is_admin: controls if the user is an admin
            enabled: controls if the users account is enabled
        """
        body = {}
        # We only want to change provided values
        if org is not None:
            body["org"] = org
        if is_admin is not None:
            body["isAdmin"] = is_admin
        if enabled is not None:
            body["enabled"] = enabled
        return self.api_request(
            "POST",
            "user",
            params={"username": username},
            json=body,
            error_handling=ErrorHandlingConfig(
                {
                    "Info:UserNotFoundError": UserNotFoundError,
                    "Info:InsufficientPermissions": InsufficientPermissionsError,
                }
            ),
            **kwargs,
        )

List users

URL: /users

HTTP Method: GET

    def api_list_users(self, **kwargs) -> requests.Response:
        """List all users."""
        return self.api_request(
            "GET",
            "users",
            error_handling=ErrorHandlingConfig({"Info:InsufficientPermissions": InsufficientPermissionsError}),
            **kwargs,
        )

Create User

URL: /users

HTTP Method: POST

Body and Query with the same definition as change-user-info, except that org is mandatory, and isAdmin is per default false, and enabled per default true if omitted.

    def api_create_user(
        self,
        username: str,
        org: str,
        is_admin: bool | None = None,
        enabled: bool | None = None,
        **kwargs,
    ) -> requests.Response:
        """Create a user.

        Args:
            username: name of the user to be created
            org: organization name
            is_admin: controls if the user is an admin, defaults to false
            enabled: controls if the users account is enabled, defaults to true
        """
        body = {"org": org}
        # We only want to set provided values
        if is_admin is not None:
            body["isAdmin"] = is_admin
        if enabled is not None:
            body["enabled"] = enabled
        return self.api_request(
            "POST",
            "users",
            params={"username": username},
            json=body,
            error_handling=ErrorHandlingConfig(
                {
                    "Info:UserAlreadyExistsError": UserAlreadyExistsError,
                    "Info:InsufficientPermissions": InsufficientPermissionsError,
                }
            ),
            **kwargs,
        )