add s3 compatible object store support with s3+https://...

Author: jhamman

pyproject.toml

@@ -434,6 +434,8 @@ file = "zarr.storage._builtin_adapters:FileSystemAdapter"
 memory = "zarr.storage._builtin_adapters:MemoryAdapter"
 https = "zarr.storage._builtin_adapters:HttpsAdapter"
 s3 = "zarr.storage._builtin_adapters:S3Adapter"
+"s3+http" = "zarr.storage._builtin_adapters:S3HttpAdapter"
+"s3+https" = "zarr.storage._builtin_adapters:S3HttpsAdapter"
 gs = "zarr.storage._builtin_adapters:GSAdapter"
 log = "zarr.storage._builtin_adapters:LoggingAdapter"
 zip = "zarr.storage._builtin_adapters:ZipAdapter"

src/zarr/abc/store_adapter.py

@@ -192,5 +192,5 @@ def __init_subclass__(cls, **kwargs: Any) -> None:
 
         import re
 
-        if not re.match(r"^[a-zA-Z][a-zA-Z0-9_-]*$", cls.adapter_name):
+        if not re.match(r"^[a-zA-Z][a-zA-Z0-9_+-]*$", cls.adapter_name):
             raise ValueError(f"Invalid adapter_name format: {cls.adapter_name}")

src/zarr/storage/_builtin_adapters.py

@@ -223,13 +223,129 @@ def get_supported_schemes(cls) -> list[str]:
 
 
 class S3Adapter(RemoteAdapter):
-    """Store adapter for S3 URLs using fsspec."""
+    """Store adapter for S3 URLs using fsspec.
+
+    Supports:
+    - Standard AWS S3: s3://bucket/path
+    - Custom S3 endpoints: s3+http://endpoint/bucket/path, s3+https://endpoint/bucket/path
+    """
 
     adapter_name = "s3"
 
     @classmethod
     def get_supported_schemes(cls) -> list[str]:
-        return ["s3"]
+        return ["s3", "s3+http", "s3+https"]
+
+    @classmethod
+    def _parse_s3_url(cls, url: str) -> tuple[str, str | None, dict[str, Any]]:
+        """Parse S3 URL and return (s3_url, endpoint_url, storage_options).
+
+        Returns:
+            - s3_url: Normalized s3:// URL for fsspec
+            - endpoint_url: Custom endpoint URL (None for AWS)
+            - storage_options: Additional fsspec configuration
+        """
+        if url.startswith("s3://"):
+            # Standard AWS S3
+            return url, None, {}
+
+        elif url.startswith("s3+http://"):
+            # Custom S3 via HTTP: s3+http://endpoint/bucket/path
+            # Remove "s3+" prefix and parse the remaining http URL
+            http_url = url[3:]  # "http://endpoint/bucket/path"
+
+            # Find the first '/' after "http://"
+            after_protocol = http_url[7:]  # Everything after "http://"
+            if "/" in after_protocol:
+                # Split at the first '/' to separate endpoint from path
+                endpoint_part, path_part = after_protocol.split("/", 1)
+                endpoint_url = f"http://{endpoint_part}"
+                s3_url = f"s3://{path_part}"
+            else:
+                # No path part, just endpoint
+                endpoint_url = http_url
+                s3_url = "s3://"
+
+            storage_options = {"endpoint_url": endpoint_url, "use_ssl": False}
+            return s3_url, endpoint_url, storage_options
+
+        elif url.startswith("s3+https://"):
+            # Custom S3 via HTTPS: s3+https://endpoint/bucket/path
+            # Remove "s3+" prefix and parse the remaining https URL
+            https_url = url[3:]  # "https://endpoint/bucket/path"
+
+            # Find the first '/' after "https://"
+            after_protocol = https_url[8:]  # Everything after "https://"
+            if "/" in after_protocol:
+                # Split at the first '/' to separate endpoint from path
+                endpoint_part, path_part = after_protocol.split("/", 1)
+                endpoint_url = f"https://{endpoint_part}"
+                s3_url = f"s3://{path_part}"
+            else:
+                # No path part, just endpoint
+                endpoint_url = https_url
+                s3_url = "s3://"
+
+            storage_options = {"endpoint_url": endpoint_url, "use_ssl": True}
+            return s3_url, endpoint_url, storage_options
+
+        else:
+            raise ValueError(f"Unsupported S3 URL format: {url}")
+
+    @classmethod
+    async def from_url_segment(
+        cls,
+        segment: URLSegment,
+        preceding_url: str,
+        **kwargs: Any,
+    ) -> Store:
+        """Create an FsspecStore for S3 URLs with custom endpoint support."""
+        from zarr.storage._fsspec import FsspecStore
+
+        # Parse the S3 URL to extract endpoint information
+        s3_url, endpoint_url, endpoint_storage_options = cls._parse_s3_url(preceding_url)
+
+        # Merge storage options (user-provided options take precedence)
+        storage_options = endpoint_storage_options.copy()
+        user_storage_options = kwargs.get("storage_options", {})
+        storage_options.update(user_storage_options)
+
+        # Determine read-only mode (S3 can be writable)
+        read_only = cls._determine_read_only_mode(preceding_url, **kwargs)
+
+        return FsspecStore.from_url(s3_url, storage_options=storage_options, read_only=read_only)
+
+    @classmethod
+    def _extract_scheme(cls, url: str) -> str:
+        """Extract scheme from URL, handling composite schemes."""
+        if url.startswith("s3+http://"):
+            return "s3+http"
+        elif url.startswith("s3+https://"):
+            return "s3+https"
+        elif url.startswith("s3://"):
+            return "s3"
+        else:
+            return url.split("://", 1)[0]
+
+
+class S3HttpAdapter(S3Adapter):
+    """Store adapter for custom S3 HTTP endpoints."""
+
+    adapter_name = "s3+http"
+
+    @classmethod
+    def get_supported_schemes(cls) -> list[str]:
+        return ["s3+http"]
+
+
+class S3HttpsAdapter(S3Adapter):
+    """Store adapter for custom S3 HTTPS endpoints."""
+
+    adapter_name = "s3+https"
+
+    @classmethod
+    def get_supported_schemes(cls) -> list[str]:
+        return ["s3+https"]
 
 
 class GSAdapter(RemoteAdapter):

