# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.

from __future__ import annotations

from typing import Optional

from typing_extensions import Literal

import httpx

from ..types import (

api_key_list_params,

api_key_create_params,

api_key_rotate_params,

api_key_update_params,

api_key_retrieve_params,

)

from .._types import Body, Omit, Query, Headers, NoneType, NotGiven, omit, not_given

from .._utils import path_template, maybe_transform, async_maybe_transform

from .._compat import cached_property

from .._resource import SyncAPIResource, AsyncAPIResource

from .._response import (

to_raw_response_wrapper,

to_streamed_response_wrapper,

async_to_raw_response_wrapper,

async_to_streamed_response_wrapper,

)

from ..pagination import SyncOffsetPagination, AsyncOffsetPagination

from .._base_client import AsyncPaginator, make_request_options

from ..types.api_key import APIKey

from ..types.created_api_key import CreatedAPIKey

__all__ = ["APIKeysResource", "AsyncAPIKeysResource"]

class APIKeysResource(SyncAPIResource):

"""Create and manage API keys for organization and project-scoped access."""

@cached_property

def with_raw_response(self) -> APIKeysResourceWithRawResponse:

"""

This property can be used as a prefix for any HTTP method call to return

the raw response object instead of the parsed content.

For more information, see https://www.github.com/kernel/kernel-python-sdk#accessing-raw-response-data-eg-headers

"""

return APIKeysResourceWithRawResponse(self)

@cached_property

def with_streaming_response(self) -> APIKeysResourceWithStreamingResponse:

"""

An alternative to `.with_raw_response` that doesn't eagerly read the response body.

For more information, see https://www.github.com/kernel/kernel-python-sdk#with_streaming_response

"""

return APIKeysResourceWithStreamingResponse(self)

def create(

self,

*,

name: str,

days_to_expire: Optional[int] | Omit = omit,

project_id: Optional[str] | Omit = omit,

# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.

# The extra values given here take precedence over values defined on the client or passed to this method.

extra_headers: Headers | None = None,

extra_query: Query | None = None,

extra_body: Body | None = None,

timeout: float | httpx.Timeout | None | NotGiven = not_given,

) -> CreatedAPIKey:

"""

Create a new API key within the authenticated organization.

Args:

name: Label for the API key (1-255 characters). API keys are not addressable by name.

days_to_expire: Number of days until expiry, up to 3650. Use null for never.

project_id: Unique project identifier

extra_headers: Send extra headers

extra_query: Add additional query parameters to the request

extra_body: Add additional JSON properties to the request

timeout: Override the client-level default timeout for this request, in seconds

"""

return self._post(

"/org/api_keys",

body=maybe_transform(

{

"name": name,

"days_to_expire": days_to_expire,

"project_id": project_id,

},

api_key_create_params.APIKeyCreateParams,

),

options=make_request_options(

extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout

),

cast_to=CreatedAPIKey,

)

def retrieve(

self,

id: str,

*,

include_deleted: bool | Omit = omit,

# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.

# The extra values given here take precedence over values defined on the client or passed to this method.

extra_headers: Headers | None = None,

extra_query: Query | None = None,

extra_body: Body | None = None,

timeout: float | httpx.Timeout | None | NotGiven = not_given,

) -> APIKey:

"""Retrieve an API key by ID for the authenticated organization.

API keys are

masked.

Args:

include_deleted: When true, return the API key even if it has been deleted (soft-deleted), for

audit purposes. Defaults to false, which returns 404 for a deleted key.

extra_headers: Send extra headers

extra_query: Add additional query parameters to the request

extra_body: Add additional JSON properties to the request

timeout: Override the client-level default timeout for this request, in seconds

"""

if not id:

raise ValueError(f"Expected a non-empty value for `id` but received {id!r}")

return self._get(

path_template("/org/api_keys/{id}", id=id),

options=make_request_options(

extra_headers=extra_headers,

extra_query=extra_query,

extra_body=extra_body,

timeout=timeout,

query=maybe_transform(

{"include_deleted": include_deleted}, api_key_retrieve_params.APIKeyRetrieveParams

),

),

cast_to=APIKey,

)

