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

from __future__ import annotations

from typing import Mapping, cast

from typing_extensions import Literal

import httpx

from ..types import extension_list_params, extension_upload_params, extension_download_from_chrome_store_params

from .._files import deepcopy_with_paths

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

from .._utils import extract_files, path_template, maybe_transform, async_maybe_transform

from .._compat import cached_property

from .._resource import SyncAPIResource, AsyncAPIResource

from .._response import (

BinaryAPIResponse,

AsyncBinaryAPIResponse,

StreamedBinaryAPIResponse,

AsyncStreamedBinaryAPIResponse,

to_raw_response_wrapper,

to_streamed_response_wrapper,

async_to_raw_response_wrapper,

to_custom_raw_response_wrapper,

async_to_streamed_response_wrapper,

to_custom_streamed_response_wrapper,

async_to_custom_raw_response_wrapper,

async_to_custom_streamed_response_wrapper,

)

from ..pagination import SyncOffsetPagination, AsyncOffsetPagination

from .._base_client import AsyncPaginator, make_request_options

from ..types.extension_get_response import ExtensionGetResponse

from ..types.extension_list_response import ExtensionListResponse

from ..types.extension_upload_response import ExtensionUploadResponse

__all__ = ["ExtensionsResource", "AsyncExtensionsResource"]

class ExtensionsResource(SyncAPIResource):

"""Create, list, retrieve, and delete browser extensions."""

@cached_property

def with_raw_response(self) -> ExtensionsResourceWithRawResponse:

"""

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

@cached_property

def with_streaming_response(self) -> ExtensionsResourceWithStreamingResponse:

"""

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

def list(

self,

*,

limit: int | Omit = omit,

name: str | Omit = omit,

offset: int | Omit = omit,

query: 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,

) -> SyncOffsetPagination[ExtensionListResponse]:

"""

List extensions in the resolved project.

Args:

limit: Limit the number of extensions to return.

name: Exact-match filter on extension name using the database collation. In

production, matching is case- and accent-insensitive. During the default-project

migration, unscoped requests prefer a concrete default-project extension over a

legacy unscoped extension with the same name.

offset: Offset the number of extensions to return.

query: Case-insensitive substring match against extension name. IDs match by exact

value.

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(

"/extensions",

page=SyncOffsetPagination[ExtensionListResponse],

options=make_request_options(

extra_headers=extra_headers,

extra_query=extra_query,

extra_body=extra_body,

timeout=timeout,

query=maybe_transform(

{

"limit": limit,

"name": name,

"offset": offset,

"query": query,

},

extension_list_params.ExtensionListParams,

),

),

model=ExtensionListResponse,

)

def delete(

self,

id_or_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,

) -> None:

"""

Delete an extension by its ID or by its name.

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_or_name:

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

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

return self._delete(

path_template("/extensions/{id_or_name}", id_or_name=id_or_name),

options=make_request_options(

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

),

cast_to=NoneType,

)

def download(

self,

id_or_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,

) -> BinaryAPIResponse:

"""

Download the extension as a ZIP archive by ID or name.

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_or_name:

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

extra_headers = {"Accept": "application/octet-stream", **(extra_headers or {})}

return self._get(

path_template("/extensions/{id_or_name}", id_or_name=id_or_name),

options=make_request_options(

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

),

cast_to=BinaryAPIResponse,

)

def download_from_chrome_store(

self,

*,

url: str,

os: Literal["win", "mac", "linux"] | 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,

) -> BinaryAPIResponse:

"""

Returns a ZIP archive containing the unpacked extension fetched from the Chrome

Web Store.

Args:

url: Chrome Web Store URL for the extension.

os: Target operating system for the extension package. Defaults to linux.

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

"""

extra_headers = {"Accept": "application/octet-stream", **(extra_headers or {})}

return self._get(

"/extensions/from_chrome_store",

options=make_request_options(

extra_headers=extra_headers,

extra_query=extra_query,

extra_body=extra_body,

timeout=timeout,

query=maybe_transform(

{

"url": url,

"os": os,

},

extension_download_from_chrome_store_params.ExtensionDownloadFromChromeStoreParams,

),

),

cast_to=BinaryAPIResponse,

)

def get(

self,

id_or_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,

) -> ExtensionGetResponse:

"""

