ApiBackend

Fetch and store data from an API endpoint.

Overview
base_url
authentication
headers
record_headers
create_headers
update_headers
delete_headers
count_headers
model_casing
api_casing
api_to_model_map
pagination_parameter_name
pagination_parameter_type
limit_parameter_name
can_create
can_update
can_delete
can_query
can_count
response_adapter
response_adapter_dependency_name
url_adapter
url_adapter_dependency_name
pagination_adapter
pagination_adapter_dependency_name
count_adapter
count_adapter_dependency_name
url_suffix

Overview

The ApiBackend gives developers a way to quickly build SDKs to connect a clearskies applications to arbitrary API endpoints. The backend has some built in flexibility to make it easy to connect it to most APIs, as well as behavioral hooks so that you can override small sections of the logic to accommodate APIs that don’t work in the expected way. This allows you to interact with APIs using the standard model methods, just like every other backend, and also means that you can attach such models to endpoints to quickly enable all kinds of pre-defined behaviors.

Usage

Configuring the API backend is pretty easy:

Provide the base_url to the constructor, or extend it and set it in the __init__ for the new backend.
Provide a clearskies.authentication.Authentication object, assuming it isn’t a public API.
Match your model class name to the path of the API (or set model.destination_name() appropriately)
Use the resulting model like you would any other model!

It’s important to understand how the Api Backend will map queries and saves to the API in question. The rules are fairly simple:

The API backend only supports searching with the equals operator (e.g. models.where("column=value")).
To specify routing parameters, use the {parameter_name} or :parameter_name syntax in either the url or in the destination name of your model. In order to query the model, you then must provide a value for any routing parameters, using a matching search condition: (e.g. models.where("routing_parameter_name=value"))
Any search clauses that don’t correspond to routing parameters will be translated into query parameters. So, if your destination_name is https://example.com/:categoy_id/products and you executed a model query: models.where("category_id=10").where("on_sale=1") then this would result in fetching a URL of https://example.com/10/products?on_sale=1
When you specifically search on the id column for the model, the id will be appended to the end of the URL rather than as a query parameter. So, with a destination name of https://example.com/products, querying for models.find("id=10") will result in fetching https://example.com/products/10.
Delete and Update operations will similarly append the id to the URL, and also set the appropriate response method (e.g. DELETE or PATCH by default).
When processing the response, the backend will attempt to automatically discover the results by looking for dictionaries that contain the expected column names (as determined from the model schema and the mapping rules).
The backend will check for a response header called link and parse this to find pagination information so it can iterate through records.

NOTE: The API backend doesn’t support joins or group_by clauses. This limitation, as well as the fact that it only supports seaching with the equals operator, isn’t a limitation in the API backend itself, but simply reflects the behavior of most API endoints. If you want to support an API that has more flexibility (for instance, perhaps it allows for more search operations than just =), then you can extend the appropritae methods, discussed below, to map a model query to an API request.

Here’s an example of how to use the API Backend to integrate with the Github API:

import clearskies


class GithubPublicBackend(clearskies.backends.ApiBackend):
    def __init__(
        self,
        # This varies from endpoint to endpoint, so we want to be able to set it for each model
        pagination_parameter_name: str = "since",
    ):
        # these are fixed for all gitlab API parameters, so there's no need to make them setable
        # from the constructor
        self.base_url = "https://api.github.com"
        self.limit_parameter_name = "per_page"
        self.pagination_parameter_name = pagination_parameter_name
        self.finalize_and_validate_configuration()