def update(

self,

id: str,

*,

name: str,

# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.

# The extra values given here take precedence over values defined on the client or passed to this method.

extra_headers: Headers | None = None,

extra_query: Query | None = None,

extra_body: Body | None = None,

timeout: float | httpx.Timeout | None | NotGiven = not_given,

) -> APIKey:

"""

Update an API key's name.

Args:

name: New API key name

extra_headers: Send extra headers

extra_query: Add additional query parameters to the request

extra_body: Add additional JSON properties to the request

timeout: Override the client-level default timeout for this request, in seconds

"""

if not id:

raise ValueError(f"Expected a non-empty value for `id` but received {id!r}")

return self._patch(

path_template("/org/api_keys/{id}", id=id),

body=maybe_transform({"name": name}, api_key_update_params.APIKeyUpdateParams),

options=make_request_options(

extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout

),

cast_to=APIKey,

)

def list(

self,

*,

include_deleted: bool | Omit = omit,

limit: int | Omit = omit,

name: str | Omit = omit,

offset: int | Omit = omit,

query: str | Omit = omit,

sort_by: Literal["created_at", "name", "expires_at"] | Omit = omit,

sort_direction: Literal["asc", "desc"] | Omit = omit,

status: Literal["active", "deleted", "all"] | Omit = omit,

# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.

# The extra values given here take precedence over values defined on the client or passed to this method.

extra_headers: Headers | None = None,

extra_query: Query | None = None,

extra_body: Body | None = None,

timeout: float | httpx.Timeout | None | NotGiven = not_given,

) -> SyncOffsetPagination[APIKey]:

"""List API keys for the authenticated organization.

API keys are masked.

Args:

include_deleted: Deprecated: use status=all instead. When true, include deleted (soft-deleted)

API keys in the results for audit purposes.

limit: Maximum number of results to return

name: Exact-match filter on API key name using the database collation. In production,

matching is case- and accent-insensitive. Names are not required to be unique,

so multiple keys may match. When status=all or include_deleted=true is set,

soft-deleted keys with the same name may also match.

offset: Number of results to skip

query: Case-insensitive substring match against API key name, creator, and project. API

key identifiers and masked keys match by exact value or prefix.

sort_by: Field to sort API keys by.

sort_direction: Sort direction for API keys.

status: Filter API keys by status. "active" returns keys that are not deleted (default;

expired-but-not-deleted keys are still included), "deleted" returns only

soft-deleted keys, "all" returns both.

extra_headers: Send extra headers

extra_query: Add additional query parameters to the request

extra_body: Add additional JSON properties to the request

timeout: Override the client-level default timeout for this request, in seconds

"""

return self._get_api_list(

"/org/api_keys",

page=SyncOffsetPagination[APIKey],

options=make_request_options(

extra_headers=extra_headers,

extra_query=extra_query,

extra_body=extra_body,

timeout=timeout,

query=maybe_transform(

{

"include_deleted": include_deleted,

"limit": limit,

"name": name,

"offset": offset,

"query": query,

"sort_by": sort_by,

"sort_direction": sort_direction,

"status": status,

},

api_key_list_params.APIKeyListParams,

),

),

model=APIKey,

)

def delete(

self,

id: str,

*,

# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.

# The extra values given here take precedence over values defined on the client or passed to this method.

extra_headers: Headers | None = None,

extra_query: Query | None = None,

extra_body: Body | None = None,

timeout: float | httpx.Timeout | None | NotGiven = not_given,

) -> None:

"""Delete an API key.

A key cannot delete itself; use a different key to delete

this one.

Args:

extra_headers: Send extra headers

extra_query: Add additional query parameters to the request

extra_body: Add additional JSON properties to the request

timeout: Override the client-level default timeout for this request, in seconds

"""

if not id:

raise ValueError(f"Expected a non-empty value for `id` but received {id!r}")

extra_headers = {"Accept": "*/*", **(extra_headers or {})}

return self._delete(

path_template("/org/api_keys/{id}", id=id),

options=make_request_options(

extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout

),

cast_to=NoneType,

)

def rotate(

self,

id: str,

*,

days_to_expire: Optional[int] | Omit = omit,

expire_in_days: Optional[int] | Omit = omit,

# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.

# The extra values given here take precedence over values defined on the client or passed to this method.

extra_headers: Headers | None = None,

extra_query: Query | None = None,

extra_body: Body | None = None,

timeout: float | httpx.Timeout | None | NotGiven = not_given,

) -> CreatedAPIKey:

