Python Wasabi Client

wasabi_client.py

from typing import Any, Optional
from urllib.parse import urlparse, urlunparse
from xml.etree import ElementTree

from botocore.auth import S3SigV4Auth
from botocore.awsrequest import AWSRequest
from botocore.credentials import Credentials
from django.conf import settings
from requests import Response, request

VALID_REGIONS = ["us-east-1", "us-west-1"]


def sanitize_url(url):
    # Validate the URL, filtering out any empty segments from the path
    parsed_url = urlparse(url)
    path_segments = [str(segment) for segment in parsed_url.path.split("/") if segment]
    sanitized_path = "/".join(path_segments)
    # Handle trailing slash
    sanitized_path += "/" if parsed_url.path.endswith("/") else ""
    sanitized_url = urlunparse(parsed_url._replace(path=sanitized_path))
    return sanitized_url


def print_xml(xml_string):
    # Helper method to pretty print XML response from Wasabi
    element = ElementTree.XML(xml_string)
    ElementTree.indent(element)
    print(ElementTree.tostring(element, encoding="unicode"))


class WasabiApiClient:
    def __init__(
        self,
        access_key_id=settings.WASABI_S3_ACCESS_KEY_ID,
        secret_access_key=settings.WASABI_S3_SECRET_ACCESS_KEY,
        region="us-west-1",
    ):
        self.credentials = Credentials(access_key_id, secret_access_key)
        if region not in VALID_REGIONS:
            raise ValueError(f"{region} is not a valid region")
        self.region = region

    def signed_request(
        self,
        method: str,
        url: str,
        data: Optional[Any] = None,
        params: Optional[Any] = None,
        headers: Optional[dict] = None,
    ) -> Response:
        aws_request = AWSRequest(method=method, url=url, data=data, params=params, headers=headers)
        S3SigV4Auth(self.credentials, "s3", self.region).add_auth(aws_request)
        prepared_request = aws_request.prepare()
        return request(
            method=method, url=prepared_request.url, headers=prepared_request.headers, data=data
        )

    def move(
        self, source: str, destination: str, overwrite=False, quiet=False, prefix=False
    ) -> Response:
        """
        See official docs for details
        https://docs.wasabi.com/docs/operations-on-objects#renaming-objects
        """

        url = f"https://s3.{self.region}.wasabisys.com/{source}"
        if prefix:
            if source[-1] != "/":
                raise ValueError(
                    "Attempting to rename a prefix without a trailing slash in the source prefix. "
                    "This will result in data loss."
                )
            if destination[0] == "/":
                raise ValueError(
                    "Attempting to rename a prefix with a leading slash in the destination prefix. "
                    "This will result in data loss."
                )
            if destination[-1] != "/":
                raise ValueError(
                    "Attempting to rename a prefix without a trailing slash in the destination prefix. "
                    "This will result in data loss."
                )
        url = sanitize_url(url)
        headers = {
            "Destination": destination,
            "Overwrite": str(overwrite).lower(),
            "X-Wasabi-Quiet": str(quiet).lower(),
            "X-Wasabi-Prefix": str(prefix).lower(),
        }
        return self.signed_request("MOVE", url, headers=headers)

Do not copy paste this blindly...
It has a lot of custom logic unique to our use case...