class UserRepo(clearskies.Model):
    # Corresponding API Docs: https://docs.github.com/en/rest/repos/repos?apiVersion=2022-11-28#list-repositories-for-a-user
    id_column_name = "full_name"
    backend = GithubPublicBackend(pagination_parameter_name="page")

    @classmethod
    def destination_name(cls) -> str:
        return "users/:login/repos"

    id = clearskies.columns.Integer()
    full_name = clearskies.columns.String()
    type = clearskies.columns.Select(["all", "owner", "member"])
    url = clearskies.columns.String()
    html_url = clearskies.columns.String()
    created_at = clearskies.columns.Datetime()
    updated_at = clearskies.columns.Datetime()

    # The API endpoint won't return "login" (e.g. username), so it may not seem like a column, but we need to search by it
    # because it's a URL parameter for this API endpoint.  Clearskies uses strict validation and won't let us search by
    # a column that doesn't exist in the model: therefore, we have to add the login column.
    login = clearskies.columns.String(is_searchable=True, is_readable=False)

    # The API endpoint let's us sort by `created`/`updated`.  Note that the names of the columns (based on the data returned
    # by the API endpoint) are `created_at`/`updated_at`.  As above, clearskies strictly validates data, so we need columns
    # named created/updated so that we can sort by them.  We can set some flags to (hopefully) avoid confusion
    updated = clearskies.columns.Datetime(
        is_searchable=False, is_readable=False, is_writeable=False
    )
    created = clearskies.columns.Datetime(
        is_searchable=False, is_readable=False, is_writeable=False
    )


class User(clearskies.Model):
    # Corresponding API docs: https://docs.github.com/en/rest/users/users?apiVersion=2022-11-28#list-users

    # github has two columns that are both effecitvely id columns: id and login.
    # We use the login column for id_column_name because that is the column that gets
    # used in the API to fetch an individual record
    id_column_name = "login"
    backend = GithubPublicBackend()

    id = clearskies.columns.Integer()
    login = clearskies.columns.String()
    gravatar_id = clearskies.columns.String()
    avatar_url = clearskies.columns.String()
    html_url = clearskies.columns.String()
    repos_url = clearskies.columns.String()

    # We can hook up relationships between models just like we would if we were using an SQL-like
    # database.  The whole point of the backend system is that the model queries work regardless of
    # backend, so clearskies can issue API calls to fetch related records just like it would be able
    # to fetch children from a related database table.
    repos = clearskies.columns.HasMany(
        UserRepo,
        foreign_column_name="login",
        readable_child_columns=["id", "full_name", "html_url"],
    )


def fetch_user(users: User, user_repos: UserRepo):
    # If we execute this models query:
    some_repos = (
        user_repos.where("login=cmancone")
        .sort_by("created", "desc")
        .where("type=owner")
        .pagination(page=2)
        .limit(5)
    )
    # the API backend will fetch this url:
    # https://api.github.com/users/cmancone/repos?type=owner&sort=created&direction=desc&per_page=5&page=2
    # and we can use the results like always
    repo_names = [repo.full_name for repo in some_repos]

    # For the below case, the backend will fetch this url:
    # https://api.github.com/users/cmancone
    # in addition, the readable column names on the callable endpoint includes "repos", which references our has_many
    # column.  This means that when converting the user model to JSON, it will also grab a page of repositories for that user.
    # To do that, it will fetch this URL:
    # https://api.github.com/users/cmancone/repos
    return users.find("login=cmancone")


wsgi = clearskies.contexts.WsgiRef(
    clearskies.endpoints.Callable(
        fetch_user,
        model_class=User,
        readable_column_names=["id", "login", "html_url", "repos"],
    ),
    classes=[User, UserRepo],
)

if __name__ == "__main__":
    wsgi()

The following example demonstrates how models using this backend can be used in other clearskies endpoints, just like any other model. Note that the following example is re-using the above models and backend, I have just omitted them for the sake of brevity:

wsgi = clearskies.contexts.WsgiRef(
    clearskies.endpoints.List(
        model_class=User,
        readable_column_names=["id", "login", "html_url"],
        sortable_column_names=["id"],
        default_sort_column_name=None,
        default_limit=10,
    ),
    classes=[User],
)

if __name__ == "__main__":
    wsgi()

And if you invoke it:

