Async storage buckets

.pre-commit-config.yaml

@@ -20,8 +20,7 @@ repos:
     -   id: autoflake
         args: ['--in-place', '--remove-all-unused-imports']
 
--   repo: https://github.com/ambv/black
+-   repo: https://github.com/psf/black
     rev: 21.5b1
     hooks:
     - id: black
-      language_version: python3.7

pyproject.toml

@@ -11,6 +11,7 @@ postgrest-py = "^0.5.0"
 realtime-py = "^0.1.2"
 gotrue = "0.2.0"
 requests = "2.25.1"
+httpx = "^0.19.0"
 
 [tool.poetry.dev-dependencies]
 pre_commit = "^2.1.0"

supabase/lib/storage/storage_bucket_api.py

@@ -1,87 +1,176 @@
-from typing import Any, Dict
+from __future__ import annotations
 
-import requests
-from requests import HTTPError
+from collections.abc import Awaitable
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Any, Literal, Optional, Type, Union
+
+from httpx import AsyncClient, Client, HTTPError
+
+__all__ = ["Bucket", "StorageBucketAPI"]
+
+_RequestMethod = Literal["GET", "POST", "PUT", "DELETE", "PATCH"]
+
+
+class StorageException(Exception):
+    """Error raised when an operation on the storage API fails."""
+
+
+@dataclass
+class Bucket:
+    id: str
+    name: str
+    owner: str
+    public: bool
+    created_at: datetime
+    updated_at: datetime
+
+    def __post_init__(self) -> None:
+        # created_at and updated_at are returned by the API as ISO timestamps
+        # so we convert them to datetime objects
+        self.created_at = datetime.fromisoformat(self.created_at)  # type: ignore
+        self.updated_at = datetime.fromisoformat(self.updated_at)  # type: ignore
+
+
+ResponseType = Union[
+    dict[
+        str, str
+    ],  # response from an endpoint without a custom response_class, example: create_bucket
+    list[
+        Bucket
+    ],  # response from an endpoint which returns a list of objects, example: list_buckets
+    Bucket,  # response from an endpoint which returns a single object, example: get_bucket
+    None,
+]
 
 
 class StorageBucketAPI:
     """This class abstracts access to the endpoint to the Get, List, Empty, and Delete operations on a bucket"""
 
-    def __init__(self, url, headers):
+    def __init__(
+        self, url: str, headers: dict[str, str], is_async: bool = False
+    ) -> None:
         self.url = url
         self.headers = headers
 
-    def list_buckets(self) -> Dict[str, Any]:
-        """Retrieves the details of all storage buckets within an existing product."""
+        self._is_async = is_async
+
+        if is_async:
+            self._client = AsyncClient(headers=self.headers)
+        else:
+            self._client = Client(headers=self.headers)
+
+    def _request(
+        self,
+        method: _RequestMethod,
+        url: str,
+        json: Optional[dict[Any, Any]] = None,
+        response_class: Optional[Type] = None,
+    ) -> Any:
+        if self._is_async:
+            return self._async_request(method, url, json, response_class)
+        else:
+            return self._sync_request(method, url, json, response_class)
+
+    def _sync_request(
+        self,
+        method: _RequestMethod,
+        url: str,
+        json: Optional[dict[Any, Any]] = None,
+        response_class: Optional[Type] = None,
+    ) -> ResponseType:
+        if isinstance(self._client, AsyncClient):  # only to appease the type checker
+            return
+
+        response = self._client.request(method, url, json=json)
         try:
-            response = requests.get(f"{self.url}/bucket", headers=self.headers)
             response.raise_for_status()
-        except HTTPError as http_err:
-            print(f"HTTP error occurred: {http_err}")  # Python 3.6
-        except Exception as err:
-            print(f"Other error occurred: {err}")  # Python 3.6
+        except HTTPError:
+            raise StorageException(response.json())
+
+        response_data = response.json()
+
+        if not response_class:
+            return response_data
+
+        if isinstance(response_data, list):
+            return [response_class(**item) for item in response_data]
         else:
-            return response.json()
+            return response_class(**response_data)
 
