fastapi-storages
fastapi-storages copied to clipboard
More s3 configuration options
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