Get an extension's metadata (name, size, timestamps) by ID or name, without

downloading the archive.

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_or_name:

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

return self._get(

path_template("/extensions/{id_or_name}/metadata", id_or_name=id_or_name),

options=make_request_options(

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

),

cast_to=ExtensionGetResponse,

)

def upload(

self,

*,

file: FileTypes,

name: 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,

) -> ExtensionUploadResponse:

"""Upload a zip file containing an unpacked browser extension.

Optionally provide a

unique name for later reference.

Args:

file: ZIP file containing the browser extension.

name: Optional unique name within the project to reference this extension.

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

"""

body = deepcopy_with_paths(

{

"file": file,

"name": name,

},

[["file"]],

)

files = extract_files(cast(Mapping[str, object], body), paths=[["file"]])

# It should be noted that the actual Content-Type header that will be

# sent to the server will contain a `boundary` parameter, e.g.

# multipart/form-data; boundary=---abc--

extra_headers = {"Content-Type": "multipart/form-data", **(extra_headers or {})}

return self._post(

"/extensions",

body=maybe_transform(body, extension_upload_params.ExtensionUploadParams),

files=files,

options=make_request_options(

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

),

cast_to=ExtensionUploadResponse,

)

class AsyncExtensionsResource(AsyncAPIResource):

"""Create, list, retrieve, and delete browser extensions."""

@cached_property

def with_raw_response(self) -> AsyncExtensionsResourceWithRawResponse:

"""

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

@cached_property

def with_streaming_response(self) -> AsyncExtensionsResourceWithStreamingResponse:

"""

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

