Skip to content

services

ApiService

Bases: BaseService

Handles api related requests.

Source code in unkey/services/apis.py 
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
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.
        """
        params = self._generate_map(apiId=api_id)
        route = routes.GET_API.compile().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(data))

    async def list_keys(
        self,
        api_id: str,
        *,
        owner_id: UndefinedOr[str] = UNDEFINED,
        limit: UndefinedOr[int] = UNDEFINED,
        cursor: UndefinedOr[str] = UNDEFINED,
    ) -> 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 optional max number of keys to include in this page.

            cursor: Optional key used to determine pagination offset.

        Returns:
            A result containing api key list or an error.
        """
        params = self._generate_map(apiId=api_id, ownerId=owner_id, limit=limit, cursor=cursor)
        route = routes.GET_KEYS.compile().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 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
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.
    """
    params = self._generate_map(apiId=api_id)
    route = routes.GET_API.compile().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(data))

list_keys async 

list_keys(
    api_id: str,
    *,
    owner_id: UndefinedOr[str] = UNDEFINED,
    limit: UndefinedOr[int] = UNDEFINED,
    cursor: UndefinedOr[str] = UNDEFINED
) -> 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 UndefinedOr[str]

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

The optional max number of keys to include in this page.
cursor UndefinedOr[str]

Optional key used to determine pagination offset.

Returns:

Type Description
ResultT[models.ApiKeyList]

A result containing api key list or an error.
Source code in unkey/services/apis.py 
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
async def list_keys(
    self,
    api_id: str,
    *,
    owner_id: UndefinedOr[str] = UNDEFINED,
    limit: UndefinedOr[int] = UNDEFINED,
    cursor: UndefinedOr[str] = UNDEFINED,
) -> 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 optional max number of keys to include in this page.

        cursor: Optional key used to determine pagination offset.

    Returns:
        A result containing api key list or an error.
    """
    params = self._generate_map(apiId=api_id, ownerId=owner_id, limit=limit, cursor=cursor)
    route = routes.GET_KEYS.compile().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 
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
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 undefined.UNDEFINED}

    def _expires_in(
        self, *, milliseconds: int = 0, seconds: int = 0, minutes: int = 0, days: int = 0
    ) -> undefined.UndefinedOr[int]:
        if not any({milliseconds, seconds, minutes, days}):
            return undefined.UNDEFINED

        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 unkey API.

Parameters:

Name Type Description Default
api_key t.Optional[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
146
147
148
149
150
151
152
class HttpService:
    """The HTTP service used to make requests to the unkey 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: t.Optional[str],
        api_version: t.Optional[int],
        api_base_url: t.Optional[str],
    ) -> None:
        if api_key == "":
            raise ValueError("Api key must not be empty.")

        self._headers = {
            "Unkey-SDK": constants.USER_AGENT,
            "User-Agent": constants.USER_AGENT,
            "x-user-agent": constants.USER_AGENT,
        }

        if api_key:
            self._headers["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

        if response.status not in self._ok_responses:
            error: t.Union[t.Any, t.Dict[str, t.Any]] = data.get("error")
            is_dict = isinstance(error, dict)

            message = error.get("message") if is_dict else error
            code = models.ErrorCode.from_str_maybe(error.get("code") if is_dict else "UNKNOWN")
            message = message or "An unexpected error occurred while making the request."

            return models.HttpResponse(response.status, str(message), code=code)

        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") or self._session.closed:
            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 
125
126
127
128
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 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
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 
104
105
106
107
108
109
110
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 
112
113
114
115
116
117
118
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 
120
121
122
123
async def start(self) -> None:
    """Starts the client session to be used by the http service."""
    if not hasattr(self, "_session") or self._session.closed:
        await self._init_session()

KeyService

Bases: BaseService

Handles api key related requests.

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
 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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
class KeyService(BaseService):
    """Handles api key related requests."""

    __slots__ = ()

    async def create_key(
        self,
        api_id: str,
        owner_id: str,
        prefix: str,
        *,
        name: UndefinedOr[str] = UNDEFINED,
        byte_length: UndefinedOr[int] = UNDEFINED,
        meta: UndefinedOr[t.Dict[str, t.Any]] = UNDEFINED,
        expires: UndefinedOr[int] = UNDEFINED,
        remaining: UndefinedOr[int] = UNDEFINED,
        ratelimit: UndefinedOr[models.Ratelimit] = UNDEFINED,
        refill: UndefinedOr[models.Refill] = UNDEFINED,
    ) -> ResultT[models.ApiKey]:
        """Creates a new api key.

        Args:
            name: The name to use for this key.

            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.

            remaining: The optional max number of times this key can be
                used. Useful for creating long lived keys but with a
                limit on total uses.

            ratelimit: The optional `Ratelimit` to set on this key.

            refill: The optional `Refill` to set on this key.

        Returns:
            A result containing the newly created key or an error.
        """
        route = routes.CREATE_KEY.compile()
        payload = self._generate_map(
            meta=meta,
            name=name,
            apiId=api_id,
            prefix=prefix,
            ownerId=owner_id,
            remaining=remaining,
            byteLength=byte_length,
            expires=self._expires_in(milliseconds=expires or 0),
            ratelimit=UNDEFINED
            if not ratelimit
            else self._generate_map(
                limit=ratelimit.limit,
                type=ratelimit.type.value,
                refillRate=ratelimit.refill_rate,
                refillInterval=ratelimit.refill_interval,
            ),
            refill=UNDEFINED
            if not refill
            else self._generate_map(
                amount=refill.amount,
                interval=refill.interval.value,
            ),
        )

        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, api_id: str) -> ResultT[models.ApiKeyVerification]:
        """Verifies a key is valid and within ratelimit.

        Args:
            key: The key to verify.

            api_id: The id of the api to verify the key against.

        Returns:
            A result containing the api key verification or an error.
        """
        route = routes.VERIFY_KEY.compile()
        payload = self._generate_map(key=key, apiId=api_id)
        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()
        payload = self._generate_map(keyId=key_id)
        data = await self._http.fetch(route, payload=payload)

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

        return result.Ok(models.HttpResponse(200, "OK"))

    async def update_key(
        self,
        key_id: str,
        *,
        name: UndefinedNoneOr[str] = UNDEFINED,
        owner_id: UndefinedNoneOr[str] = UNDEFINED,
        meta: UndefinedNoneOr[t.Dict[str, t.Any]] = UNDEFINED,
        expires: UndefinedNoneOr[int] = UNDEFINED,
        remaining: UndefinedNoneOr[int] = UNDEFINED,
        ratelimit: UndefinedNoneOr[models.Ratelimit] = UNDEFINED,
        refill: UndefinedOr[models.Refill] = UNDEFINED,
    ) -> ResultT[models.HttpResponse]:
        """Updates an existing api key. To delete a key set its value
        to `None`.

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

        Keyword Args:
            name: The new name to use for this key.

            owner_id: The new owner id to use for this key.

            meta: The new dynamic mapping of information used
                to provide context around this keys user.

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

            remaining: The new max number of times this key can be
                used.

            ratelimit: The new `Ratelimit` to set on this key.

            refill: The optional `Refill` to set on this key.

        Returns:
            A result containing the OK response or an error.
        """
        if all_undefined(name, owner_id, meta, expires, remaining, ratelimit, refill):
            raise errors.MissingRequiredArgument("At least one value is required to be updated.")

        route = routes.UPDATE_KEY.compile()
        payload = self._generate_map(
            name=name,
            meta=meta,
            keyId=key_id,
            ownerId=owner_id,
            remaining=remaining,
            expires=self._expires_in(milliseconds=expires or 0),
            ratelimit=UNDEFINED
            if not ratelimit
            else self._generate_map(
                limit=ratelimit.limit,
                type=ratelimit.type.value,
                refillRate=ratelimit.refill_rate,
                refillInterval=ratelimit.refill_interval,
            ),
            refill=UNDEFINED
            if not refill
            else self._generate_map(
                amount=refill.amount,
                interval=refill.interval.value,
            ),
        )

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

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

        return result.Ok(models.HttpResponse(200, "OK"))

    async def get_key(self, key_id: str) -> ResultT[models.ApiKeyMeta]:
        """Retrieves details for the given key.

        Args:
            key_id: The id of the key.

        Returns:
            A result containing the api key metadata or an error.
        """
        params = self._generate_map(keyId=key_id)
        route = routes.GET_KEY.compile().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_meta(data))

    async def update_remaining(
        self, key_id: str, value: t.Optional[int], op: models.UpdateOp
    ) -> ResultT[t.Optional[int]]:
        """Updates a keys remaining limit.

        Args:
            key_id: The id of the key.

            value: The value to perform the operation on.

            op: The update operation to perform.

        Returns:
            A result containing the new remaining limit of the key or an error.
        """
        payload = self._generate_map(keyId=key_id, value=value, op=op.value)
        route = routes.UPDATE_REMAINING.compile()
        data = await self._http.fetch(route, payload=payload)

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

        return result.Ok(data["remaining"])

create_key async 

create_key(
    api_id: str,
    owner_id: str,
    prefix: str,
    *,
    name: UndefinedOr[str] = UNDEFINED,
    byte_length: UndefinedOr[int] = UNDEFINED,
    meta: UndefinedOr[t.Dict[str, t.Any]] = UNDEFINED,
    expires: UndefinedOr[int] = UNDEFINED,
    remaining: UndefinedOr[int] = UNDEFINED,
    ratelimit: UndefinedOr[models.Ratelimit] = UNDEFINED,
    refill: UndefinedOr[models.Refill] = UNDEFINED
) -> ResultT[models.ApiKey]

Creates a new api key.

Parameters:

Name Type Description Default
name UndefinedOr[str]

The name to use for this key.

 UNDEFINED
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 UndefinedOr[int]

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

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

The optional number of milliseconds into the future when this key should expire.
remaining UndefinedOr[int]

The optional max number of times this key can be used. Useful for creating long lived keys but with a limit on total uses.
ratelimit UndefinedOr[models.Ratelimit]

The optional Ratelimit to set on this key.
refill UndefinedOr[models.Refill]

The optional Refill to set on this key.

Returns:

Type Description
ResultT[models.ApiKey]

A result containing the newly created key or an error.
Source code in unkey/services/keys.py 
 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
async def create_key(
    self,
    api_id: str,
    owner_id: str,
    prefix: str,
    *,
    name: UndefinedOr[str] = UNDEFINED,
    byte_length: UndefinedOr[int] = UNDEFINED,
    meta: UndefinedOr[t.Dict[str, t.Any]] = UNDEFINED,
    expires: UndefinedOr[int] = UNDEFINED,
    remaining: UndefinedOr[int] = UNDEFINED,
    ratelimit: UndefinedOr[models.Ratelimit] = UNDEFINED,
    refill: UndefinedOr[models.Refill] = UNDEFINED,
) -> ResultT[models.ApiKey]:
    """Creates a new api key.

    Args:
        name: The name to use for this key.

        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.

        remaining: The optional max number of times this key can be
            used. Useful for creating long lived keys but with a
            limit on total uses.

        ratelimit: The optional `Ratelimit` to set on this key.

        refill: The optional `Refill` to set on this key.

    Returns:
        A result containing the newly created key or an error.
    """
    route = routes.CREATE_KEY.compile()
    payload = self._generate_map(
        meta=meta,
        name=name,
        apiId=api_id,
        prefix=prefix,
        ownerId=owner_id,
        remaining=remaining,
        byteLength=byte_length,
        expires=self._expires_in(milliseconds=expires or 0),
        ratelimit=UNDEFINED
        if not ratelimit
        else self._generate_map(
            limit=ratelimit.limit,
            type=ratelimit.type.value,
            refillRate=ratelimit.refill_rate,
            refillInterval=ratelimit.refill_interval,
        ),
        refill=UNDEFINED
        if not refill
        else self._generate_map(
            amount=refill.amount,
            interval=refill.interval.value,
        ),
    )

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

get_key async 

get_key(key_id: str) -> ResultT[models.ApiKeyMeta]

Retrieves details for the given key.

Parameters:

Name Type Description Default
key_id str

The id of the key.

 required

Returns:

Type Description
ResultT[models.ApiKeyMeta]

A result containing the api key metadata or an error.
Source code in unkey/services/keys.py 
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
async def get_key(self, key_id: str) -> ResultT[models.ApiKeyMeta]:
    """Retrieves details for the given key.

    Args:
        key_id: The id of the key.

    Returns:
        A result containing the api key metadata or an error.
    """
    params = self._generate_map(keyId=key_id)
    route = routes.GET_KEY.compile().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_meta(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 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
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()
    payload = self._generate_map(keyId=key_id)
    data = await self._http.fetch(route, payload=payload)

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

    return result.Ok(models.HttpResponse(200, "OK"))

update_key async 

update_key(
    key_id: str,
    *,
    name: UndefinedNoneOr[str] = UNDEFINED,
    owner_id: UndefinedNoneOr[str] = UNDEFINED,
    meta: UndefinedNoneOr[t.Dict[str, t.Any]] = UNDEFINED,
    expires: UndefinedNoneOr[int] = UNDEFINED,
    remaining: UndefinedNoneOr[int] = UNDEFINED,
    ratelimit: UndefinedNoneOr[
        models.Ratelimit
    ] = UNDEFINED,
    refill: UndefinedOr[models.Refill] = UNDEFINED
) -> ResultT[models.HttpResponse]

Updates an existing api key. To delete a key set its value to None.

Parameters:

Name Type Description Default
key_id str

The id of the key to update.

 required

Other Parameters:

Name Type Description
name UndefinedNoneOr[str]

The new name to use for this key.
owner_id UndefinedNoneOr[str]

The new owner id to use for this key.
meta UndefinedNoneOr[t.Dict[str, t.Any]]

The new dynamic mapping of information used to provide context around this keys user.
expires UndefinedNoneOr[int]

The new number of milliseconds into the future when this key should expire.
remaining UndefinedNoneOr[int]

The new max number of times this key can be used.
ratelimit UndefinedNoneOr[models.Ratelimit]

The new Ratelimit to set on this key.
refill UndefinedOr[models.Refill]

The optional Refill to set on this key.

Returns:

Type Description
ResultT[models.HttpResponse]

A result containing the OK response or an error.
Source code in unkey/services/keys.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
async def update_key(
    self,
    key_id: str,
    *,
    name: UndefinedNoneOr[str] = UNDEFINED,
    owner_id: UndefinedNoneOr[str] = UNDEFINED,
    meta: UndefinedNoneOr[t.Dict[str, t.Any]] = UNDEFINED,
    expires: UndefinedNoneOr[int] = UNDEFINED,
    remaining: UndefinedNoneOr[int] = UNDEFINED,
    ratelimit: UndefinedNoneOr[models.Ratelimit] = UNDEFINED,
    refill: UndefinedOr[models.Refill] = UNDEFINED,
) -> ResultT[models.HttpResponse]:
    """Updates an existing api key. To delete a key set its value
    to `None`.

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

    Keyword Args:
        name: The new name to use for this key.

        owner_id: The new owner id to use for this key.

        meta: The new dynamic mapping of information used
            to provide context around this keys user.

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

        remaining: The new max number of times this key can be
            used.

        ratelimit: The new `Ratelimit` to set on this key.

        refill: The optional `Refill` to set on this key.

    Returns:
        A result containing the OK response or an error.
    """
    if all_undefined(name, owner_id, meta, expires, remaining, ratelimit, refill):
        raise errors.MissingRequiredArgument("At least one value is required to be updated.")

    route = routes.UPDATE_KEY.compile()
    payload = self._generate_map(
        name=name,
        meta=meta,
        keyId=key_id,
        ownerId=owner_id,
        remaining=remaining,
        expires=self._expires_in(milliseconds=expires or 0),
        ratelimit=UNDEFINED
        if not ratelimit
        else self._generate_map(
            limit=ratelimit.limit,
            type=ratelimit.type.value,
            refillRate=ratelimit.refill_rate,
            refillInterval=ratelimit.refill_interval,
        ),
        refill=UNDEFINED
        if not refill
        else self._generate_map(
            amount=refill.amount,
            interval=refill.interval.value,
        ),
    )

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

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

    return result.Ok(models.HttpResponse(200, "OK"))

update_remaining async 

update_remaining(
    key_id: str, value: t.Optional[int], op: models.UpdateOp
) -> ResultT[t.Optional[int]]

Updates a keys remaining limit.

Parameters:

Name Type Description Default
key_id str

The id of the key.

 required
value t.Optional[int]

The value to perform the operation on.

 required
op models.UpdateOp

The update operation to perform.

 required

Returns:

Type Description
ResultT[t.Optional[int]]

A result containing the new remaining limit of the key or an error.
Source code in unkey/services/keys.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
async def update_remaining(
    self, key_id: str, value: t.Optional[int], op: models.UpdateOp
) -> ResultT[t.Optional[int]]:
    """Updates a keys remaining limit.

    Args:
        key_id: The id of the key.

        value: The value to perform the operation on.

        op: The update operation to perform.

    Returns:
        A result containing the new remaining limit of the key or an error.
    """
    payload = self._generate_map(keyId=key_id, value=value, op=op.value)
    route = routes.UPDATE_REMAINING.compile()
    data = await self._http.fetch(route, payload=payload)

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

    return result.Ok(data["remaining"])

verify_key async 

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

Verifies a key is valid and within ratelimit.

Parameters:

Name Type Description Default
key str

The key to verify.

 required
api_id str

The id of the api to verify the key against.

 required

Returns:

Type Description
ResultT[models.ApiKeyVerification]

A result containing the api key verification or an error.
Source code in unkey/services/keys.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
async def verify_key(self, key: str, api_id: str) -> ResultT[models.ApiKeyVerification]:
    """Verifies a key is valid and within ratelimit.

    Args:
        key: The key to verify.

        api_id: The id of the api to verify the key against.

    Returns:
        A result containing the api key verification or an error.
    """
    route = routes.VERIFY_KEY.compile()
    payload = self._generate_map(key=key, apiId=api_id)
    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))