SnykV1Backend

Backend for interacting with the Snyk v1 API.

Overview
base_url
authentication
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

Overview

This backend extends the ApiBackend to provide seamless integration with the Snyk platform’s legacy v1 API. It handles the specific response format and pagination used by Snyk v1 APIs.

The v1 API uses camelCase field names (automatically converted to snake_case) and returns responses with various wrapper keys like orgs, projects, etc.

Usage

The SnykV1Backend is typically used with models that represent Snyk entities available through the v1 API:

import clearskies
from clearskies_snyk.backends import SnykV1Backend


class SnykOrgV1(clearskies.Model):
    backend = SnykV1Backend()

    @classmethod
    def destination_name(cls) -> str:
        return "orgs"

    id = clearskies.columns.String()
    name = clearskies.columns.String()
    slug = clearskies.columns.String()

Authentication

By default, the backend uses the snyk_auth binding for authentication, which should be configured in your application’s dependency injection container. You can also provide a custom authentication instance:

backend = SnykV1Backend(
    authentication=clearskies.authentication.SecretBearer(
        environment_key="SNYK_API_KEY",
        header_prefix="token ",
    )
)

Response Format

The Snyk v1 API returns responses in various formats depending on the endpoint:

{"orgs": [...]} for organization lists
{"projects": [...]} for project lists
{"snapshots": [...]} for project history
Direct list [...] for some endpoints
Single object {...} for single record endpoints

This backend automatically handles these variations.

Pagination

The Snyk v1 API uses offset-based pagination with page and perPage parameters.

base_url

Optional

Given a URL, this will append the base URL, fill in any routing data, and also return any used routing parameters.

For example, consider a base URL of `/my/api/{record_id}/:other_id` and then this is called as so:

```python
(url, used_routing_parameters) = api_backend.finalize_url(
    "entries",
    {
        "record_id": "1-2-3-4",
        "other_id": "a-s-d-f",
        "more_things": "qwerty",
    },
)
```

The returned url would be `/my/api/1-2-3-4/a-s-d-f/entries`, and used_routing_parameters would be ["record_id", "other_id"].
The latter is returned so you can understand what parameters were absorbed into the URL.  Often, when some piece of data
becomes a routing parameter, it needs to be ignored in the rest of the request.  `used_routing_parameters` helps with that.

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()

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.