$ curl 'http://localhost:8080' | jq
{
    "status": "success",
    "error": "",
    "data": [
        {
            "id": 1,
            "login": "mojombo",
            "html_url": "https://github.com/mojombo"
        },
        {
            "id": 2,
            "login": "defunkt",
            "html_url": "https://github.com/defunkt"
        },
        {
            "id": 3,
            "login": "pjhyett",
            "html_url": "https://github.com/pjhyett"
        },
        {
            "id": 4,
            "login": "wycats",
            "html_url": "https://github.com/wycats"
        },
        {
            "id": 5,
            "login": "ezmobius",
            "html_url": "https://github.com/ezmobius"
        },
        {
            "id": 6,
            "login": "ivey",
            "html_url": "https://github.com/ivey"
        },
        {
            "id": 7,
            "login": "evanphx",
            "html_url": "https://github.com/evanphx"
        },
        {
            "id": 17,
            "login": "vanpelt",
            "html_url": "https://github.com/vanpelt"
        },
        {
            "id": 18,
            "login": "wayneeseguin",
            "html_url": "https://github.com/wayneeseguin"
        },
        {
            "id": 19,
            "login": "brynary",
            "html_url": "https://github.com/brynary"
        }
    ],
    "pagination": {
        "number_results": null,
        "limit": 10,
        "next_page": {
            "since": "19"
        }
    },
    "input_errors": {}
}

In essence, we now have an endpoint that lists results but, instead of pulling its data from a database, it makes API calls. It also tracks pagination as expected, so you can use the data in pagination.next_page to fetch the next set of results, just as you would if this were backed by a database, e.g.:

$ curl http://localhost:8080?since=19

Mapping from Queries to API calls

The process of mapping a model query into an API request involves a few different methods which can be overwritten to fully control the process. This is necessary in cases where an API behaves differently than expected by the API backend. This table outlines the method involved and how they are used:

Method	Description
records_url	Return the absolute URL to fetch, as well as any columns that were used to fill in routing parameters
records_method	Reurn the HTTP request method to use for the API call
conditions_to_request_parameters	Translate the query conditions into URL fragments, query parameters, or JSON body parameters
pagination_to_request_parameters	Translate the pagination data into URL fragments, query parameters, or JSON body parameters
sorts_to_request_parameters	Translate the sort directive(s) into URL fragments, query parameters, or JSON body parameters
map_records_response	Take the response from the API and return a list of dictionaries with the resulting records

In short, the details of the query are stored in a clearskies.query.Query object which is passed around to these various methods. They use that information to adjust the URL, add query parameters, or add parameters into the JSON body. The API Backend will then execute an API call with those final details, and use the map_record_response method to pull the returned records out of the response from the API endpoint.

base_url

Required

The Base URL for the requests - will be prepended to the destination_name() from the model.

Note: this is treated as a ‘folder’ path: if set, it becomes the URL prefix and is followed with a ‘/’

authentication

Optional

An instance of clearskies.authentication.Authentication that handles authentication to the API.

The following example is a modification of the Github Backends used above that shows how to setup authentication. Github, like many APIs, uses an API key attached to the request via the authorization header. The SecretBearer authentication class in clearskies is designed for this common use case, and pulls the secret key out of either an environment variable or the secret manager (I use the former in this case, because it’s hard to have a self-contained example with a secret manager). Of course, any authentication method can be attached to your API backend - SecretBearer authentication is used here simply because it’s a common approach.

Note that, when used in conjunction with a secret manager, the API Backend and the SecretBearer class will work together to check for a new secret in the event of an authentication failure from the API endpoint (specifically, a 401 error). This allows you to automate credential rotation: create a new API key, put it in the secret manager, and then revoke the old API key. The next time an API call is made, the SecretBearer will provide the old key from it’s cache and the request will fail. The API backend will detect this and try the request again, but this time will tell the SecretBearer class to refresh it’s cache with a fresh copy of the key from the secrets manager. Therefore, as long as you put the new key in your secret manager before disabling the old key, this second request will succeed and the service will continue to operate successfully with only a slight delay in response time caused by refreshing the cache.

