Add S3Client for metadata management

Why these changes are being introduced:

With the addition of dataset/metadata assets in S3, we will
need to perform actions like downloading the static DB file,
uploading a new one, and deleting append deltas.

How this addresses that need:

Creates new utility class S3Client that performs these
actions.

Side effects of this change:
* None

Relevant ticket(s):
* https://mitlibraries.atlassian.net/browse/TIMX-530

Author: ghukill

Pipfile

@@ -12,7 +12,6 @@ pyarrow = "*"
 
 [dev-packages]
 black = "*"
-boto3-stubs = {version = "*", extras = ["s3"]}
 coveralls = "*"
 ipython = "*"
 moto = "*"
@@ -25,6 +24,7 @@ pytest = "*"
 ruff = "*"
 setuptools = "*"
 pip-audit = "*"
+boto3-stubs = {extras = ["essential"], version = "*"}
 
 [requires]
 python_version = "3.12"

Pipfile.lock

@@ -1,9 +1,9 @@
 """tests/conftest.py"""
 
-# ruff: noqa: D205, D209
-
 import os
 
+import boto3
+import moto
 import pytest
 
 from tests.utils import (
@@ -198,3 +198,16 @@ def dataset_with_same_day_runs(tmp_path) -> TIMDEXDataset:
 @pytest.fixture
 def timdex_dataset_metadata(dataset_with_same_day_runs):
     return TIMDEXDatasetMetadata(timdex_dataset=dataset_with_same_day_runs)
+
+
+@pytest.fixture
+def timdex_bucket():
+    return "timdex"
+
+
+@pytest.fixture
+def mock_s3_resource(timdex_bucket):
+    with moto.mock_aws():
+        conn = boto3.resource("s3", region_name="us-east-1")
+        conn.create_bucket(Bucket=timdex_bucket)
+        yield conn

tests/conftest.py

@@ -0,0 +1,110 @@
+"""tests/test_s3client.py"""
+
+# ruff: noqa: PLR2004, SLF001
+
+import pytest
+
+from timdex_dataset_api.utils import S3Client
+
+
+def test_s3client_init():
+    """Test S3Client initialization."""
+    client = S3Client()
+    assert client.resource is not None
+
+
+def test_s3client_init_with_minio_env(caplog, monkeypatch):
+    """Test S3Client initialization with MinIO environment variables."""
+    caplog.set_level("DEBUG")
+
+    monkeypatch.setenv("MINIO_S3_ENDPOINT_URL", "http://localhost:9000")
+    monkeypatch.setenv("MINIO_USERNAME", "minioadmin")
+    monkeypatch.setenv("MINIO_PASSWORD", "minioadmin")
+    monkeypatch.setenv("MINIO_REGION", "us-east-1")
+
+    client = S3Client()
+    assert client.resource is not None
+    assert "MinIO env vars detected, using for S3Client" in caplog.text
+
+
+def test_split_s3_uri():
+    """Test _split_s3_uri method."""
+    client = S3Client()
+    bucket, key = client._split_s3_uri("s3://timdex/path/to/file.txt")
+    assert bucket == "timdex"
+    assert key == "path/to/file.txt"
+
+
+def test_split_s3_uri_invalid():
+    """Test _split_s3_uri method with invalid URI."""
+    client = S3Client()
+    with pytest.raises(ValueError, match="Invalid S3 URI"):
+        client._split_s3_uri("timdex/path/to/file.txt")
+
+
+def test_upload_download_file(mock_s3_resource, tmp_path):
+    """Test upload_file and download_file methods."""
+    client = S3Client()
+
+    # Create a test file
+    test_file = tmp_path / "test.txt"
+    test_file.write_text("test content")
+
+    # Upload the file
+    s3_uri = "s3://timdex/test.txt"
+    client.upload_file(test_file, s3_uri)
+
+    # Download the file to a different location
+    download_path = tmp_path / "downloaded.txt"
+    client.download_file(s3_uri, download_path)
+
+    # Verify the content
+    assert download_path.read_text() == "test content"
+
+
+def test_delete_file(mock_s3_resource, tmp_path):
+    """Test delete_file method."""
+    client = S3Client()
+
+    # Create and upload a test file
+    test_file = tmp_path / "test.txt"
+    test_file.write_text("test content")
+    s3_uri = "s3://timdex/test.txt"
+    client.upload_file(test_file, s3_uri)
+
+    # Delete the file
+    client.delete_file(s3_uri)
+
+    # Verify the file is deleted
+    bucket = mock_s3_resource.Bucket("timdex")
+    objects = list(bucket.objects.all())
+    assert len(objects) == 0
+
+
+def test_delete_folder(mock_s3_resource, tmp_path):
+    """Test delete_folder method."""
+    client = S3Client()
+
+    # Create and upload test files
+    for i in range(3):
+        test_file = tmp_path / f"test{i}.txt"
+        test_file.write_text(f"test content {i}")
+        s3_uri = f"s3://timdex/folder/test{i}.txt"
+        client.upload_file(test_file, s3_uri)
+
+    # Upload a file outside the folder
+    other_file = tmp_path / "other.txt"
+    other_file.write_text("other content")
+    client.upload_file(other_file, "s3://timdex/other.txt")
+
+    # Delete the folder
+    deleted_keys = client.delete_folder("s3://timdex/folder/")
+
+    # Verify only folder contents are deleted
+    assert len(deleted_keys) == 3
+    assert all(key.startswith("folder/") for key in deleted_keys)
+
+    bucket = mock_s3_resource.Bucket("timdex")
+    objects = list(bucket.objects.all())
+    assert len(objects) == 1
+    assert objects[0].key == "other.txt"

tests/test_s3client.py

@@ -0,0 +1,76 @@
+"""timdex_dataset_api/utils.py"""
+
+import logging
+import os
+import pathlib
+from urllib.parse import urlparse
+
+import boto3
+from mypy_boto3_s3.service_resource import S3ServiceResource
+
+logger = logging.getLogger(__name__)
+
+
+class S3Client:
+    def __init__(
+        self,
+    ) -> None:
+        self.resource = self._create_resource()
+
+    def _create_resource(self) -> S3ServiceResource:
+        """Instantiate a boto3 S3 resource.
+
+        If env var MINIO_S3_ENDPOINT_URL is set, assume using local set of MinIO env vars.
+        """
+        endpoint_url = os.getenv("MINIO_S3_ENDPOINT_URL")
+        if endpoint_url:
+            logger.debug("MinIO env vars detected, using for S3Client.")
+            return boto3.resource(
+                "s3",
+                endpoint_url=endpoint_url,
+                aws_access_key_id=os.getenv("MINIO_USERNAME"),
+                aws_secret_access_key=os.getenv("MINIO_PASSWORD"),
+                region_name=os.getenv("MINIO_REGION", "us-east-1"),
+            )
+        return boto3.resource("s3")
+
+    def download_file(self, s3_uri: str, local_path: str | pathlib.Path) -> None:
+        bucket, key = self._split_s3_uri(s3_uri)
+        local_path = pathlib.Path(local_path)
+        local_path.parent.mkdir(parents=True, exist_ok=True)
+        self.resource.Bucket(bucket).download_file(key, str(local_path))
+        logger.info(f"Downloaded {s3_uri} to {local_path}")
+
+    def upload_file(self, local_path: str | pathlib.Path, s3_uri: str) -> None:
+        bucket, key = self._split_s3_uri(s3_uri)
+        local_path = pathlib.Path(local_path)
+        self.resource.Bucket(bucket).upload_file(str(local_path), key)
+        logger.info(f"Uploaded {local_path} to {s3_uri}")
+
+    def delete_file(self, s3_uri: str) -> None:
+        bucket, key = self._split_s3_uri(s3_uri)
+        self.resource.Object(bucket, key).delete()
+        logger.info(f"Deleted {s3_uri}")
+
+    def delete_folder(self, s3_uri: str) -> list[str]:
+        """Delete all objects whose keys start with the given prefix."""
+        bucket, prefix = self._split_s3_uri(s3_uri)
+        bucket_obj = self.resource.Bucket(bucket)
+        receipt = bucket_obj.objects.filter(Prefix=prefix).delete()
+
+        deleted_keys = []
+        for request in receipt:
+            deleted_keys.extend([item["Key"] for item in request["Deleted"]])
+        logger.info(f"Deleted {deleted_keys}")
+        return deleted_keys
+
+    @staticmethod
+    def _split_s3_uri(s3_uri: str) -> tuple[str, str]:
+        """Validate and split an S3 URI into (bucket, key)."""
+        parsed = urlparse(s3_uri)
+        if parsed.scheme != "s3" or not parsed.netloc or not parsed.path:
+            raise ValueError(f"Invalid S3 URI: {s3_uri!r}")
+
+        bucket = parsed.netloc
+        key = parsed.path.lstrip("/")  # strip leading slash from /key
+        return bucket, key