src/zarr/storage/_register_adapters.py

@@ -22,6 +22,8 @@ def register_builtin_adapters() -> None:
         LoggingAdapter,
         MemoryAdapter,
         S3Adapter,
+        S3HttpAdapter,
+        S3HttpsAdapter,
         ZipAdapter,
     )
 
@@ -33,6 +35,8 @@ def register_builtin_adapters() -> None:
         MemoryAdapter,
         HttpsAdapter,
         S3Adapter,
+        S3HttpAdapter,
+        S3HttpsAdapter,
         GSAdapter,
         LoggingAdapter,
         ZipAdapter,

tests/test_store/test_zep8.py

@@ -18,7 +18,14 @@
 from zarr.core.buffer import default_buffer_prototype
 from zarr.registry import get_store_adapter, register_store_adapter
 from zarr.storage import FsspecStore, LocalStore, LoggingStore, MemoryStore, ZipStore
-from zarr.storage._builtin_adapters import GSAdapter, HttpsAdapter, LoggingAdapter, S3Adapter
+from zarr.storage._builtin_adapters import (
+    GSAdapter,
+    HttpsAdapter,
+    LoggingAdapter,
+    S3Adapter,
+    S3HttpAdapter,
+    S3HttpsAdapter,
+)
 from zarr.storage._common import make_store_path
 from zarr.storage._zep8 import URLParser, URLStoreResolver, ZEP8URLError, is_zep8_url
 
@@ -513,6 +520,8 @@ def test_fsspec_zep8_url_detection() -> None:
     # These should be detected as ZEP 8 URLs
     zep8_urls = [
         "s3://bucket/data.zip|zip:",
+        "s3+http://minio.local:9000/bucket/data.zip|zip:",
+        "s3+https://storage.example.com/bucket/data.zarr|zarr3:",
         "https://example.com/data|zip:|zarr3:",
         "gs://bucket/data.zarr|zarr2:",
     ]
@@ -538,7 +547,7 @@ async def test_fsspec_adapter_error_handling() -> None:
     # Test S3 adapter with invalid URL
     segment = URLSegment(scheme="s3", path="bucket/data", adapter=None)
 
-    with pytest.raises(ValueError, match="Unsupported scheme"):
+    with pytest.raises(ValueError, match="Unsupported S3 URL format"):
         await S3Adapter.from_url_segment(segment, "invalid://url")
 
     # Test HTTPS adapter with invalid URL
@@ -566,7 +575,7 @@ def test_fsspec_schemes_support() -> None:
 
     # Test S3 adapter
     assert S3Adapter.can_handle_scheme("s3")
-    assert S3Adapter.get_supported_schemes() == ["s3"]
+    assert S3Adapter.get_supported_schemes() == ["s3", "s3+http", "s3+https"]
 
     # Test HTTPS adapter
     assert HttpsAdapter.can_handle_scheme("https")
