WIP: Add PUT uploads to object storage client

fastenv/cloud/object_storage.py

@@ -189,9 +189,10 @@ def generate_presigned_url(
         bucket_path: os.PathLike[str] | str,
         *,
         expires: int = 3600,
+        server_side_encryption: Literal["AES256", None] = None,
         service: str = "s3",
     ) -> httpx.URL:
-        """Generate a presigned URL for downloads from S3-compatible object storage.
+        """Generate a presigned URL for S3-compatible object storage.
 
         Requests to S3-compatible object storage can be authenticated either with
         request headers or query parameters. Presigned URLs use query parameters.
@@ -207,6 +208,16 @@ def generate_presigned_url(
         `expires`: seconds until the URL expires. The default and maximum
         expiration times are the same as the AWS CLI and Boto3.
 
+        `server_side_encryption`: optional encryption algorithm to specify for
+        the object storage platform to use to encrypt the file for storage.
+        This method supports AES256 encryption with managed keys,
+        referred to as "SSE-B2" on Backblaze or "SSE-S3" on AWS S3.
+        Backblaze B2 does not currently support server-side encryption configuration
+        in PUT or POST requests to the S3 API. The B2 native API must be used.
+        https://www.backblaze.com/apidocs/b2-upload-file
+        https://www.backblaze.com/b2/docs/server_side_encryption.html
+        https://docs.aws.amazon.com/AmazonS3/latest/userguide/UsingServerSideEncryption.html
+
         `service`: cloud service for which to generate the presigned URL.
 
         https://awscli.amazonaws.com/v2/documentation/api/latest/reference/s3/presign.html
@@ -220,7 +231,11 @@ def generate_presigned_url(
             raise ValueError("Expiration time must be between one second and one week.")
         key = key if (key := str(bucket_path)).startswith("/") else f"/{key}"
         params = self._set_presigned_url_query_params(
-            method, key, expires=expires, service=service
+            method,
+            key,
+            expires=expires,
+            server_side_encryption=server_side_encryption,
+            service=service,
         )
         return httpx.URL(
             scheme="https", host=self._config.bucket_host, path=key, params=params
@@ -232,8 +247,9 @@ def _set_presigned_url_query_params(
         key: str,
         *,
         expires: int,
-        service: str = "s3",
         payload_hash: str = "UNSIGNED-PAYLOAD",
+        server_side_encryption: Literal["AES256", None] = None,
+        service: str = "s3",
     ) -> httpx.QueryParams:
         """Set query parameters for a presigned URL.
 
@@ -270,6 +286,8 @@ def _set_presigned_url_query_params(
         }
         if self._config.session_token:
             params["X-Amz-Security-Token"] = self._config.session_token
+        if server_side_encryption:
+            params["X-Amz-Server-Side-Encryption"] = server_side_encryption
         params["X-Amz-SignedHeaders"] = "host"
         headers = {"host": self._config.bucket_host}
         # 1. create canonical request
@@ -392,6 +410,7 @@ async def upload(
         source: os.PathLike[str] | str | bytes = ".env",
         *,
         content_type: str = "text/plain",
+        method: Literal["POST", "PUT"] = "POST",
         server_side_encryption: Literal["AES256", None] = None,
     ) -> httpx.Response | None:
         """Upload a file to cloud object storage.
@@ -407,16 +426,31 @@ async def upload(
         See Backblaze for a list of supported content types.
         https://www.backblaze.com/b2/docs/content-types.html
 
-        `server_side_encryption`: optional encryption algorithm to specify,
-        which the object storage platform will use to encrypt the file for storage.
+        `method`: HTTP method to use for upload. S3-compatible object storage accepts
+        uploads with HTTP PUT via the PutObject API and presigned URLs, or POST
+        with authentication information in form fields.
+
+        `server_side_encryption`: optional encryption algorithm to specify for
+        the object storage platform to use to encrypt the file for storage.
         This method supports AES256 encryption with managed keys,
         referred to as "SSE-B2" on Backblaze or "SSE-S3" on AWS S3.
+        Backblaze B2 does not currently support server-side encryption configuration
+        in PUT or POST requests to the S3 API. The B2 native API must be used.
+        https://www.backblaze.com/apidocs/b2-upload-file
         https://www.backblaze.com/b2/docs/server_side_encryption.html
         https://docs.aws.amazon.com/AmazonS3/latest/userguide/UsingServerSideEncryption.html
         """
         try:
             content, message = await self._encode_source(source)
-            if self._config.bucket_host.endswith(".backblazeb2.com"):
+            if method == "PUT":
+                url = self.generate_presigned_url(
+                    method,
+                    bucket_path,
+                    expires=30,
+                    server_side_encryption=server_side_encryption,
+                )
+                response = await self._client.put(url, content=content)
+            elif self._config.bucket_host.endswith(".backblazeb2.com"):
                 response = await self.upload_to_backblaze_b2(
                     bucket_path,
                     content,
@@ -479,10 +513,13 @@ def generate_presigned_post(
         See Backblaze for a list of supported content types.
         https://www.backblaze.com/b2/docs/content-types.html
 
-        `server_side_encryption`: optional encryption algorithm to specify,
-        which the object storage platform will use to encrypt the file for storage.
+        `server_side_encryption`: optional encryption algorithm to specify for
+        the object storage platform to use to encrypt the file for storage.
         This method supports AES256 encryption with managed keys,
         referred to as "SSE-B2" on Backblaze or "SSE-S3" on AWS S3.
+        Backblaze B2 does not currently support server-side encryption configuration
+        in PUT or POST requests to the S3 API. The B2 native API must be used.
+        https://www.backblaze.com/apidocs/b2-upload-file
         https://www.backblaze.com/b2/docs/server_side_encryption.html
         https://docs.aws.amazon.com/AmazonS3/latest/userguide/UsingServerSideEncryption.html