SnykBackend

Backend for interacting with the Snyk REST API.

1. Overview
2. base_url
3. api_version
4. authentication
5. model_casing
6. api_casing
7. api_to_model_map
8. pagination_parameter_name
9. pagination_parameter_type
10. limit_parameter_name
11. can_create
12. can_update
13. can_delete
14. can_query
15. headers
16. update_headers
17. create_headers
18. delete_headers
19. records_headers
20. resource_type

Overview

This backend extends the ApiBackend to provide seamless integration with the Snyk platform. It handles the specific pagination and response format used by Snyk REST APIs, where pagination uses cursor-based navigation with starting_after parameter.

The Snyk REST API uses JSON:API format, so responses have a data key containing records with id, type, and attributes fields. This backend automatically flattens these into a simple dictionary format expected by clearskies models.

Usage

The SnykBackend is typically used with models that represent Snyk entities:

import clearskies
from clearskies_snyk.backends import SnykBackend


class SnykOrg(clearskies.Model):
    backend = SnykBackend()

    @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 = SnykBackend(
    authentication=clearskies.authentication.SecretBearer(
        environment_key="SNYK_API_KEY",
        header_prefix="token ",
    )
)

Pagination

The Snyk REST API uses cursor-based pagination with the following response format:

{
    "data": [...],
    "links": {
        "next": "/rest/orgs?starting_after=abc123&version=2024-10-15"
    }
}

The backend automatically handles extracting pagination data and provides the next page information to clearskies for seamless iteration through results.

API Version

The Snyk REST API requires a version parameter. By default, this is set to “2024-10-15”. You can override this by setting the api_version parameter.

Relationship Mapping

The backend automatically extracts relationship IDs from JSON:API relationships. For example, if a record has:

{
    "relationships": {
        "organization": {
            "data": {"id": "org-123", "type": "org"}
        }
    }
}

The backend will add organization_id: "org-123" to the flattened record.

JSON:API Resource Type

When creating or updating records, the backend needs to know the JSON:API resource type (e.g., “project”, “org”, etc.). By default, it will try to infer this from the model’s destination_name by taking the last path segment and singularizing it. However, you can explicitly set this using the resource_type parameter:

backend = SnykBackend(resource_type="project")

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.

api_version

Optional

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.

headers

Optional

A dictionary of headers to attach to all outgoing API requests

update_headers

Optional

A dictionary of headers to attach to update requests (PATCH/PUT). 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.

delete_headers

Optional

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

records_headers

Optional

resource_type

Optional

Map create data to JSON:API format required by Snyk REST API.

The Snyk REST API expects: {"data": {"type": "...", "attributes": {...}}}

This hook is called by the ApiBackend.create() method to transform the data before
sending it to the API.