@@ -589,6 +598,8 @@ async def test_fsspec_url_chain_parsing() -> None:
     # Test complex chained URLs
     complex_urls = [
         "s3://bucket/archive.zip|zip:data/|zarr3:group",
+        "s3+http://minio.local:9000/bucket/archive.zip|zip:data/|zarr3:group",
+        "s3+https://storage.example.com/bucket/nested.zip|zip:inner/|zarr2:",
         "https://example.com/data.tar.gz|tar:|zip:|zarr2:",
         "gs://bucket/dataset.zarr|zarr3:array/subarray",
     ]
@@ -1223,16 +1234,149 @@ async def test_remote_adapter_comprehensive() -> None:
 
 
 async def test_s3_adapter_functionality() -> None:
-    """Test S3Adapter functionality."""
+    """Test S3Adapter functionality including custom endpoints."""
     from zarr.storage._builtin_adapters import S3Adapter
 
-    # Test can_handle_scheme
+    # Test can_handle_scheme for all supported schemes
     assert S3Adapter.can_handle_scheme("s3")
+    assert S3Adapter.can_handle_scheme("s3+http")
+    assert S3Adapter.can_handle_scheme("s3+https")
     assert not S3Adapter.can_handle_scheme("gs")
+    assert not S3Adapter.can_handle_scheme("http")
 
     # Test get_supported_schemes
     schemes = S3Adapter.get_supported_schemes()
     assert "s3" in schemes