import clearskies

class GithubBackend(clearskies.backends.ApiBackend):
    def __init__(
        self,
        pagination_parameter_name: str = "page",
        authentication: clearskies.authentication.Authentication | None = None,
    ):
        self.base_url = "https://api.github.com"
        self.limit_parameter_name = "per_page"
        self.pagination_parameter_name = pagination_parameter_name
        self.authentication = clearskies.authentication.SecretBearer(
            environment_key="GITHUB_API_KEY",
            header_prefix="Bearer ", # Because github expects a header of 'Authorization: Bearer API_KEY'
        )
        self.finalize_and_validate_configuration()

class Repo(clearskies.Model):
    id_column_name = "login"
    backend = GithubBackend()

    @classmethod
    def destination_name(cls):
        return "/user/repos"

    id = clearskies.columns.Integer()
    name = clearskies.columns.String()
    full_name = clearskies.columns.String()
    html_url = clearskies.columns.String()
    visibility = clearskies.columns.Select(["all", "public", "private"])

wsgi = clearskies.contexts.WsgiRef(
    clearskies.endpoints.List(
        model_class=Repo,
        readable_column_names=["id", "name", "full_name", "html_url"],
        sortable_column_names=["full_name"],
        default_sort_column_name="full_name",
        default_limit=10,
        where=["visibility=private"],
    ),
    classes=[Repo],
)

if __name__ == "__main__":
    wsgi()

headers

Optional

A dictionary of headers to attach to all outgoing API requests

record_headers

Optional

A dictionary of headers to attach to record fetch requests (GET). If not set, falls back to headers.

create_headers

Optional

A dictionary of headers to attach to create requests (POST). If not set, falls back to headers.

update_headers

Optional

A dictionary of headers to attach to update requests (PATCH/PUT). If not set, falls back to headers.

delete_headers

Optional

A dictionary of headers to attach to delete requests (DELETE). If not set, falls back to headers.

count_headers

Optional

A dictionary of headers to attach to count requests. If not set, falls back to headers.

model_casing

Optional

The casing used in the model (snake_case, camelCase, TitleCase)

This is used in conjunction with api_casing to tell the processing layer when you and the API are using different casing standards. The API backend will then automatically covnert the casing style of the API to match your model. This can be helpful when you have a standard naming convention in your own code which some external API doesn’t follow, that way you can at least standardize things in your code. In the following example, these parameters are used to convert from the snake_casing native to the Github API into the TitleCasing used in the model class:

import clearskies

class User(clearskies.Model):
    id_column_name = "login"
    backend = clearskies.backends.ApiBackend(
        base_url="https://api.github.com",
        limit_parameter_name="per_page",
        pagination_parameter_name="since",
        model_casing="TitleCase",
        api_casing="snake_case",
    )

    Id = clearskies.columns.Integer()
    Login = clearskies.columns.String()
    GravatarId = clearskies.columns.String()
    AvatarUrl = clearskies.columns.String()
    HtmlUrl = clearskies.columns.String()
    ReposUrl = clearskies.columns.String()

wsgi = clearskies.contexts.WsgiRef(
    clearskies.endpoints.List(
        model_class=User,
        readable_column_names=["Login", "AvatarUrl", "HtmlUrl", "ReposUrl"],
        sortable_column_names=["Id"],
        default_sort_column_name=None,
        default_limit=2,
        internal_casing="TitleCase",
        external_casing="TitleCase",
    ),
    classes=[User],
)

if __name__ == "__main__":
    wsgi()

and when executed:

$ curl http://localhost:8080 | jq
{
    "Status": "Success",
    "Error": "",
    "Data": [
        {
            "Login": "mojombo",
            "AvatarUrl": "https://avatars.githubusercontent.com/u/1?v=4",
            "HtmlUrl": "https://github.com/mojombo",
            "ReposUrl": "https://api.github.com/users/mojombo/repos"
        },
        {
            "Login": "defunkt",
            "AvatarUrl": "https://avatars.githubusercontent.com/u/2?v=4",
            "HtmlUrl": "https://github.com/defunkt",
            "ReposUrl": "https://api.github.com/users/defunkt/repos"
        }
    ],
    "Pagination": {
        "NumberResults": null,
        "Limit": 2,
        "NextPage": {
            "Since": "2"
        }
    },
    "InputErrors": {}
}

api_casing

Optional

The casing used by the API response (snake_case, camelCase, TitleCase)

See model_casing for details and usage.

api_to_model_map

Optional

A mapping from the data keys returned by the API to the data keys expected in the model

This comes into play when you want your model columns to use different names than what is returned by the API itself. Provide a dictionary where the key is the name of a piece of data from the API, and the value is the name of the column in the model. The API Backend will use this to match the API data to your model. In the example below, html_url from the API has been mapped to profile_url in the model:

import clearskies

class User(clearskies.Model):
    id_column_name = "login"
    backend = clearskies.backends.ApiBackend(
        base_url="https://api.github.com",
        limit_parameter_name="per_page",
        pagination_parameter_name="since",
        api_to_model_map={"html_url": "profile_url"},
    )

    id = clearskies.columns.Integer()
    login = clearskies.columns.String()
    profile_url = clearskies.columns.String()

wsgi = clearskies.contexts.WsgiRef(
    clearskies.endpoints.List(
        model_class=User,
        readable_column_names=["login", "profile_url"],
        sortable_column_names=["id"],
        default_sort_column_name=None,
        default_limit=2,
    ),
    classes=[User],
)

if __name__ == "__main__":
    wsgi()

And if you invoke it:

$ curl http://localhost:8080 | jq
{
    "status": "success",
    "error": "",
    "data": [
        {
            "login": "mojombo",
            "profile_url": "https://github.com/mojombo"
        },
        {
            "login": "defunkt",
            "profile_url": "https://github.com/defunkt"
        }
    ],
    "pagination": {
        "number_results": null,
        "limit": 2,
        "next_page": {
            "since": "2"
        }
    },
    "input_errors": {}
}

pagination_parameter_name

Optional

The name of the pagination parameter

pagination_parameter_type

Optional

The expected ‘type’ of the pagination parameter: must be either ‘int’ or ‘str’

Note: this is set as a literal string, not as a type.

limit_parameter_name

Optional

The name of the parameter that sets the number of records per page (if empty, setting the page size will not be allowed)

can_create

Optional

Whether creating new records is allowed for this backend.

When set to False, any attempt to create a record will raise a ValueError. This can be set as a class attribute or passed to the constructor.

can_update

Optional

Whether updating existing records is allowed for this backend.

When set to False, any attempt to update a record will raise a ValueError. This can be set as a class attribute or passed to the constructor.

can_delete

Optional

Whether deleting records is allowed for this backend.

When set to False, any attempt to delete a record will raise a ValueError. This can be set as a class attribute or passed to the constructor.

can_query

Optional

Whether querying/reading records is allowed for this backend.

When set to False, any attempt to query records (via iteration, count, find, etc.) will raise a ValueError. This can be set as a class attribute or passed to the constructor.

can_count

Optional

Whether this backend supports count operations via len(model) and bool(model).

When can_count is True, calling len() or bool() on a model will issue a lightweight HEAD request to the records URL and extract the count from response headers (e.g. X-Total-Count) via the count_adapter. A HEAD request returns only headers — no response body — so this is much cheaper than fetching all records.

Additionally, count information is always extracted from response headers when fetching records via records(), so after iterating over a model the count is cached and available without an additional request.

When can_count is False (the default), len() and bool() on models will raise NotImplementedError, and endpoints won’t include number_results in the pagination response.