"""Rotate an API key.

Issues a new key that copies the name and project of the

rotated key, and schedules the rotated key to expire after a grace period so

in-flight callers can swap over. The new plaintext key is returned once.

Args:

days_to_expire: Lifetime in days for the new key, up to 3650. Omit to reuse the rotated key's

original lifetime, or never-expires if it had none.

expire_in_days: Grace period in days before the rotated key expires. Use 0 to expire it

immediately. Omit for the default grace period of 7 days.

extra_headers: Send extra headers

extra_query: Add additional query parameters to the request

extra_body: Add additional JSON properties to the request

timeout: Override the client-level default timeout for this request, in seconds

"""

if not id:

raise ValueError(f"Expected a non-empty value for `id` but received {id!r}")

return self._post(

path_template("/org/api_keys/{id}/rotate", id=id),

body=maybe_transform(

{

"days_to_expire": days_to_expire,

"expire_in_days": expire_in_days,

},

api_key_rotate_params.APIKeyRotateParams,

),

options=make_request_options(

extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout

),

cast_to=CreatedAPIKey,

)

class AsyncAPIKeysResource(AsyncAPIResource):

"""Create and manage API keys for organization and project-scoped access."""

@cached_property

def with_raw_response(self) -> AsyncAPIKeysResourceWithRawResponse:

"""

This property can be used as a prefix for any HTTP method call to return

the raw response object instead of the parsed content.

For more information, see https://www.github.com/kernel/kernel-python-sdk#accessing-raw-response-data-eg-headers

"""

return AsyncAPIKeysResourceWithRawResponse(self)

@cached_property

def with_streaming_response(self) -> AsyncAPIKeysResourceWithStreamingResponse:

"""

An alternative to `.with_raw_response` that doesn't eagerly read the response body.

For more information, see https://www.github.com/kernel/kernel-python-sdk#with_streaming_response

"""

return AsyncAPIKeysResourceWithStreamingResponse(self)

async def create(

self,

*,

name: str,

days_to_expire: Optional[int] | Omit = omit,

project_id: Optional[str] | Omit = omit,

# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.

# The extra values given here take precedence over values defined on the client or passed to this method.

extra_headers: Headers | None = None,

extra_query: Query | None = None,

extra_body: Body | None = None,

timeout: float | httpx.Timeout | None | NotGiven = not_given,

) -> CreatedAPIKey:

"""

Create a new API key within the authenticated organization.

Args:

name: Label for the API key (1-255 characters). API keys are not addressable by name.

days_to_expire: Number of days until expiry, up to 3650. Use null for never.

project_id: Unique project identifier

extra_headers: Send extra headers

extra_query: Add additional query parameters to the request

extra_body: Add additional JSON properties to the request

timeout: Override the client-level default timeout for this request, in seconds

"""

return await self._post(

"/org/api_keys",

body=await async_maybe_transform(

{

"name": name,

"days_to_expire": days_to_expire,

"project_id": project_id,

},

api_key_create_params.APIKeyCreateParams,

),

options=make_request_options(

extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout

),

cast_to=CreatedAPIKey,

)

async def retrieve(

self,

id: str,

*,

include_deleted: bool | Omit = omit,

# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.

# The extra values given here take precedence over values defined on the client or passed to this method.

extra_headers: Headers | None = None,

extra_query: Query | None = None,

extra_body: Body | None = None,

timeout: float | httpx.Timeout | None | NotGiven = not_given,

) -> APIKey:

"""Retrieve an API key by ID for the authenticated organization.

API keys are

masked.

Args:

include_deleted: When true, return the API key even if it has been deleted (soft-deleted), for

audit purposes. Defaults to false, which returns 404 for a deleted key.

extra_headers: Send extra headers

extra_query: Add additional query parameters to the request

extra_body: Add additional JSON properties to the request

timeout: Override the client-level default timeout for this request, in seconds

"""

if not id:

raise ValueError(f"Expected a non-empty value for `id` but received {id!r}")