-    def get_bucket(self, id: str) -> Dict[str, Any]:
+    async def _async_request(
+        self,
+        method: _RequestMethod,
+        url: str,
+        json: Optional[dict[Any, Any]] = None,
+        response_class: Optional[Type] = None,
+    ) -> ResponseType:
+        if isinstance(self._client, Client):  # only to appease the type checker
+            return
+
+        response = await self._client.request(method, url, json=json)
+        try:
+            response.raise_for_status()
+        except HTTPError:
+            raise StorageException(response.json())
+
+        response_data = response.json()
+
+        if not response_class:
+            return response_data
+
+        if isinstance(response_data, list):
+            return [response_class(**item) for item in response_data]
+        else:
+            return response_class(**response_data)
+
+    def list_buckets(self) -> Union[list[Bucket], Awaitable[list[Bucket]], None]:
+        """Retrieves the details of all storage buckets within an existing product."""
+        return self._request("GET", f"{self.url}/bucket", response_class=Bucket)
+
+    def get_bucket(self, id: str) -> Union[Bucket, Awaitable[Bucket], None]:
         """Retrieves the details of an existing storage bucket.
 
         Parameters
         ----------
         id
             The unique identifier of the bucket you would like to retrieve.
         """
-        try:
-            response = requests.get(f"{self.url}/bucket/{id}", headers=self.headers)
-            response.raise_for_status()
-        except HTTPError as http_err:
-            print(f"HTTP error occurred: {http_err}")  # Python 3.6
-        except Exception as err:
-            print(f"Other error occurred: {err}")  # Python 3.6
-        else:
-            return response.json()
+        return self._request("GET", f"{self.url}/bucket/{id}", response_class=Bucket)
 
-    def create_bucket(self, id: str) -> Dict[str, Any]:
+    def create_bucket(
+        self, id: str, name: str = None, public: bool = False
+    ) -> Union[dict[str, str], Awaitable[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.
+        public
+            Whether the bucket you are creating should be publicly accessible. Defaults to False.
         """
-        try:
-            response = requests.post(
-                f"{self.url}/bucket", data={"id": id}, headers=self.headers
-            )
-            response.raise_for_status()
-        except HTTPError as http_err:
-            print(f"HTTP error occurred: {http_err}")  # Python 3.6
-        except Exception as err:
-            print(f"Other error occurred: {err}")  # Python 3.6
-        else:
-            return response.json()
+        return self._request(
+            "POST",
+            f"{self.url}/bucket",
+            json={"id": id, "name": name or id, "public": public},
+        )
 
-    def empty_bucket(self, id: str) -> Dict[str, Any]:
+    def empty_bucket(self, id: str) -> Union[dict[str, str], Awaitable[dict[str, str]]]:
         """Removes all objects inside a single bucket.
 
         Parameters
         ----------
         id
             The unique identifier of the bucket you would like to empty.
         """
-        try:
-            response = requests.post(
-                f"{self.url}/bucket/{id}/empty", data={}, headers=self.headers
-            )
-            response.raise_for_status()
-        except HTTPError as http_err:
-            print(f"HTTP error occurred: {http_err}")  # Python 3.6
-        except Exception as err:
-            print(f"Other error occurred: {err}")  # Python 3.6
-        else:
-            return response.json()
+        return self._request("POST", f"{self.url}/bucket/{id}/empty", json={})
 
-    def delete_bucket(self, id: str) -> Dict[str, Any]:
+    def delete_bucket(
+        self, id: str
+    ) -> Union[dict[str, str], Awaitable[dict[str, str]]]:
         """Deletes an existing bucket. Note that you cannot delete buckets with existing objects inside. You must first
         `empty()` the bucket.
 
@@ -90,15 +179,4 @@ def delete_bucket(self, id: str) -> Dict[str, Any]:
         id
             The unique identifier of the bucket you would like to delete.
         """
-        try:
-            response = requests.delete(
-                f"{self.url}/bucket/{id}", data={}, headers=self.headers
-            )
-
-            response.raise_for_status()
-        except HTTPError as http_err:
-            print(f"HTTP error occurred: {http_err}")  # Python 3.6
-        except Exception as err:
-            print(f"Other error occurred: {err}")  # Python 3.6
-        else:
-            return response.json()
+        return self._request("DELETE", f"{self.url}/bucket/{id}", json={})