Skip to content

services

ApiService

Bases: BaseService

Handles api related requests.

Source code in unkey/services/apis.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
class ApiService(BaseService):
    """Handles api related requests."""

    __slots__ = ()

    async def get_api(self, api_id: str) -> ResultT[models.Api]:
        """Gets information about an api.

        Args:
            api_id: The id of the api.

        Returns:
            A result containing the requested information or an error.
        """
        route = routes.GET_API.compile(api_id)
        data = await self._http.fetch(route)

        if isinstance(data, models.HttpResponse):
            return result.Err(data)

        return result.Ok(self._serializer.to_api(data))

    async def list_keys(
        self, api_id: str, *, owner_id: t.Optional[str] = None, limit: int = 100, offset: int = 0
    ) -> ResultT[models.ApiKeyList]:
        """Gets a paginated list of keys for the given api.

        Args:
            api_id: The id of the api.

        Keyword Args:
            owner_id: The optional owner id to list the keys for.

            limit: The max number of keys to include in this page.
                Defaults to 100.

            offset: How many keys to offset by, for pagination.

        Returns:
            A result containing api key list or an error.
        """
        params = self._generate_map(ownerId=owner_id, limit=limit, offset=offset)
        print(params)
        route = routes.GET_KEYS.compile(api_id).with_params(params)
        data = await self._http.fetch(route)

        if isinstance(data, models.HttpResponse):
            return result.Err(data)

        return result.Ok(self._serializer.to_api_key_list(data))

get_api async 

get_api(api_id: str) -> ResultT[models.Api]

Gets information about an api.

Parameters:

Name Type Description Default
api_id str

The id of the api.

 required

Returns:

Type Description
ResultT[models.Api]

A result containing the requested information or an error.
Source code in unkey/services/apis.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
async def get_api(self, api_id: str) -> ResultT[models.Api]:
    """Gets information about an api.

    Args:
        api_id: The id of the api.

    Returns:
        A result containing the requested information or an error.
    """
    route = routes.GET_API.compile(api_id)
    data = await self._http.fetch(route)

    if isinstance(data, models.HttpResponse):
        return result.Err(data)

    return result.Ok(self._serializer.to_api(data))

list_keys async 

list_keys(
    api_id: str,
    *,
    owner_id: t.Optional[str] = None,
    limit: int = 100,
    offset: int = 0
) -> ResultT[models.ApiKeyList]

Gets a paginated list of keys for the given api.

Parameters:

Name Type Description Default
api_id str

The id of the api.

 required

Other Parameters:

Name Type Description
owner_id t.Optional[str]

The optional owner id to list the keys for.
limit int

The max number of keys to include in this page. Defaults to 100.
offset int

How many keys to offset by, for pagination.

Returns:

Type Description
ResultT[models.ApiKeyList]

A result containing api key list or an error.
Source code in unkey/services/apis.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
async def list_keys(
    self, api_id: str, *, owner_id: t.Optional[str] = None, limit: int = 100, offset: int = 0
) -> ResultT[models.ApiKeyList]:
    """Gets a paginated list of keys for the given api.

    Args:
        api_id: The id of the api.

    Keyword Args:
        owner_id: The optional owner id to list the keys for.

        limit: The max number of keys to include in this page.
            Defaults to 100.

        offset: How many keys to offset by, for pagination.

    Returns:
        A result containing api key list or an error.
    """
    params = self._generate_map(ownerId=owner_id, limit=limit, offset=offset)
    print(params)
    route = routes.GET_KEYS.compile(api_id).with_params(params)
    data = await self._http.fetch(route)

    if isinstance(data, models.HttpResponse):
        return result.Err(data)

    return result.Ok(self._serializer.to_api_key_list(data))

BaseService

Bases: abc.ABC

The base service all API services inherit from.

Parameters:

Name Type Description Default
http_service HttpService

The http service to use for requests.

 required
serializer serializer.Serializer

The serializer to use for handling incoming JSON data from the API.

 required
Source code in unkey/services/base.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
class BaseService(abc.ABC):
    """The base service all API services inherit from.

    Args:
        http_service: The http service to use for requests.

        serializer: The serializer to use for handling incoming
            JSON data from the API.
    """

    __slots__ = ("_http", "_serializer")

    def __init__(self, http_service: HttpService, serializer: serializer.Serializer) -> None:
        self._http = http_service
        self._serializer = serializer

    def _generate_map(self, **kwargs: t.Any) -> t.Dict[str, t.Any]:
        return {k: v for k, v in kwargs.items() if v is not None}

    def _expires_in(
        self, *, milliseconds: int = 0, seconds: int = 0, minutes: int = 0, days: int = 0
    ) -> int | None:
        if not any({milliseconds, seconds, minutes, days}):
            return None

        delta = timedelta(days=days, minutes=minutes, seconds=seconds, milliseconds=milliseconds)
        return int((datetime.now() + delta).timestamp()) * 1000

