fastapi-storages icon indicating copy to clipboard operation
fastapi-storages copied to clipboard

More s3 configuration options

Open bufke opened this issue 5 months ago • 1 comments

Hello, I've customized this project a bit and would consider submitting patches here. Would you be interested in:

  • AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY should default to None and not get passed into boto3.client. Passing a empty string errors. We can rely instead on boto3 default behavior.
  • AWS_S3_ENDPOINT_URL should allow None, this is useful when using AWS_REGION or AWS_DEFAULT_REGION env vars.
  • Add support for setting ServerSideEncryption and SSEKMSKeyId

Here's my version, but I could submit this in multiple pull requests and review it a bit more first. This is not intended to submit as is, it only works for me.

class CustomS3Storage(S3Storage):
    def __init__(self) -> None:
        self._http_scheme = "https" if self.AWS_S3_USE_SSL else "http"

        client_kwargs = {
            "use_ssl": self.AWS_S3_USE_SSL,
            "config": Config(signature_version="s3v4"), # Should be an option, not hard coded
        }

        if self.AWS_S3_ENDPOINT_URL is not None:
            self._url = f"{self._http_scheme}://{self.AWS_S3_ENDPOINT_URL}"
            client_kwargs["endpoint_url"] = self._url

        if self.AWS_ACCESS_KEY_ID:
            client_kwargs["aws_access_key_id"] = self.AWS_ACCESS_KEY_ID
        if self.AWS_SECRET_ACCESS_KEY:
            client_kwargs["aws_secret_access_key"] = self.AWS_SECRET_ACCESS_KEY

        self._s3 = boto3.client("s3", **client_kwargs)  # type: ignore

        if self.AWS_S3_ENDPOINT_URL is None:
            # Build the URL from the boto client endpoint (which respects AWS_REGION)
            self._url = self._s3.meta.endpoint_url

    def get_full_path(self, name: str) -> str:
        full_path: str = self.get_path(name)
        return full_path

    def write(self, file: BinaryIO, name: str) -> str:
        """
        Write input file which is opened in binary mode to destination.
        Modified to optionally support KMS key
        """
        # If not KMS key is set, use upstream
        if not self.AWS_S3_BUCKET_KMS_KEY_ARN:
            result: str = super().write(file, name)
            return result

        file.seek(0, 0)
        key: str = self.get_name(name)
        content_type, _ = mimetypes.guess_type(key)
        params = {
            "ACL": self.AWS_DEFAULT_ACL,
            "ContentType": content_type or self.default_content_type,
            "ServerSideEncryption": "aws:kms",
            "SSEKMSKeyId": self.AWS_S3_BUCKET_KMS_KEY_ARN,
        }
        self._s3.upload_fileobj(file, self.AWS_S3_BUCKET_NAME, key, ExtraArgs=params)
        return key

bufke avatar Jun 25 '25 18:06 bufke