return await self._get(

path_template("/org/api_keys/{id}", id=id),

options=make_request_options(

extra_headers=extra_headers,

extra_query=extra_query,

extra_body=extra_body,

timeout=timeout,

query=await async_maybe_transform(

{"include_deleted": include_deleted}, api_key_retrieve_params.APIKeyRetrieveParams

),

),

cast_to=APIKey,

)

async def update(

self,

id: str,

*,

name: str,

# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.

# The extra values given here take precedence over values defined on the client or passed to this method.

extra_headers: Headers | None = None,

extra_query: Query | None = None,

extra_body: Body | None = None,

timeout: float | httpx.Timeout | None | NotGiven = not_given,

) -> APIKey:

"""

Update an API key's name.

Args:

name: New API key name

extra_headers: Send extra headers

extra_query: Add additional query parameters to the request

extra_body: Add additional JSON properties to the request

timeout: Override the client-level default timeout for this request, in seconds

"""

if not id:

raise ValueError(f"Expected a non-empty value for `id` but received {id!r}")

return await self._patch(

path_template("/org/api_keys/{id}", id=id),

body=await async_maybe_transform({"name": name}, api_key_update_params.APIKeyUpdateParams),

options=make_request_options(

extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout

),

cast_to=APIKey,

)

def list(

self,

*,

include_deleted: bool | Omit = omit,

limit: int | Omit = omit,

name: str | Omit = omit,

offset: int | Omit = omit,

query: str | Omit = omit,

sort_by: Literal["created_at", "name", "expires_at"] | Omit = omit,

sort_direction: Literal["asc", "desc"] | Omit = omit,

status: Literal["active", "deleted", "all"] | Omit = omit,

# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.

# The extra values given here take precedence over values defined on the client or passed to this method.

extra_headers: Headers | None = None,

extra_query: Query | None = None,

extra_body: Body | None = None,

timeout: float | httpx.Timeout | None | NotGiven = not_given,

) -> AsyncPaginator[APIKey, AsyncOffsetPagination[APIKey]]:

"""List API keys for the authenticated organization.

API keys are masked.

Args:

include_deleted: Deprecated: use status=all instead. When true, include deleted (soft-deleted)

API keys in the results for audit purposes.

limit: Maximum number of results to return

name: Exact-match filter on API key name using the database collation. In production,

matching is case- and accent-insensitive. Names are not required to be unique,

so multiple keys may match. When status=all or include_deleted=true is set,

soft-deleted keys with the same name may also match.

offset: Number of results to skip

query: Case-insensitive substring match against API key name, creator, and project. API

key identifiers and masked keys match by exact value or prefix.

sort_by: Field to sort API keys by.

sort_direction: Sort direction for API keys.

status: Filter API keys by status. "active" returns keys that are not deleted (default;

expired-but-not-deleted keys are still included), "deleted" returns only

soft-deleted keys, "all" returns both.

extra_headers: Send extra headers

extra_query: Add additional query parameters to the request

extra_body: Add additional JSON properties to the request

timeout: Override the client-level default timeout for this request, in seconds

"""

return self._get_api_list(

"/org/api_keys",

page=AsyncOffsetPagination[APIKey],

options=make_request_options(

extra_headers=extra_headers,

extra_query=extra_query,

extra_body=extra_body,

timeout=timeout,

query=maybe_transform(

{

"include_deleted": include_deleted,

"limit": limit,

"name": name,

"offset": offset,

"query": query,

"sort_by": sort_by,

"sort_direction": sort_direction,

"status": status,

},

api_key_list_params.APIKeyListParams,

),

),

model=APIKey,

)

async def delete(

self,

id: str,

*,

# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.

# The extra values given here take precedence over values defined on the client or passed to this method.

extra_headers: Headers | None = None,

extra_query: Query | None = None,

extra_body: Body | None = None,

timeout: float | httpx.Timeout | None | NotGiven = not_given,

) -> None:

"""Delete an API key.

A key cannot delete itself; use a different key to delete

this one.

Args:

extra_headers: Send extra headers

extra_query: Add additional query parameters to the request

extra_body: Add additional JSON properties to the request

timeout: Override the client-level default timeout for this request, in seconds

"""

if not id:

raise ValueError(f"Expected a non-empty value for `id` but received {id!r}")