backend = clearskies.backends.ApiBackend(
    base_url="https://api.example.com",
    can_count=True,
)

response_adapter

Optional

A :class:~clearskies.backends.ResponseAdapter that handles extracting records and single records from the raw API response before the model-mapping pipeline runs.

The adapter answers the structural question — where is the data inside the envelope? — while api_to_model_map and map_to_model continue to answer the field-level question (what are the fields called?) as before.

Provide a :class:~clearskies.backends.ResponseAdapter subclass for full control, or a plain callable (response_data: Any) -> list | dict | None for simple unwrapping::

# Full adapter — different logic per operation:
backend = clearskies.backends.ApiBackend(
    base_url="https://api.example.com",
    response_adapter=MyHalJsonResponseAdapter(),
)

# Callable — same extractor for both list and single-record responses:
backend = clearskies.backends.ApiBackend(
    base_url="https://api.example.com",
    response_adapter=lambda data: data.get("items"),
)

Return None from either the adapter method or the callable to fall through to the built-in ApiBackend extraction logic.

If not provided, clearskies checks DI for a dependency named response_adapter and uses it when available.

response_adapter_dependency_name

Optional

Dependency name used to lazily resolve a response adapter from DI when response_adapter is not explicitly configured.

url_adapter

Optional

Calculate the URL to use for an update request.  Also, return the list of any query parameters used to construct the URL.

See finalize_url for more details on the return value.

When a callable is provided as the ``url_adapter``, it receives ``(model, id, data)`` and can return either
a plain URL string or a tuple of ``(url, used_routing_parameters)``.  If it returns just a URL string, no
routing parameters are reported as consumed, so all data keys remain in the request body.  If it returns a
tuple, the second element lists the parameter names that were absorbed into the URL and should be removed
from the request body before sending the API request::

    # Simple callable — just return a URL string:
    url_adapter = lambda model, id, data: f"/api/v2/{model.destination_name()}/{id}"

    # Callable with routing parameter reporting:
    url_adapter = lambda model, id, data: (
        f"/api/v2/tenants/{data['tenant_id']}/{model.destination_name()}/{id}",
        ["tenant_id"],
    )

url_adapter_dependency_name

Optional

Dependency name used to lazily resolve a URL adapter from DI when url_adapter is not explicitly configured.

pagination_adapter

Optional

Extract pagination data from the API response via adapter.

This method has a very important job, which is to inform clearskies about how to make another API call to fetch the next
page of records.  It returns a dictionary with whatever pagination information is necessary.

Returns:
    A dictionary containing pagination data (e.g., cursor, page number, total counts).
    Returns an empty dict if there is no next page.

pagination_adapter_dependency_name

Optional

Dependency name used to lazily resolve a pagination adapter from DI when pagination_adapter is not explicitly configured.

count_adapter

Optional

Extract count information from API response via the count adapter.

Delegates to the configured ``count_adapter`` to extract total record count and total page
count from the API response.  The default ``CountAdapter`` checks for common count headers
used by REST APIs:

- ``X-Total-Count`` or ``X-Total`` for total record count
- ``X-Total-Pages`` for total pages

You can customize this by providing a ``count_adapter`` when configuring the backend, or by
overriding this method in a subclass:

```python
def extract_count_from_response(
    self,
    response_headers: dict[str, str] | None = None,
    response_data: Any = None,
) -> tuple[int | None, int | None]:
    if not response_headers:
        return (None, None)
    # Custom header names for your API
    total = response_headers.get("X-My-Api-Total")
    pages = response_headers.get("X-My-Api-Pages")
    if total is not None:
        return (int(total), int(pages) if pages else None)
    return (None, None)
```

count_adapter_dependency_name

Optional

Dependency name used to lazily resolve a count adapter from DI when count_adapter is not explicitly configured.

url_suffix

Optional

A suffix to append to the end of the URL.

Note: this is treated as a ‘folder’ path: if set, it becomes the URL suffix and is prefixed with a ‘/’