File Uploads

class S3StorageBackend(S3Storage):
    """``S3Storage`` configured per instance, for AWS S3 and S3-compatible services.

    ``fastapi_storages.S3Storage`` reads its configuration from ``AWS_*``
    environment variables when the module is imported. This subclass takes the
    same configuration as keyword arguments (or from settings via
    :meth:`from_settings`), so two backends can point at different buckets and
    nothing is read at import time.

    Args:
        bucket_name: The bucket objects are written to.
        access_key_id: Static credentials. When unset, boto3's own credential
            chain applies (profiles, instance and task roles, ``AWS_*`` env).
        secret_access_key: See ``access_key_id``.
        region: The region name (``"eu-central-1"``). Optional for services that
            ignore it (MinIO) but required by AWS without a profile default.
        endpoint_url: ``http(s)://host[:port]`` of an S3-compatible service
            (MinIO, R2, Hetzner, Spaces). Unset for AWS S3. ``http://`` turns SSL
            off for that endpoint.
        custom_domain: Serve object URLs from this host instead of the endpoint
            (a CDN or an R2 public bucket domain). URLs are then not signed.
        querystring_auth: Sign object URLs (presigned, time-limited; default).
            Set ``False`` only for a public-read bucket.
        default_acl: ACL set on every written object (``"public-read"``).
            Unset (default) keeps objects private and sends no ACL, which
            services without ACL support (R2) require.
        client: An existing boto3 S3 client to use instead of building one
            (tests, custom sessions). The other connection arguments are then
            only used to build URLs.
    """

    def __init__(  # noqa: PLR0913 - one keyword per S3 connection setting
        self,
        *,
        bucket_name: str,
        access_key_id: str | None = None,
        secret_access_key: str | None = None,
        region: str | None = None,
        endpoint_url: str | None = None,
        custom_domain: str | None = None,
        querystring_auth: bool = True,
        default_acl: str | None = None,
        client: Any = None,
    ) -> None:
        if not bucket_name:
            msg = "S3StorageBackend requires a bucket name"
            raise ValueError(msg)
        scheme, host = _split_endpoint(endpoint_url)
        self.bucket_name = bucket_name
        self.region = region
        self.custom_domain = custom_domain
        self.querystring_auth = querystring_auth
        self.default_acl = default_acl
        self._http_scheme = scheme
        endpoint = f"{scheme}://{host}" if host else None
        if client is None:
            client = _s3_client(
                **_client_kwargs(
                    use_ssl=scheme == "https",
                    endpoint=endpoint,
                    region=region,
                    access_key_id=access_key_id,
                    secret_access_key=secret_access_key,
                )
            )
        self._s3 = client
        self._url = endpoint or str(client.meta.endpoint_url)

    @classmethod
    def from_settings(cls, settings: HyperAdminSettings, *, client: Any = None) -> S3StorageBackend:
        """Build a backend from the ``aws_*`` settings (``HYPERADMIN_AWS_*`` / ``AWS_*`` env).

        Raises:
            ValueError: ``aws_s3_bucket_name`` is unset.
        """
        if not settings.aws_s3_bucket_name:
            msg = (
                "S3 storage requires a bucket: set aws_s3_bucket_name "
                "(HYPERADMIN_AWS_S3_BUCKET_NAME or AWS_S3_BUCKET_NAME)"
            )
            raise ValueError(msg)
        secret = settings.aws_secret_access_key
        return cls(
            bucket_name=settings.aws_s3_bucket_name,
            access_key_id=settings.aws_access_key_id,
            secret_access_key=secret.get_secret_value() if secret is not None else None,
            region=settings.aws_s3_region,
            endpoint_url=settings.aws_s3_endpoint_url,
            custom_domain=settings.aws_s3_custom_domain,
            querystring_auth=settings.aws_s3_querystring_auth,
            default_acl=settings.aws_s3_default_acl,
            client=client,
        )

    def get_name(self, name: str) -> str:
        """Return the object key for ``name``: the basename sanitised, directories kept."""
        path = PurePosixPath(name.replace("\\", "/"))
        safe = secure_filename(path.name)
        return str(path.parent / safe) if safe else name

    def get_path(self, name: str) -> str:
        """Return the URL of the object ``name``.

        Served from ``custom_domain`` when set, otherwise presigned through the
        client when ``querystring_auth`` is on (default), otherwise the plain
        path-style URL under the endpoint.
        """
        key = self.get_name(name)
        if self.custom_domain:
            return f"{self._http_scheme}://{self.custom_domain}/{key}"
        if self.querystring_auth:
            params = {"Bucket": self.bucket_name, "Key": key}
            return str(self._s3.generate_presigned_url("get_object", Params=params))
        return f"{self._url}/{self.bucket_name}/{key}"

    def get_size(self, name: str) -> int:
        """Return the object size in bytes (one ``HeadObject`` call)."""
        response = self._s3.head_object(Bucket=self.bucket_name, Key=self.get_name(name))
        return int(response["ContentLength"])

    def write(self, file: BinaryIO, name: str) -> str:
        """Upload ``file`` as ``name`` and return the object key.

        No ``ACL`` parameter is sent when ``default_acl`` is unset: boto3
        rejects ``ACL=None``, and services such as R2 reject any ACL at all.
        """
        file.seek(0, 0)
        key = self.get_name(name)
        content_type, _ = mimetypes.guess_type(key)
        params: dict[str, Any] = {"ContentType": content_type or _DEFAULT_CONTENT_TYPE}
        if self.default_acl:
            params["ACL"] = self.default_acl
        self._s3.upload_fileobj(file, self.bucket_name, key, ExtraArgs=params)
        return key

    def delete(self, name: str) -> None:
        """Delete the object ``name`` from the bucket."""
        self._s3.delete_object(Bucket=self.bucket_name, Key=self.get_name(name))