def list(

self,

*,

limit: int | Omit = omit,

name: str | Omit = omit,

offset: int | Omit = omit,

query: 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,

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

"""

List extensions in the resolved project.

Args:

limit: Limit the number of extensions to return.

name: Exact-match filter on extension name using the database collation. In

production, matching is case- and accent-insensitive. During the default-project

migration, unscoped requests prefer a concrete default-project extension over a

legacy unscoped extension with the same name.

offset: Offset the number of extensions to return.

query: Case-insensitive substring match against extension name. IDs match by exact

value.

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(

"/extensions",

page=AsyncOffsetPagination[ExtensionListResponse],

options=make_request_options(

extra_headers=extra_headers,

extra_query=extra_query,

extra_body=extra_body,

timeout=timeout,

query=maybe_transform(

{

"limit": limit,

"name": name,

"offset": offset,

"query": query,

},

extension_list_params.ExtensionListParams,

),

),

model=ExtensionListResponse,

)

async def delete(

self,

id_or_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,

) -> None:

"""

Delete an extension by its ID or by its name.

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_or_name:

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

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

return await self._delete(

path_template("/extensions/{id_or_name}", id_or_name=id_or_name),

options=make_request_options(

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

),

cast_to=NoneType,

)

async def download(

self,

id_or_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,

) -> AsyncBinaryAPIResponse:

"""

Download the extension as a ZIP archive by ID or name.

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_or_name:

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

extra_headers = {"Accept": "application/octet-stream", **(extra_headers or {})}

return await self._get(

path_template("/extensions/{id_or_name}", id_or_name=id_or_name),

options=make_request_options(

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

),

cast_to=AsyncBinaryAPIResponse,

)

async def download_from_chrome_store(

self,

*,

url: str,

os: Literal["win", "mac", "linux"] | 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,

) -> AsyncBinaryAPIResponse:

"""

Returns a ZIP archive containing the unpacked extension fetched from the Chrome

Web Store.

Args:

url: Chrome Web Store URL for the extension.

os: Target operating system for the extension package. Defaults to linux.

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

"""

extra_headers = {"Accept": "application/octet-stream", **(extra_headers or {})}

return await self._get(

"/extensions/from_chrome_store",

options=make_request_options(

extra_headers=extra_headers,

extra_query=extra_query,

extra_body=extra_body,

timeout=timeout,

query=await async_maybe_transform(

{

"url": url,

"os": os,

},

extension_download_from_chrome_store_params.ExtensionDownloadFromChromeStoreParams,

),

),

cast_to=AsyncBinaryAPIResponse,

)

async def get(

self,

id_or_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,

) -> ExtensionGetResponse:

"""

Get an extension's metadata (name, size, timestamps) by ID or name, without

downloading the archive.

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_or_name:

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

return await self._get(

path_template("/extensions/{id_or_name}/metadata", id_or_name=id_or_name),

options=make_request_options(

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

),

cast_to=ExtensionGetResponse,

)

async def upload(

self,

*,

file: FileTypes,

name: 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,

) -> ExtensionUploadResponse:

"""Upload a zip file containing an unpacked browser extension.

Optionally provide a

unique name for later reference.

Args:

file: ZIP file containing the browser extension.

name: Optional unique name within the project to reference this extension.

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

"""

body = deepcopy_with_paths(

{

"file": file,

"name": name,

},

[["file"]],

)

files = extract_files(cast(Mapping[str, object], body), paths=[["file"]])

# It should be noted that the actual Content-Type header that will be

# sent to the server will contain a `boundary` parameter, e.g.

# multipart/form-data; boundary=---abc--

extra_headers = {"Content-Type": "multipart/form-data", **(extra_headers or {})}

return await self._post(

"/extensions",

body=await async_maybe_transform(body, extension_upload_params.ExtensionUploadParams),

files=files,

options=make_request_options(

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

),

cast_to=ExtensionUploadResponse,

)

class ExtensionsResourceWithRawResponse:

def __init__(self, extensions: ExtensionsResource) -> None:

self._extensions = extensions

self.list = to_raw_response_wrapper(

extensions.list,

)

self.delete = to_raw_response_wrapper(

extensions.delete,

)

self.download = to_custom_raw_response_wrapper(

extensions.download,

BinaryAPIResponse,

)

self.download_from_chrome_store = to_custom_raw_response_wrapper(

extensions.download_from_chrome_store,

BinaryAPIResponse,

)

self.get = to_raw_response_wrapper(

extensions.get,

)

self.upload = to_raw_response_wrapper(

extensions.upload,

)

class AsyncExtensionsResourceWithRawResponse:

def __init__(self, extensions: AsyncExtensionsResource) -> None:

self._extensions = extensions

self.list = async_to_raw_response_wrapper(

extensions.list,

)

self.delete = async_to_raw_response_wrapper(

extensions.delete,

)

self.download = async_to_custom_raw_response_wrapper(

extensions.download,

AsyncBinaryAPIResponse,

)

self.download_from_chrome_store = async_to_custom_raw_response_wrapper(

extensions.download_from_chrome_store,

AsyncBinaryAPIResponse,

)

self.get = async_to_raw_response_wrapper(

extensions.get,

)

self.upload = async_to_raw_response_wrapper(

extensions.upload,

)

class ExtensionsResourceWithStreamingResponse:

def __init__(self, extensions: ExtensionsResource) -> None:

self._extensions = extensions

self.list = to_streamed_response_wrapper(

extensions.list,

)

self.delete = to_streamed_response_wrapper(

extensions.delete,

)

self.download = to_custom_streamed_response_wrapper(

extensions.download,

StreamedBinaryAPIResponse,

)

self.download_from_chrome_store = to_custom_streamed_response_wrapper(

extensions.download_from_chrome_store,

StreamedBinaryAPIResponse,

)

self.get = to_streamed_response_wrapper(

extensions.get,

)

self.upload = to_streamed_response_wrapper(

extensions.upload,

)

class AsyncExtensionsResourceWithStreamingResponse:

def __init__(self, extensions: AsyncExtensionsResource) -> None:

self._extensions = extensions

self.list = async_to_streamed_response_wrapper(

extensions.list,

)

self.delete = async_to_streamed_response_wrapper(

extensions.delete,

)

self.download = async_to_custom_streamed_response_wrapper(

extensions.download,

AsyncStreamedBinaryAPIResponse,

)

self.download_from_chrome_store = async_to_custom_streamed_response_wrapper(

extensions.download_from_chrome_store,

AsyncStreamedBinaryAPIResponse,

)

self.get = async_to_streamed_response_wrapper(

extensions.get,

)

self.upload = async_to_streamed_response_wrapper(

extensions.upload,

)