HttpService

The HTTP service used to make requests to the WOM API.

Parameters:

Name Type Description Default
api_key str

The api key to use.

 required
api_version t.Optional[int]

The optional version of the api to use.

 required
api_base_url t.Optional[str]

The optional api base url to use.

 required
Source code in unkey/services/http.py 
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
class HttpService:
    """The HTTP service used to make requests to the WOM API.

    Args:
        api_key: The api key to use.

        api_version: The optional version of the api to use.

        api_base_url: The optional api base url to use.
    """

    __slots__ = (
        "_api_version",
        "_base_url",
        "_headers",
        "_ok_responses",
        "_method_mapping",
        "_session",
    )

    def __init__(
        self,
        api_key: str,
        api_version: t.Optional[int],
        api_base_url: t.Optional[str],
    ) -> None:
        if not api_key:
            raise ValueError("Api key must be provided.")

        self._headers = {
            "x-user-agent": constants.USER_AGENT,
            "Authorization": f"Bearer {api_key}",
        }

        self._ok_responses = {200, 202}
        self._api_version = f"/v{api_version or 1}"
        self._base_url = api_base_url or constants.API_BASE_URL

    async def _try_get_json(self, response: aiohttp.ClientResponse) -> t.Any:
        try:
            return await response.json()
        except Exception:
            if response.status not in self._ok_responses:
                return models.HttpResponse(response.status, await response.text())

            return await response.text()

    async def _request(
        self, req: t.Callable[..., t.Awaitable[t.Any]], url: str, **kwargs: t.Any
    ) -> t.Any:
        response = await req(url, **kwargs)
        data = await self._try_get_json(response)

        if isinstance(data, models.HttpResponse):
            return data

        # Skipping 404's seems hacky but whatever
        if response.status not in (*self._ok_responses, 404):
            return models.HttpResponse(
                response.status,
                data.get("message", "An unexpected error occurred while making the request."),
            )

        return data

    def _get_request_func(self, method: str) -> t.Callable[..., t.Awaitable[t.Any]]:
        if not hasattr(self, "_method_mapping"):
            raise RuntimeError("HttpService.start was never called, aborting...")

        return self._method_mapping[method]  # type: ignore

    async def _init_session(self) -> None:
        self._session = aiohttp.ClientSession()
        self._method_mapping = {
            constants.GET: self._session.get,
            constants.PUT: self._session.put,
            constants.POST: self._session.post,
            constants.PATCH: self._session.patch,
            constants.DELETE: self._session.delete,
        }

    def set_api_key(self, api_key: str) -> None:
        """Sets the api key used by the http service.

        Args:
            api_key: The new api key to use.
        """
        self._headers["x-api-key"] = api_key

    def set_base_url(self, base_url: str) -> None:
        """Sets the api base url used by the http service.

        Args:
            base_url: The new base url to use.
        """
        self._base_url = base_url

    async def start(self) -> None:
        """Starts the client session to be used by the http service."""
        if not hasattr(self, "_session"):
            await self._init_session()

    async def close(self) -> None:
        """Closes the existing client session, if it's still open."""
        if hasattr(self, "_session") and not self._session.closed:
            await self._session.close()

    async def fetch(
        self,
        route: routes.CompiledRoute,
        *,
        payload: t.Optional[t.Dict[str, t.Any]] = None,
    ) -> dict[str, t.Any] | models.HttpResponse:
        """Fetches the given route.

        Args:
            route: The route to make the request to.

            payload: The optional payload to send in the request body.

        Returns:
            The requested json data or the error response.
        """
        return await self._request(  # type: ignore[no-any-return]
            self._get_request_func(route.method),
            self._base_url + self._api_version + route.uri,
            headers=self._headers,
            params=route.params,
            json=payload or None,
        )

close async 

close() -> None

Closes the existing client session, if it's still open.

Source code in unkey/services/http.py 
118
119
120
121
async def close(self) -> None:
    """Closes the existing client session, if it's still open."""
    if hasattr(self, "_session") and not self._session.closed:
        await self._session.close()

fetch async 

fetch(
    route: routes.CompiledRoute,
    *,
    payload: t.Optional[t.Dict[str, t.Any]] = None
) -> dict[str, t.Any] | models.HttpResponse

Fetches the given route.

Parameters:

Name Type Description Default
route routes.CompiledRoute

The route to make the request to.

 required
payload t.Optional[t.Dict[str, t.Any]]

The optional payload to send in the request body.

 None

Returns:

Type Description
dict[str, t.Any] | models.HttpResponse

The requested json data or the error response.
Source code in unkey/services/http.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
async def fetch(
    self,
    route: routes.CompiledRoute,
    *,
    payload: t.Optional[t.Dict[str, t.Any]] = None,
) -> dict[str, t.Any] | models.HttpResponse:
    """Fetches the given route.

    Args:
        route: The route to make the request to.

        payload: The optional payload to send in the request body.

    Returns:
        The requested json data or the error response.
    """
    return await self._request(  # type: ignore[no-any-return]
        self._get_request_func(route.method),
        self._base_url + self._api_version + route.uri,
        headers=self._headers,
        params=route.params,
        json=payload or None,
    )

set_api_key 

set_api_key(api_key: str) -> None

Sets the api key used by the http service.

Parameters:

Name Type Description Default
api_key str

The new api key to use.

 required
Source code in unkey/services/http.py 
 97
 98
 99
100
101
102
103
def set_api_key(self, api_key: str) -> None:
    """Sets the api key used by the http service.

    Args:
        api_key: The new api key to use.
    """
    self._headers["x-api-key"] = api_key

set_base_url 

set_base_url(base_url: str) -> None

Sets the api base url used by the http service.

Parameters:

Name Type Description Default
base_url str

The new base url to use.

 required
Source code in unkey/services/http.py 
105
106
107
108
109
110
111
def set_base_url(self, base_url: str) -> None:
    """Sets the api base url used by the http service.

    Args:
        base_url: The new base url to use.
    """
    self._base_url = base_url

start async 

start() -> None

Starts the client session to be used by the http service.

Source code in unkey/services/http.py 
113
114
115
116
async def start(self) -> None:
    """Starts the client session to be used by the http service."""
    if not hasattr(self, "_session"):
        await self._init_session()

KeyService

Bases: BaseService

Handles api key related requests.

Source code in unkey/services/keys.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
class KeyService(BaseService):
    """Handles api key related requests."""

    __slots__ = ()

    async def create_key(
        self,
        api_id: str,
        owner_id: str,
        prefix: str,
        *,
        byte_length: int = 16,
        meta: t.Dict[str, t.Any] = {},
        expires: t.Optional[int] = None,
        ratelimit: t.Optional[models.Ratelimit] = None,
    ) -> ResultT[models.ApiKey]:
        """Creates a new api key.

        Args:
            api_id: The id of the api this key is for.

            owner_id: The owner id to use for this key. Represents the
                user who will use this key.

            prefix: The prefix to place at the beginning of the key.

        Keyword Args:
            byte_length: The optional desired length of they in bytes.
                Defaults to 16.

            meta: An optional dynamic mapping of information used to
                provide context around this keys user.

            expires: The optional number of milliseconds into the future
                when this key should expire.

            ratelimit: The optional Ratelimit to set on this key.

        Returns:
            A result containing the requested information or an error.
        """
        route = routes.CREATE_KEY.compile()
        payload = self._generate_map(
            meta=meta,
            apiId=api_id,
            prefix=prefix,
            ownerId=owner_id,
            byteLength=byte_length,
            expires=self._expires_in(milliseconds=expires or 0),
            ratelimit=None
            if not ratelimit
            else self._generate_map(
                limit=ratelimit.limit,
                type=ratelimit.type.value,
                refillRate=ratelimit.refill_rate,
                refillInterval=ratelimit.refill_interval,
            ),
        )

        data = await self._http.fetch(route, payload=payload)

        if isinstance(data, models.HttpResponse):
            return result.Err(data)

        return result.Ok(self._serializer.to_api_key(data))

    async def verify_key(self, key: str) -> ResultT[models.ApiKeyVerification]:
        """Verifies a key is valid and within ratelimit.

        Args:
            key: The key to verify.

        Returns:
            A result containing the api key verification or an error.
        """
        route = routes.VERIFY_KEY.compile()
        payload = self._generate_map(key=key)
        data = await self._http.fetch(route, payload=payload)

        if isinstance(data, models.HttpResponse):
            return result.Err(data)

        return result.Ok(self._serializer.to_api_key_verification(data))

    async def revoke_key(self, key_id: str) -> ResultT[models.HttpResponse]:
        """Revokes a keys validity.

        Args:
            key_id: The id of the key to revoke.

        Returns:
            A result containing the http response or an error.
        """
        route = routes.REVOKE_KEY.compile(key_id)
        data: str | models.HttpResponse = await self._http.fetch(route)  # type: ignore

        if isinstance(data, models.HttpResponse):
            return result.Err(data)

        return result.Ok(models.HttpResponse(202, data))

