feat: bucket level file controls and an update_bucket method (#103)

* feat: bucket level file controls and an update_bucket method

* fix: import __future__ annotations

python 3.8 doesn't allow subscripting types like list

Author: anand2312

storage3/_async/bucket.py

@@ -4,7 +4,7 @@
 
 from httpx import HTTPError, Response
 
-from ..types import RequestMethod
+from ..types import CreateOrUpdateBucketOptions, RequestMethod
 from ..utils import AsyncClient, StorageException
 from .file_api import AsyncBucket
 
@@ -52,7 +52,10 @@ async def get_bucket(self, id: str) -> AsyncBucket:
         return AsyncBucket(**json, _client=self._client)
 
     async def create_bucket(
-        self, id: str, name: Optional[str] = None, public: bool = False
+        self,
+        id: str,
+        name: Optional[str] = None,
+        options: Optional[CreateOrUpdateBucketOptions] = None,
     ) -> dict[str, str]:
         """Creates a new storage bucket.
 
@@ -62,16 +65,37 @@ async def create_bucket(
             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.
-        public
-            Whether the bucket you are creating should be publicly accessible. Defaults to False.
+        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 = await self._request(
             "POST",
             "/bucket",
-            json={"id": id, "name": name or id, "public": public},
+            json=json,
         )
         return res.json()
 
+    async 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 = await self._request("PUT", f"/bucket/{id}", json=json)
+        return res.json()
+
     async def empty_bucket(self, id: str) -> dict[str, str]:
         """Removes all objects inside a single bucket.
 

storage3/_sync/bucket.py

@@ -4,7 +4,7 @@
 
 from httpx import HTTPError, Response
 
-from ..types import RequestMethod
+from ..types import CreateOrUpdateBucketOptions, RequestMethod
 from ..utils import StorageException, SyncClient
 from .file_api import SyncBucket
 
@@ -52,7 +52,10 @@ def get_bucket(self, id: str) -> SyncBucket:
         return SyncBucket(**json, _client=self._client)
 
     def create_bucket(
-        self, id: str, name: Optional[str] = None, public: bool = False
+        self,
+        id: str,
+        name: Optional[str] = None,
+        options: Optional[CreateOrUpdateBucketOptions] = None,
     ) -> dict[str, str]:
         """Creates a new storage bucket.
 
@@ -62,16 +65,37 @@ def create_bucket(
             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.
-        public
-            Whether the bucket you are creating should be publicly accessible. Defaults to False.
+        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={"id": id, "name": name or id, "public": public},
+            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.
 

storage3/types.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from dataclasses import dataclass
 from datetime import datetime
 from typing import Optional, Union
@@ -19,7 +21,7 @@ class BaseBucket:
     created_at: datetime
     updated_at: datetime
     file_size_limit: Optional[int]
-    allowed_mime_types: Optional[int]
+    allowed_mime_types: Optional[list[str]]
 
     def __post_init__(self) -> None:
         # created_at and updated_at are returned by the API as ISO timestamps
@@ -34,6 +36,12 @@ class _sortByType(TypedDict):
     order: Literal["asc", "desc"]
 
 
+class CreateOrUpdateBucketOptions(TypedDict, total=False):
+    public: bool
+    file_size_limit: int
+    allowed_mime_types: list[str]
+
+
 class ListBucketFilesOptions(TypedDict):
     limit: int
     offset: int

tests/_async/test_client.py

@@ -89,7 +89,7 @@ async def public_bucket(
     global temp_test_buckets_ids
     temp_test_buckets_ids.append(bucket_id)
 
-    await storage.create_bucket(id=bucket_id, public=True)
+    await storage.create_bucket(id=bucket_id, options={"public": True})
 
     yield bucket_id
 

tests/_sync/test_client.py

@@ -87,7 +87,7 @@ def public_bucket(storage: SyncStorageClient, uuid_factory: Callable[[], str]) -
     global temp_test_buckets_ids
     temp_test_buckets_ids.append(bucket_id)
 
-    storage.create_bucket(id=bucket_id, public=True)
+    storage.create_bucket(id=bucket_id, options={"public": True})
 
     yield bucket_id