+    assert "s3+http" in schemes
+    assert "s3+https" in schemes
+    assert len(schemes) == 3
+
+
+async def test_s3_custom_endpoint_url_parsing() -> None:
+    """Test S3Adapter URL parsing for custom endpoints."""
+    from zarr.storage._builtin_adapters import S3Adapter
+
+    # Test standard AWS S3 URL parsing
+    s3_url, endpoint_url, storage_options = S3Adapter._parse_s3_url("s3://my-bucket/path/to/data")
+    assert s3_url == "s3://my-bucket/path/to/data"
+    assert endpoint_url is None
+    assert storage_options == {}
+
+    # Test custom HTTP endpoint parsing
+    s3_url, endpoint_url, storage_options = S3Adapter._parse_s3_url(
+        "s3+http://minio.local:9000/my-bucket/data"
+    )
+    assert s3_url == "s3://my-bucket/data"
+    assert endpoint_url == "http://minio.local:9000"
+    assert storage_options == {"endpoint_url": "http://minio.local:9000", "use_ssl": False}
+
+    # Test custom HTTPS endpoint parsing
+    s3_url, endpoint_url, storage_options = S3Adapter._parse_s3_url(
+        "s3+https://storage.example.com/bucket/path/file.zarr"
+    )
+    assert s3_url == "s3://bucket/path/file.zarr"
+    assert endpoint_url == "https://storage.example.com"
+    assert storage_options == {"endpoint_url": "https://storage.example.com", "use_ssl": True}
+
+    # Test custom HTTP endpoint with port
+    s3_url, endpoint_url, storage_options = S3Adapter._parse_s3_url(
+        "s3+http://localhost:9000/test-bucket"
+    )
+    assert s3_url == "s3://test-bucket"
+    assert endpoint_url == "http://localhost:9000"
+    assert storage_options["endpoint_url"] == "http://localhost:9000"
+    assert storage_options["use_ssl"] is False
+
+    # Test edge case: endpoint without path
+    s3_url, endpoint_url, storage_options = S3Adapter._parse_s3_url("s3+https://minio.example.com")
+    assert s3_url == "s3://"
+    assert endpoint_url == "https://minio.example.com"
+
+
+async def test_s3_custom_endpoint_scheme_extraction() -> None:
+    """Test S3Adapter scheme extraction for custom endpoints."""
+    from zarr.storage._builtin_adapters import S3Adapter
+
+    # Test scheme extraction
+    assert S3Adapter._extract_scheme("s3://bucket/path") == "s3"
+    assert S3Adapter._extract_scheme("s3+http://minio.local:9000/bucket") == "s3+http"
+    assert S3Adapter._extract_scheme("s3+https://storage.example.com/bucket") == "s3+https"
+
+
+async def test_s3_custom_endpoint_error_handling() -> None:
+    """Test S3Adapter error handling for invalid URLs."""
+    from zarr.storage._builtin_adapters import S3Adapter
+
+    # Test unsupported URL format
+    with pytest.raises(ValueError, match="Unsupported S3 URL format"):
+        S3Adapter._parse_s3_url("invalid://not-s3")
+
+    with pytest.raises(ValueError, match="Unsupported S3 URL format"):
+        S3Adapter._parse_s3_url("gs://bucket/path")
+
+
+async def test_s3_custom_endpoint_registration() -> None:
+    """Test that custom S3 endpoint schemes are properly registered."""
+    from zarr.registry import get_store_adapter
+
+    # Test that all S3 schemes can be retrieved
+    s3_adapter = get_store_adapter("s3")
+    assert s3_adapter is not None
+    assert s3_adapter == S3Adapter
+
+    s3_http_adapter = get_store_adapter("s3+http")
+    assert s3_http_adapter is not None
+    assert s3_http_adapter == S3HttpAdapter
+
+    s3_https_adapter = get_store_adapter("s3+https")
+    assert s3_https_adapter is not None
+    assert s3_https_adapter == S3HttpsAdapter
+
+
+async def test_s3_http_adapter_functionality() -> None:
+    """Test S3HttpAdapter specific functionality."""
+    # Test adapter name
+    assert S3HttpAdapter.adapter_name == "s3+http"
+
+    # Test supported schemes
+    schemes = S3HttpAdapter.get_supported_schemes()
+    assert schemes == ["s3+http"]
+
+    # Test can_handle_scheme
+    assert S3HttpAdapter.can_handle_scheme("s3+http")
+    assert not S3HttpAdapter.can_handle_scheme("s3")
+    assert not S3HttpAdapter.can_handle_scheme("s3+https")
+
+
+async def test_s3_https_adapter_functionality() -> None:
+    """Test S3HttpsAdapter specific functionality."""
+    # Test adapter name
+    assert S3HttpsAdapter.adapter_name == "s3+https"
+
+    # Test supported schemes
+    schemes = S3HttpsAdapter.get_supported_schemes()
+    assert schemes == ["s3+https"]
+
+    # Test can_handle_scheme
+    assert S3HttpsAdapter.can_handle_scheme("s3+https")
+    assert not S3HttpsAdapter.can_handle_scheme("s3")
+    assert not S3HttpsAdapter.can_handle_scheme("s3+http")
+
+
+async def test_s3_custom_endpoint_zep8_url_detection() -> None:
+    """Test ZEP 8 URL detection with custom S3 endpoints."""
+    from zarr.storage._zep8 import is_zep8_url
+
+    # Standard S3 URLs (not ZEP 8)
+    assert not is_zep8_url("s3://bucket/data")
+    assert not is_zep8_url("s3+http://minio.local:9000/bucket/data")
+    assert not is_zep8_url("s3+https://storage.example.com/bucket/data")
+
+    # ZEP 8 URLs with custom S3 endpoints
+    assert is_zep8_url("s3://bucket/data.zip|zip:")
+    assert is_zep8_url("s3+http://minio.local:9000/bucket/data.zip|zip:")
+    assert is_zep8_url("s3+https://storage.example.com/bucket/data|zarr3:")
+    assert is_zep8_url("s3+http://localhost:9000/bucket/archive.zip|zip:data/|zarr2:")
 
 
 async def test_gcs_adapter_functionality() -> None:
@@ -1346,6 +1490,8 @@ def test_builtin_adapters_imports_and_module_structure() -> None:
         MemoryAdapter,
         RemoteAdapter,
         S3Adapter,
+        S3HttpAdapter,
+        S3HttpsAdapter,
         ZipAdapter,
     )
 
@@ -1354,13 +1500,17 @@ def test_builtin_adapters_imports_and_module_structure() -> None:
     assert MemoryAdapter.adapter_name == "memory"
     assert RemoteAdapter.adapter_name == "remote"
     assert S3Adapter.adapter_name == "s3"
+    assert S3HttpAdapter.adapter_name == "s3+http"
+    assert S3HttpsAdapter.adapter_name == "s3+https"
     assert GSAdapter.adapter_name == "gs"
     assert HttpsAdapter.adapter_name == "https"
     assert LoggingAdapter.adapter_name == "log"
     assert ZipAdapter.adapter_name == "zip"
 
     # Test inheritance relationships
     assert issubclass(S3Adapter, RemoteAdapter)
+    assert issubclass(S3HttpAdapter, S3Adapter)
+    assert issubclass(S3HttpsAdapter, S3Adapter)
     assert issubclass(GSAdapter, RemoteAdapter)
     assert issubclass(HttpsAdapter, RemoteAdapter)