# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details. from __future__ import annotations from typing import Dict, Optional from typing_extensions import Literal import httpx from .files import ( FilesResource, AsyncFilesResource, FilesResourceWithRawResponse, AsyncFilesResourceWithRawResponse, FilesResourceWithStreamingResponse, AsyncFilesResourceWithStreamingResponse, ) from .agents import ( AgentsResource, AsyncAgentsResource, AgentsResourceWithRawResponse, AsyncAgentsResourceWithRawResponse, AgentsResourceWithStreamingResponse, AsyncAgentsResourceWithStreamingResponse, ) from ...types import folder_list_params, folder_create_params, folder_update_params from ..._types import Body, Omit, Query, Headers, 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 SyncArrayPage, AsyncArrayPage from ..._base_client import AsyncPaginator, make_request_options from ...types.folder import Folder from ...types.embedding_config_param import EmbeddingConfigParam __all__ = ["FoldersResource", "AsyncFoldersResource"] class FoldersResource(SyncAPIResource): @cached_property def files(self) -> FilesResource: return FilesResource(self._client) @cached_property def agents(self) -> AgentsResource: return AgentsResource(self._client) @cached_property def with_raw_response(self) -> FoldersResourceWithRawResponse: """ 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/letta-ai/letta-python#accessing-raw-response-data-eg-headers """ return FoldersResourceWithRawResponse(self) @cached_property def with_streaming_response(self) -> FoldersResourceWithStreamingResponse: """ An alternative to `.with_raw_response` that doesn't eagerly read the response body. For more information, see https://www.github.com/letta-ai/letta-python#with_streaming_response """ return FoldersResourceWithStreamingResponse(self) def create( self, *, name: str, description: Optional[str] | Omit = omit, embedding: Optional[str] | Omit = omit, embedding_chunk_size: Optional[int] | Omit = omit, embedding_config: Optional[EmbeddingConfigParam] | Omit = omit, instructions: Optional[str] | Omit = omit, metadata: Optional[Dict[str, object]] | 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, ) -> Folder: """ Create a new data folder. Args: name: The name of the source. description: The description of the source. embedding: The handle for the embedding config used by the source. embedding_chunk_size: The chunk size of the embedding. embedding_config: Configuration for embedding model connection and processing parameters. instructions: Instructions for how to use the source. metadata: Metadata associated with the source. 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( "/v1/folders/", body=maybe_transform( { "name": name, "description": description, "embedding": embedding, "embedding_chunk_size": embedding_chunk_size, "embedding_config": embedding_config, "instructions": instructions, "metadata": metadata, }, folder_create_params.FolderCreateParams, ), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=Folder, ) def retrieve( self, folder_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, ) -> Folder: """ Get a folder by ID Args: folder_id: The ID of the source in the format 'source-' 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 folder_id: raise ValueError(f"Expected a non-empty value for `folder_id` but received {folder_id!r}") return self._get( path_template("/v1/folders/{folder_id}", folder_id=folder_id), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=Folder, ) def update( self, folder_id: str, *, description: Optional[str] | Omit = omit, embedding_config: Optional[EmbeddingConfigParam] | Omit = omit, instructions: Optional[str] | Omit = omit, metadata: Optional[Dict[str, object]] | Omit = omit, name: 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, ) -> Folder: """ Update the name or documentation of an existing data folder. Args: folder_id: The ID of the source in the format 'source-' description: The description of the source. embedding_config: Configuration for embedding model connection and processing parameters. instructions: Instructions for how to use the source. metadata: Metadata associated with the source. name: The name of the source. 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 folder_id: raise ValueError(f"Expected a non-empty value for `folder_id` but received {folder_id!r}") return self._patch( path_template("/v1/folders/{folder_id}", folder_id=folder_id), body=maybe_transform( { "description": description, "embedding_config": embedding_config, "instructions": instructions, "metadata": metadata, "name": name, }, folder_update_params.FolderUpdateParams, ), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=Folder, ) def list( self, *, after: Optional[str] | Omit = omit, before: Optional[str] | Omit = omit, limit: Optional[int] | Omit = omit, name: Optional[str] | Omit = omit, order: Literal["asc", "desc"] | Omit = omit, order_by: Literal["created_at"] | 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, ) -> SyncArrayPage[Folder]: """ List all data folders created by a user. Args: after: Cursor for pagination (source ID). Returns results relative to this ID in the specified sort order. Expected format: 'source-' before: Cursor for pagination (source ID). Returns results relative to this ID in the specified sort order. Expected format: 'source-' limit: Maximum number of folders to return name: Folder name to filter by order: Sort order for folders by creation time. 'asc' for oldest first, 'desc' for newest first order_by: Field to sort by 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( "/v1/folders/", page=SyncArrayPage[Folder], options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout, query=maybe_transform( { "after": after, "before": before, "limit": limit, "name": name, "order": order, "order_by": order_by, }, folder_list_params.FolderListParams, ), ), model=Folder, ) def delete( self, folder_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, ) -> object: """ Delete a data folder. Args: folder_id: The ID of the source in the format 'source-' 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 folder_id: raise ValueError(f"Expected a non-empty value for `folder_id` but received {folder_id!r}") return self._delete( path_template("/v1/folders/{folder_id}", folder_id=folder_id), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=object, ) class AsyncFoldersResource(AsyncAPIResource): @cached_property def files(self) -> AsyncFilesResource: return AsyncFilesResource(self._client) @cached_property def agents(self) -> AsyncAgentsResource: return AsyncAgentsResource(self._client) @cached_property def with_raw_response(self) -> AsyncFoldersResourceWithRawResponse: """ 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/letta-ai/letta-python#accessing-raw-response-data-eg-headers """ return AsyncFoldersResourceWithRawResponse(self) @cached_property def with_streaming_response(self) -> AsyncFoldersResourceWithStreamingResponse: """ An alternative to `.with_raw_response` that doesn't eagerly read the response body. For more information, see https://www.github.com/letta-ai/letta-python#with_streaming_response """ return AsyncFoldersResourceWithStreamingResponse(self) async def create( self, *, name: str, description: Optional[str] | Omit = omit, embedding: Optional[str] | Omit = omit, embedding_chunk_size: Optional[int] | Omit = omit, embedding_config: Optional[EmbeddingConfigParam] | Omit = omit, instructions: Optional[str] | Omit = omit, metadata: Optional[Dict[str, object]] | 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, ) -> Folder: """ Create a new data folder. Args: name: The name of the source. description: The description of the source. embedding: The handle for the embedding config used by the source. embedding_chunk_size: The chunk size of the embedding. embedding_config: Configuration for embedding model connection and processing parameters. instructions: Instructions for how to use the source. metadata: Metadata associated with the source. 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( "/v1/folders/", body=await async_maybe_transform( { "name": name, "description": description, "embedding": embedding, "embedding_chunk_size": embedding_chunk_size, "embedding_config": embedding_config, "instructions": instructions, "metadata": metadata, }, folder_create_params.FolderCreateParams, ), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=Folder, ) async def retrieve( self, folder_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, ) -> Folder: """ Get a folder by ID Args: folder_id: The ID of the source in the format 'source-' 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 folder_id: raise ValueError(f"Expected a non-empty value for `folder_id` but received {folder_id!r}") return await self._get( path_template("/v1/folders/{folder_id}", folder_id=folder_id), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=Folder, ) async def update( self, folder_id: str, *, description: Optional[str] | Omit = omit, embedding_config: Optional[EmbeddingConfigParam] | Omit = omit, instructions: Optional[str] | Omit = omit, metadata: Optional[Dict[str, object]] | Omit = omit, name: 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, ) -> Folder: """ Update the name or documentation of an existing data folder. Args: folder_id: The ID of the source in the format 'source-' description: The description of the source. embedding_config: Configuration for embedding model connection and processing parameters. instructions: Instructions for how to use the source. metadata: Metadata associated with the source. name: The name of the source. 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 folder_id: raise ValueError(f"Expected a non-empty value for `folder_id` but received {folder_id!r}") return await self._patch( path_template("/v1/folders/{folder_id}", folder_id=folder_id), body=await async_maybe_transform( { "description": description, "embedding_config": embedding_config, "instructions": instructions, "metadata": metadata, "name": name, }, folder_update_params.FolderUpdateParams, ), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=Folder, ) def list( self, *, after: Optional[str] | Omit = omit, before: Optional[str] | Omit = omit, limit: Optional[int] | Omit = omit, name: Optional[str] | Omit = omit, order: Literal["asc", "desc"] | Omit = omit, order_by: Literal["created_at"] | 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[Folder, AsyncArrayPage[Folder]]: """ List all data folders created by a user. Args: after: Cursor for pagination (source ID). Returns results relative to this ID in the specified sort order. Expected format: 'source-' before: Cursor for pagination (source ID). Returns results relative to this ID in the specified sort order. Expected format: 'source-' limit: Maximum number of folders to return name: Folder name to filter by order: Sort order for folders by creation time. 'asc' for oldest first, 'desc' for newest first order_by: Field to sort by 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( "/v1/folders/", page=AsyncArrayPage[Folder], options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout, query=maybe_transform( { "after": after, "before": before, "limit": limit, "name": name, "order": order, "order_by": order_by, }, folder_list_params.FolderListParams, ), ), model=Folder, ) async def delete( self, folder_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, ) -> object: """ Delete a data folder. Args: folder_id: The ID of the source in the format 'source-' 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 folder_id: raise ValueError(f"Expected a non-empty value for `folder_id` but received {folder_id!r}") return await self._delete( path_template("/v1/folders/{folder_id}", folder_id=folder_id), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=object, ) class FoldersResourceWithRawResponse: def __init__(self, folders: FoldersResource) -> None: self._folders = folders self.create = to_raw_response_wrapper( folders.create, ) self.retrieve = to_raw_response_wrapper( folders.retrieve, ) self.update = to_raw_response_wrapper( folders.update, ) self.list = to_raw_response_wrapper( folders.list, ) self.delete = to_raw_response_wrapper( folders.delete, ) @cached_property def files(self) -> FilesResourceWithRawResponse: return FilesResourceWithRawResponse(self._folders.files) @cached_property def agents(self) -> AgentsResourceWithRawResponse: return AgentsResourceWithRawResponse(self._folders.agents) class AsyncFoldersResourceWithRawResponse: def __init__(self, folders: AsyncFoldersResource) -> None: self._folders = folders self.create = async_to_raw_response_wrapper( folders.create, ) self.retrieve = async_to_raw_response_wrapper( folders.retrieve, ) self.update = async_to_raw_response_wrapper( folders.update, ) self.list = async_to_raw_response_wrapper( folders.list, ) self.delete = async_to_raw_response_wrapper( folders.delete, ) @cached_property def files(self) -> AsyncFilesResourceWithRawResponse: return AsyncFilesResourceWithRawResponse(self._folders.files) @cached_property def agents(self) -> AsyncAgentsResourceWithRawResponse: return AsyncAgentsResourceWithRawResponse(self._folders.agents) class FoldersResourceWithStreamingResponse: def __init__(self, folders: FoldersResource) -> None: self._folders = folders self.create = to_streamed_response_wrapper( folders.create, ) self.retrieve = to_streamed_response_wrapper( folders.retrieve, ) self.update = to_streamed_response_wrapper( folders.update, ) self.list = to_streamed_response_wrapper( folders.list, ) self.delete = to_streamed_response_wrapper( folders.delete, ) @cached_property def files(self) -> FilesResourceWithStreamingResponse: return FilesResourceWithStreamingResponse(self._folders.files) @cached_property def agents(self) -> AgentsResourceWithStreamingResponse: return AgentsResourceWithStreamingResponse(self._folders.agents) class AsyncFoldersResourceWithStreamingResponse: def __init__(self, folders: AsyncFoldersResource) -> None: self._folders = folders self.create = async_to_streamed_response_wrapper( folders.create, ) self.retrieve = async_to_streamed_response_wrapper( folders.retrieve, ) self.update = async_to_streamed_response_wrapper( folders.update, ) self.list = async_to_streamed_response_wrapper( folders.list, ) self.delete = async_to_streamed_response_wrapper( folders.delete, ) @cached_property def files(self) -> AsyncFilesResourceWithStreamingResponse: return AsyncFilesResourceWithStreamingResponse(self._folders.files) @cached_property def agents(self) -> AsyncAgentsResourceWithStreamingResponse: return AsyncAgentsResourceWithStreamingResponse(self._folders.agents)