extra_headers = {"Accept": "*/*", **(extra_headers or {})}

return await self._delete(

path_template("/org/api_keys/{id}", id=id),

options=make_request_options(

extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout

),

cast_to=NoneType,

)

async def rotate(

self,

id: str,

*,

days_to_expire: Optional[int] | Omit = omit,

expire_in_days: Optional[int] | Omit = omit,

# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.

# The extra values given here take precedence over values defined on the client or passed to this method.

extra_headers: Headers | None = None,

extra_query: Query | None = None,

extra_body: Body | None = None,

timeout: float | httpx.Timeout | None | NotGiven = not_given,

) -> CreatedAPIKey:

"""Rotate an API key.

Issues a new key that copies the name and project of the

rotated key, and schedules the rotated key to expire after a grace period so

in-flight callers can swap over. The new plaintext key is returned once.

Args:

days_to_expire: Lifetime in days for the new key, up to 3650. Omit to reuse the rotated key's

original lifetime, or never-expires if it had none.

expire_in_days: Grace period in days before the rotated key expires. Use 0 to expire it

immediately. Omit for the default grace period of 7 days.

extra_headers: Send extra headers

extra_query: Add additional query parameters to the request

extra_body: Add additional JSON properties to the request

timeout: Override the client-level default timeout for this request, in seconds

"""

if not id:

raise ValueError(f"Expected a non-empty value for `id` but received {id!r}")

return await self._post(

path_template("/org/api_keys/{id}/rotate", id=id),

body=await async_maybe_transform(

{

"days_to_expire": days_to_expire,

"expire_in_days": expire_in_days,

},

api_key_rotate_params.APIKeyRotateParams,

),

options=make_request_options(

extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout

),

cast_to=CreatedAPIKey,

)

class APIKeysResourceWithRawResponse:

def __init__(self, api_keys: APIKeysResource) -> None:

self._api_keys = api_keys

self.create = to_raw_response_wrapper(

api_keys.create,

)

self.retrieve = to_raw_response_wrapper(

api_keys.retrieve,

)

self.update = to_raw_response_wrapper(

api_keys.update,

)

self.list = to_raw_response_wrapper(

api_keys.list,

)

self.delete = to_raw_response_wrapper(

api_keys.delete,

)

self.rotate = to_raw_response_wrapper(

api_keys.rotate,

)

class AsyncAPIKeysResourceWithRawResponse:

def __init__(self, api_keys: AsyncAPIKeysResource) -> None:

self._api_keys = api_keys

self.create = async_to_raw_response_wrapper(

api_keys.create,

)

self.retrieve = async_to_raw_response_wrapper(

api_keys.retrieve,

)

self.update = async_to_raw_response_wrapper(

api_keys.update,

)

self.list = async_to_raw_response_wrapper(

api_keys.list,

)

self.delete = async_to_raw_response_wrapper(

api_keys.delete,

)

self.rotate = async_to_raw_response_wrapper(

api_keys.rotate,

)

class APIKeysResourceWithStreamingResponse:

def __init__(self, api_keys: APIKeysResource) -> None:

self._api_keys = api_keys

self.create = to_streamed_response_wrapper(

api_keys.create,

)

self.retrieve = to_streamed_response_wrapper(

api_keys.retrieve,

)

self.update = to_streamed_response_wrapper(

api_keys.update,

)

self.list = to_streamed_response_wrapper(

api_keys.list,

)

self.delete = to_streamed_response_wrapper(

api_keys.delete,

)

self.rotate = to_streamed_response_wrapper(

api_keys.rotate,

)

class AsyncAPIKeysResourceWithStreamingResponse:

def __init__(self, api_keys: AsyncAPIKeysResource) -> None:

self._api_keys = api_keys

self.create = async_to_streamed_response_wrapper(

api_keys.create,

)

self.retrieve = async_to_streamed_response_wrapper(

api_keys.retrieve,

)

self.update = async_to_streamed_response_wrapper(

api_keys.update,

)

self.list = async_to_streamed_response_wrapper(

api_keys.list,

)

self.delete = async_to_streamed_response_wrapper(

api_keys.delete,

)

self.rotate = async_to_streamed_response_wrapper(

api_keys.rotate,

)