create_key async 

create_key(
    api_id: str,
    owner_id: str,
    prefix: str,
    *,
    byte_length: int = 16,
    meta: t.Dict[str, t.Any] = {},
    expires: t.Optional[int] = None,
    ratelimit: t.Optional[models.Ratelimit] = None
) -> ResultT[models.ApiKey]

Creates a new api key.

Parameters:

Name Type Description Default
api_id str

The id of the api this key is for.

 required
owner_id str

The owner id to use for this key. Represents the user who will use this key.

 required
prefix str

The prefix to place at the beginning of the key.

 required

Other Parameters:

Name Type Description
byte_length int

The optional desired length of they in bytes. Defaults to 16.
meta t.Dict[str, t.Any]

An optional dynamic mapping of information used to provide context around this keys user.
expires t.Optional[int]

The optional number of milliseconds into the future when this key should expire.
ratelimit t.Optional[models.Ratelimit]

The optional Ratelimit to set on this key.

Returns:

Type Description
ResultT[models.ApiKey]

A result containing the requested information or an error.
Source code in unkey/services/keys.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
async def create_key(
    self,
    api_id: str,
    owner_id: str,
    prefix: str,
    *,
    byte_length: int = 16,
    meta: t.Dict[str, t.Any] = {},
    expires: t.Optional[int] = None,
    ratelimit: t.Optional[models.Ratelimit] = None,
) -> ResultT[models.ApiKey]:
    """Creates a new api key.

    Args:
        api_id: The id of the api this key is for.

        owner_id: The owner id to use for this key. Represents the
            user who will use this key.

        prefix: The prefix to place at the beginning of the key.

    Keyword Args:
        byte_length: The optional desired length of they in bytes.
            Defaults to 16.

        meta: An optional dynamic mapping of information used to
            provide context around this keys user.

        expires: The optional number of milliseconds into the future
            when this key should expire.

        ratelimit: The optional Ratelimit to set on this key.

    Returns:
        A result containing the requested information or an error.
    """
    route = routes.CREATE_KEY.compile()
    payload = self._generate_map(
        meta=meta,
        apiId=api_id,
        prefix=prefix,
        ownerId=owner_id,
        byteLength=byte_length,
        expires=self._expires_in(milliseconds=expires or 0),
        ratelimit=None
        if not ratelimit
        else self._generate_map(
            limit=ratelimit.limit,
            type=ratelimit.type.value,
            refillRate=ratelimit.refill_rate,
            refillInterval=ratelimit.refill_interval,
        ),
    )

    data = await self._http.fetch(route, payload=payload)

    if isinstance(data, models.HttpResponse):
        return result.Err(data)

    return result.Ok(self._serializer.to_api_key(data))

revoke_key async 

revoke_key(key_id: str) -> ResultT[models.HttpResponse]

Revokes a keys validity.

Parameters:

Name Type Description Default
key_id str

The id of the key to revoke.

 required

Returns:

Type Description
ResultT[models.HttpResponse]

A result containing the http response or an error.
Source code in unkey/services/keys.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
async def revoke_key(self, key_id: str) -> ResultT[models.HttpResponse]:
    """Revokes a keys validity.

    Args:
        key_id: The id of the key to revoke.

    Returns:
        A result containing the http response or an error.
    """
    route = routes.REVOKE_KEY.compile(key_id)
    data: str | models.HttpResponse = await self._http.fetch(route)  # type: ignore

    if isinstance(data, models.HttpResponse):
        return result.Err(data)

    return result.Ok(models.HttpResponse(202, data))

verify_key async 

verify_key(key: str) -> ResultT[models.ApiKeyVerification]

Verifies a key is valid and within ratelimit.

Parameters:

Name Type Description Default
key str

The key to verify.

 required

Returns:

Type Description
ResultT[models.ApiKeyVerification]

A result containing the api key verification or an error.
Source code in unkey/services/keys.py 
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
async def verify_key(self, key: str) -> ResultT[models.ApiKeyVerification]:
    """Verifies a key is valid and within ratelimit.

    Args:
        key: The key to verify.

    Returns:
        A result containing the api key verification or an error.
    """
    route = routes.VERIFY_KEY.compile()
    payload = self._generate_map(key=key)
    data = await self._http.fetch(route, payload=payload)

    if isinstance(data, models.HttpResponse):
        return result.Err(data)

    return result.Ok(self._serializer.to_api_key_verification(data))