# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details. from __future__ import annotations from typing import Dict from ..._httpx import httpx from ..._types import Body, Omit, Query, Headers, NoneType, NotGiven, SequenceNotStr, 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 ..._base_client import make_request_options from ...types.sandboxes import snapshot_list_params, snapshot_create_params from ...types.snapshot_response import SnapshotResponse from ...types.snapshot_list_response import SnapshotListResponse from ...types.sandboxes.snapshot_retrieve_by_name_response import SnapshotRetrieveByNameResponse __all__ = ["SnapshotsResource", "AsyncSnapshotsResource"] class SnapshotsResource(SyncAPIResource): @cached_property def with_raw_response(self) -> SnapshotsResourceWithRawResponse: """ This property can be used as a prefix for any HTTP method call to return the raw response object instead of the parsed content. """ return SnapshotsResourceWithRawResponse(self) @cached_property def with_streaming_response(self) -> SnapshotsResourceWithStreamingResponse: """ An alternative to `.with_raw_response` that doesn't eagerly read the response body. """ return SnapshotsResourceWithStreamingResponse(self) def create( self, *, docker_image: str, fs_capacity_bytes: int, name: str, labels: Dict[str, str] | Omit = omit, registry_id: str | Omit = omit, tag: 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, ) -> SnapshotResponse: """ Create a snapshot from a Docker image (async build). Args: labels: Labels seed the snapshot's labels, overriding any label of the same key derived from the Docker image. tag: mutable Docker-style tag; defaults to "latest" 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( "/api/v2/sandboxes/snapshots", body=maybe_transform( { "docker_image": docker_image, "fs_capacity_bytes": fs_capacity_bytes, "name": name, "labels": labels, "registry_id": registry_id, "tag": tag, }, snapshot_create_params.SnapshotCreateParams, ), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=SnapshotResponse, ) def retrieve( self, snapshot_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, ) -> SnapshotResponse: """Get a sandbox snapshot by ID or by a Docker-style reference. A bare name means name:latest, falling back to the newest ready untagged snapshot of that name. To list the tags under a name, use /api/v2/sandboxes/snapshots-by-name/{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 snapshot_id: raise ValueError(f"Expected a non-empty value for `snapshot_id` but received {snapshot_id!r}") return self._get( path_template("/api/v2/sandboxes/snapshots/{snapshot_id}", snapshot_id=snapshot_id), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=SnapshotResponse, ) def list( self, *, created_by: str | Omit = omit, label: SequenceNotStr[str] | Omit = omit, limit: int | Omit = omit, name_contains: str | Omit = omit, offset: int | Omit = omit, sort_by: str | Omit = omit, sort_direction: str | Omit = omit, status: 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, ) -> SnapshotListResponse: """ List sandbox snapshots for the authenticated tenant, with optional filtering, sorting, and pagination. Args: created_by: Filter by creator identity. Only 'me' is supported. label: Filter by label. Repeatable; all must match. Use 'key' to match on key presence or 'key=value' for equality. limit: Maximum number of results name_contains: Filter by name substring offset: Pagination offset sort_by: Sort column (name, status, created_at) sort_direction: Sort direction (asc, desc) status: Filter by status (building, ready, failed, deleting) 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/v2/sandboxes/snapshots", options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout, query=maybe_transform( { "created_by": created_by, "label": label, "limit": limit, "name_contains": name_contains, "offset": offset, "sort_by": sort_by, "sort_direction": sort_direction, "status": status, }, snapshot_list_params.SnapshotListParams, ), ), cast_to=SnapshotListResponse, ) def delete( self, snapshot_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 a snapshot by ID or by a Docker-style name[:tag] reference. The underlying storage is reclaimed asynchronously. 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 snapshot_id: raise ValueError(f"Expected a non-empty value for `snapshot_id` but received {snapshot_id!r}") extra_headers = {"Accept": "*/*", **(extra_headers or {})} return self._delete( path_template("/api/v2/sandboxes/snapshots/{snapshot_id}", snapshot_id=snapshot_id), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=NoneType, ) def retrieve_by_name( self, 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, ) -> SnapshotRetrieveByNameResponse: """ Get a snapshot name and every tag under it, with the snapshot each tag resolves to. To fetch one snapshot, use /api/v2/sandboxes/snapshots/{snapshot_id}. 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 name: raise ValueError(f"Expected a non-empty value for `name` but received {name!r}") return self._get( path_template("/api/v2/sandboxes/snapshots-by-name/{name}", name=name), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=SnapshotRetrieveByNameResponse, ) class AsyncSnapshotsResource(AsyncAPIResource): @cached_property def with_raw_response(self) -> AsyncSnapshotsResourceWithRawResponse: """ This property can be used as a prefix for any HTTP method call to return the raw response object instead of the parsed content. """ return AsyncSnapshotsResourceWithRawResponse(self) @cached_property def with_streaming_response(self) -> AsyncSnapshotsResourceWithStreamingResponse: """ An alternative to `.with_raw_response` that doesn't eagerly read the response body. """ return AsyncSnapshotsResourceWithStreamingResponse(self) async def create( self, *, docker_image: str, fs_capacity_bytes: int, name: str, labels: Dict[str, str] | Omit = omit, registry_id: str | Omit = omit, tag: 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, ) -> SnapshotResponse: """ Create a snapshot from a Docker image (async build). Args: labels: Labels seed the snapshot's labels, overriding any label of the same key derived from the Docker image. tag: mutable Docker-style tag; defaults to "latest" 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( "/api/v2/sandboxes/snapshots", body=await async_maybe_transform( { "docker_image": docker_image, "fs_capacity_bytes": fs_capacity_bytes, "name": name, "labels": labels, "registry_id": registry_id, "tag": tag, }, snapshot_create_params.SnapshotCreateParams, ), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=SnapshotResponse, ) async def retrieve( self, snapshot_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, ) -> SnapshotResponse: """Get a sandbox snapshot by ID or by a Docker-style reference. A bare name means name:latest, falling back to the newest ready untagged snapshot of that name. To list the tags under a name, use /api/v2/sandboxes/snapshots-by-name/{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 snapshot_id: raise ValueError(f"Expected a non-empty value for `snapshot_id` but received {snapshot_id!r}") return await self._get( path_template("/api/v2/sandboxes/snapshots/{snapshot_id}", snapshot_id=snapshot_id), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=SnapshotResponse, ) async def list( self, *, created_by: str | Omit = omit, label: SequenceNotStr[str] | Omit = omit, limit: int | Omit = omit, name_contains: str | Omit = omit, offset: int | Omit = omit, sort_by: str | Omit = omit, sort_direction: str | Omit = omit, status: 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, ) -> SnapshotListResponse: """ List sandbox snapshots for the authenticated tenant, with optional filtering, sorting, and pagination. Args: created_by: Filter by creator identity. Only 'me' is supported. label: Filter by label. Repeatable; all must match. Use 'key' to match on key presence or 'key=value' for equality. limit: Maximum number of results name_contains: Filter by name substring offset: Pagination offset sort_by: Sort column (name, status, created_at) sort_direction: Sort direction (asc, desc) status: Filter by status (building, ready, failed, deleting) 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._get( "/api/v2/sandboxes/snapshots", options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout, query=await async_maybe_transform( { "created_by": created_by, "label": label, "limit": limit, "name_contains": name_contains, "offset": offset, "sort_by": sort_by, "sort_direction": sort_direction, "status": status, }, snapshot_list_params.SnapshotListParams, ), ), cast_to=SnapshotListResponse, ) async def delete( self, snapshot_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 a snapshot by ID or by a Docker-style name[:tag] reference. The underlying storage is reclaimed asynchronously. 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 snapshot_id: raise ValueError(f"Expected a non-empty value for `snapshot_id` but received {snapshot_id!r}") extra_headers = {"Accept": "*/*", **(extra_headers or {})} return await self._delete( path_template("/api/v2/sandboxes/snapshots/{snapshot_id}", snapshot_id=snapshot_id), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=NoneType, ) async def retrieve_by_name( self, 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, ) -> SnapshotRetrieveByNameResponse: """ Get a snapshot name and every tag under it, with the snapshot each tag resolves to. To fetch one snapshot, use /api/v2/sandboxes/snapshots/{snapshot_id}. 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 name: raise ValueError(f"Expected a non-empty value for `name` but received {name!r}") return await self._get( path_template("/api/v2/sandboxes/snapshots-by-name/{name}", name=name), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=SnapshotRetrieveByNameResponse, ) class SnapshotsResourceWithRawResponse: def __init__(self, snapshots: SnapshotsResource) -> None: self._snapshots = snapshots self.create = to_raw_response_wrapper( snapshots.create, ) self.retrieve = to_raw_response_wrapper( snapshots.retrieve, ) self.list = to_raw_response_wrapper( snapshots.list, ) self.delete = to_raw_response_wrapper( snapshots.delete, ) self.retrieve_by_name = to_raw_response_wrapper( snapshots.retrieve_by_name, ) class AsyncSnapshotsResourceWithRawResponse: def __init__(self, snapshots: AsyncSnapshotsResource) -> None: self._snapshots = snapshots self.create = async_to_raw_response_wrapper( snapshots.create, ) self.retrieve = async_to_raw_response_wrapper( snapshots.retrieve, ) self.list = async_to_raw_response_wrapper( snapshots.list, ) self.delete = async_to_raw_response_wrapper( snapshots.delete, ) self.retrieve_by_name = async_to_raw_response_wrapper( snapshots.retrieve_by_name, ) class SnapshotsResourceWithStreamingResponse: def __init__(self, snapshots: SnapshotsResource) -> None: self._snapshots = snapshots self.create = to_streamed_response_wrapper( snapshots.create, ) self.retrieve = to_streamed_response_wrapper( snapshots.retrieve, ) self.list = to_streamed_response_wrapper( snapshots.list, ) self.delete = to_streamed_response_wrapper( snapshots.delete, ) self.retrieve_by_name = to_streamed_response_wrapper( snapshots.retrieve_by_name, ) class AsyncSnapshotsResourceWithStreamingResponse: def __init__(self, snapshots: AsyncSnapshotsResource) -> None: self._snapshots = snapshots self.create = async_to_streamed_response_wrapper( snapshots.create, ) self.retrieve = async_to_streamed_response_wrapper( snapshots.retrieve, ) self.list = async_to_streamed_response_wrapper( snapshots.list, ) self.delete = async_to_streamed_response_wrapper( snapshots.delete, ) self.retrieve_by_name = async_to_streamed_response_wrapper( snapshots.retrieve_by_name, )