two version of R2R are here HEAD master

1 files changed, 120 insertions, 0 deletions

diff --git a/.venv/lib/python3.12/site-packages/storage3/_sync/bucket.py b/.venv/lib/python3.12/site-packages/storage3/_sync/bucket.py
new file mode 100644
index 00000000..8700bf06
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/storage3/_sync/bucket.py
@@ -0,0 +1,120 @@
+from __future__ import annotations
+
+from typing import Any, Optional
+
+from httpx import HTTPStatusError, Response
+
+from ..exceptions import StorageApiError
+from ..types import CreateOrUpdateBucketOptions, RequestMethod
+from ..utils import SyncClient
+from .file_api import SyncBucket
+
+__all__ = ["SyncStorageBucketAPI"]
+
+
+class SyncStorageBucketAPI:
+    """This class abstracts access to the endpoint to the Get, List, Empty, and Delete operations on a bucket"""
+
+    def __init__(self, session: SyncClient) -> None:
+        self._client = session
+
+    def _request(
+        self,
+        method: RequestMethod,
+        url: str,
+        json: Optional[dict[Any, Any]] = None,
+    ) -> Response:
+        try:
+            response = self._client.request(method, url, json=json)
+            response.raise_for_status()
+        except HTTPStatusError as exc:
+            resp = exc.response.json()
+            raise StorageApiError(resp["message"], resp["error"], resp["statusCode"])
+
+        return response
+
+    def list_buckets(self) -> list[SyncBucket]:
+        """Retrieves the details of all storage buckets within an existing product."""
+        # if the request doesn't error, it is assured to return a list
+        res = self._request("GET", "/bucket")
+        return [SyncBucket(**bucket) for bucket in res.json()]
+
+    def get_bucket(self, id: str) -> SyncBucket:
+        """Retrieves the details of an existing storage bucket.
+
+        Parameters
+        ----------
+        id
+            The unique identifier of the bucket you would like to retrieve.
+        """
+        res = self._request("GET", f"/bucket/{id}")
+        json = res.json()
+        return SyncBucket(**json)
+
+    def create_bucket(
+        self,
+        id: str,
+        name: Optional[str] = None,
+        options: Optional[CreateOrUpdateBucketOptions] = None,
+    ) -> dict[str, str]:
+        """Creates a new storage bucket.
+
+        Parameters
+        ----------
+        id
+            A unique identifier for the bucket you are creating.
+        name
+            A name for the bucket you are creating. If not passed, the id is used as the name as well.
+        options
+            Extra options to send while creating the bucket. Valid options are `public`, `file_size_limit` and
+            `allowed_mime_types`.
+        """
+        json: dict[str, Any] = {"id": id, "name": name or id}
+        if options:
+            json.update(**options)
+        res = self._request(
+            "POST",
+            "/bucket",
+            json=json,
+        )
+        return res.json()
+
+    def update_bucket(
+        self, id: str, options: CreateOrUpdateBucketOptions
+    ) -> dict[str, str]:
+        """Update a storage bucket.
+
+        Parameters
+        ----------
+        id
+            The unique identifier of the bucket you would like to update.
+        options
+            The properties you want to update. Valid options are `public`, `file_size_limit` and
+            `allowed_mime_types`.
+        """
+        json = {"id": id, "name": id, **options}
+        res = self._request("PUT", f"/bucket/{id}", json=json)
+        return res.json()
+
+    def empty_bucket(self, id: str) -> dict[str, str]:
+        """Removes all objects inside a single bucket.
+
+        Parameters
+        ----------
+        id
+            The unique identifier of the bucket you would like to empty.
+        """
+        res = self._request("POST", f"/bucket/{id}/empty", json={})
+        return res.json()
+
+    def delete_bucket(self, id: str) -> dict[str, str]:
+        """Deletes an existing bucket. Note that you cannot delete buckets with existing objects inside. You must first
+        `empty()` the bucket.
+
+        Parameters
+        ----------
+        id
+            The unique identifier of the bucket you would like to delete.
+        """
+        res = self._request("DELETE", f"/bucket/{id}", json={})
+        return res.json()