Fixing all but one test

Author: Bryan Lawrence

activestorage/backends.py

@@ -44,25 +44,47 @@ def __init__(self, active):
     def reduce_chunk(self, request: ChunkRequest) -> ChunkResult:
         simulate_cbor = bool((self._active.storage_options or {}).get("local_simulate_cbor"))
         method = self._active._methods.get(request.method) if request.method else None
-        data, count = reduce_chunk(
-            request.uri,
-            request.offset,
-            request.size,
-            request.compressor,
-            request.filters,
-            (
-                request.missing.fill_value,
-                request.missing.missing_value,
-                request.missing.valid_min,
-                request.missing.valid_max,
-            ),
-            request.dtype,
-            request.chunks,
-            request.order,
-            request.chunk_selection,
-            method=method,
-            axis=request.axis,
+        missing = (
+            request.missing.fill_value,
+            request.missing.missing_value,
+            request.missing.valid_min,
+            request.missing.valid_max,
         )
+
+        # Remote datasets loaded through fsspec/pyfive need fh-based reads,
+        # not open(uri, "rb") on the URL string.
+        parsed = urllib.parse.urlparse(str(request.uri))
+        if parsed.scheme in ("http", "https", "s3"):
+            fh = self._active._format.file_handle
+            data, count = reduce_opens3_chunk(
+                fh,
+                request.offset,
+                request.size,
+                request.compressor,
+                request.filters,
+                missing,
+                request.dtype,
+                request.chunks,
+                request.order,
+                request.chunk_selection,
+                method=method,
+                axis=request.axis,
+            )
+        else:
+            data, count = reduce_chunk(
+                request.uri,
+                request.offset,
+                request.size,
+                request.compressor,
+                request.filters,
+                missing,
+                request.dtype,
+                request.chunks,
+                request.order,
+                request.chunk_selection,
+                method=method,
+                axis=request.axis,
+            )
         if simulate_cbor:
             payload = reductionist.encode_result(data, count)
             data, count = reductionist.decode_result_buffer(payload)
@@ -118,12 +140,20 @@ def reduce_chunk(self, request: ChunkRequest) -> ChunkResult:
 
         bucket, obj = self._resolve_bucket_object(request.uri, self._active.storage_options)
         if self._active.storage_options is None:
-            source = S3_URL
+            endpoint = S3_URL
             server = S3_ACTIVE_STORAGE_URL
         else:
-            source = get_endpoint_url(self._active.storage_options, request.uri)
+            endpoint = get_endpoint_url(self._active.storage_options, request.uri)
             server = self._active.active_storage_url or S3_ACTIVE_STORAGE_URL
 
+        endpoint = str(endpoint).rstrip("/")
+        bucket = str(bucket).strip("/")
+        obj = str(obj).lstrip("/")
+        if bucket:
+            source = f"{endpoint}/{bucket}/{obj}"
+        else:
+            source = f"{endpoint}/{obj}"
+
         session = self.get_session()
         data, count = reductionist.reduce_chunk(
             session,
@@ -146,6 +176,7 @@ def reduce_chunk(self, request: ChunkRequest) -> ChunkResult:
             axis=request.axis,
             operation=request.method,
             interface_type='s3',
+            option_disable_chunk_cache=self._active._option_disable_chunk_cache,
         )
         self.close_session(session)
         return ChunkResult(data=data, count=count, out_selection=())
@@ -184,6 +215,7 @@ def reduce_chunk(self, request: ChunkRequest) -> ChunkResult:
             axis=request.axis,
             operation=request.method,
             interface_type='https',
+            option_disable_chunk_cache=self._active._option_disable_chunk_cache,
         )
         self.close_session(session)
         return ChunkResult(data=data, count=count, out_selection=())

activestorage/core.py

@@ -144,6 +144,7 @@ def _select_backend(interface_type, version):
         ("s3", 1): S3Backend,
         ("s3", 2): S3Backend,
         ("https", 0): HttpsBackend,
+        ("https", 1): LocalBackend,
         ("https", 2): HttpsBackend,
         ("p5rem", 0): P5RemBackend,
         ("p5rem", 1): P5RemBackend,
@@ -183,28 +184,58 @@ def __new__(cls, *args, **kwargs):
     def __init__(
         self,
         uri,
-        ncvar,
+        ncvar=None,
         storage_type=None,
+        interface_type=None,
         max_threads=100,
         storage_options=None,
         active_storage_url=None,
         axis=None,
+        option_disable_chunk_cache=False,
     ):
         self.uri = uri
         if self.uri is None:
             raise ValueError(f"Must use a valid file for uri. Got {uri}")
 
-        self.storage_type = storage_type or return_interface_type(uri)
-        self.storage_options = storage_options
+        # Keep source URI when a dataset/variable object is provided.
+        is_pathlike = isinstance(uri, (str, bytes, os.PathLike))
+        source_uri = uri
+        if not is_pathlike:
+            file_obj = getattr(uri, "file", None)
+            fh = getattr(file_obj, "_fh", None)
+            source_uri = (
+                getattr(fh, "path", None)
+                or getattr(fh, "url", None)
+                or str(uri)
+            )
+
+        # interface_type is an alias for storage_type
+        if interface_type is not None:
+            storage_type = interface_type
+
+        self.storage_type = storage_type or return_interface_type(source_uri)
+        self.storage_options = storage_options or {}
+        if self.storage_type is None:
+            # Backward-compatible inference for bare S3 object paths like
+            # "bucket/key.nc" when only storage_options indicate S3 access.
+            # Only infer when the URI is not an existing local file.
+            s3_hints = {"key", "secret", "anon", "client_kwargs", "endpoint_url"}
+            if any(k in self.storage_options for k in s3_hints):
+                if not (is_pathlike and os.path.isfile(str(uri))):
+                    self.storage_type = "s3"
+        self._option_disable_chunk_cache = bool(option_disable_chunk_cache)
         self.active_storage_url = active_storage_url
 
-        if not os.path.isfile(self.uri) and not self.storage_type:
+        # Allow passing dataset/variable objects directly (ncvar optional).
+        is_file_object = not is_pathlike
+        if is_pathlike and not os.path.isfile(self.uri) and not self.storage_type:
             raise ValueError(f"Must use existing file for uri. {self.uri} not found")
 
-        self.ncvar = ncvar
-        if self.ncvar is None:
+        # When uri is a dataset object, ncvar can be None (user will select variable via indexing)
+        if ncvar is None and not is_file_object:
             raise ValueError("Must set a netCDF variable name to slice")
 
+        self._ncvar = ncvar
         self._version = 1
         self._components = False
         self._method = None
@@ -213,9 +244,19 @@ def __init__(
         self.metric_data = {}
         self.data_read = 0
 
-        self._format = _select_format(uri)()
-        self._format.open(uri, storage_options)
-        self.ds = self._format.get_variable(ncvar)
+        self._format = _select_format(source_uri)()
+        self._format._storage_type = self.storage_type
+        if is_file_object:
+            # uri is already a pyfive.Group or similar
+            self._format._dataset = uri
+            self._format._uri = str(source_uri)
+            self.ds = uri
+        else:
+            self._format.open(uri, self.storage_options)
+            if ncvar is not None:
+                self.ds = self._format.get_variable(ncvar)
+            else:
+                self.ds = None
         self.missing = None
 
         self._refresh_backend()
@@ -228,6 +269,10 @@ def ncvar(self):
     def ncvar(self, value):
         self._ncvar = value
 
+    @property
+    def interface_type(self):
+        return self.storage_type
+
     def _refresh_backend(self):
         backend_cls = _select_backend(self.storage_type, self._version)
         self._backend = backend_cls(self)

activestorage/formats.py

@@ -19,10 +19,15 @@ def open(self, uri, storage_options):
         scheme = ""
         if "://" in self._uri:
             scheme = self._uri.split("://", 1)[0]
-        if scheme == "s3":
+        storage_type = getattr(self, '_storage_type', None) or scheme
+        if storage_type == "s3":
             from activestorage import active as active_module
 
             self._dataset_file = active_module.load_from_s3(self._uri, storage_options)
+        elif storage_type in ("http", "https"):
+            from activestorage import active as active_module
+
+            self._dataset_file = active_module.load_from_https(self._uri, storage_options)
         else:
             self._dataset_file = pyfive.File(self._uri)
 

activestorage/helpers.py

@@ -15,18 +15,26 @@ def load_from_https(uri, storage_options=None):
     Load a pyfive.high_level.Dataset from a netCDF4 file on an https server.
     Works for both http and https endpoints.
     """
-    if storage_options is None:
-        client_kwargs = {'auth': None}
-        fs = fsspec.filesystem('http', **client_kwargs)
-        http_file = fs.open(uri, 'rb')
-    else:
-        username = storage_options.get("username", None)
-        password = storage_options.get("password", None)
-        client_kwargs = {
-            'auth': aiohttp.BasicAuth(username, password) if username and password else None
-        }
-        fs = fsspec.filesystem('http', **client_kwargs)
-        http_file = fs.open(uri, 'rb')
+    try:
+        if storage_options is None:
+            client_kwargs = {'auth': None}
+            fs = fsspec.filesystem('http', **client_kwargs)
+            http_file = fs.open(uri, 'rb')
+        else:
+            username = storage_options.get("username", None)
+            password = storage_options.get("password", None)
+            client_kwargs = {
+                'auth': aiohttp.BasicAuth(username, password) if username and password else None
+            }
+            fs = fsspec.filesystem('http', **client_kwargs)
+            http_file = fs.open(uri, 'rb')
+    except FileNotFoundError as exc:
+        # fsspec wraps all failures as FileNotFoundError.
+        # Distinguish by cause: connection-level errors (bad hostname, refused)
+        # have an OSError cause; HTTP-level errors (404) do not.
+        if isinstance(exc.__cause__, OSError):
+            raise ValueError(f"Failed to access HTTPS dataset: {uri}") from exc
+        raise
 
     ds = pyfive.File(http_file)
     print(f"Dataset loaded from https with Pyfive: {uri}")

pyproject.toml

@@ -1,3 +1,79 @@
 [build-system]
 requires = ["setuptools >= 40.6.0", "wheel", "setuptools_scm>=6.2"]
 build-backend = "setuptools.build_meta"
+
+[project]
+name = "ActiveStorage"
+dynamic = ["version"]
+description = ""
+readme = {file = "README.md", content-type = "text/markdown"}
+classifiers = [
+    "Development Status :: 0 - Prototype",
+    "Environment :: Console",
+    "Intended Audience :: Developers",
+    "Intended Audience :: Science/Research",
+    "Natural Language :: English",
+    "Operating System :: POSIX :: Linux",
+    "Programming Language :: Python :: 3",
+    "Programming Language :: Python :: 3.10",
+    "Programming Language :: Python :: 3.11",
+    "Programming Language :: Python :: 3.12",
+    "Programming Language :: Python :: 3.13",
+    "Topic :: Scientific/Engineering",
+    "Topic :: Scientific/Engineering :: Atmospheric Science",
+    "Topic :: Scientific/Engineering :: GIS",
+    "Topic :: Scientific/Engineering :: Hydrology",
+    "Topic :: Scientific/Engineering :: Physics",
+]
+dependencies = [
+    "dask!=2024.8.0",  # github.com/dask/dask/issues/11296
+    "fsspec",
+    "h5netcdf",
+    "h5py",
+    "kerchunk>=0.2.4",
+    "netcdf4",
+    "numcodecs>=0.12",  # github/issues/162
+    "numpy!=1.24.3",   # severe masking bug
+    "requests",
+    "s3fs>=2024.2.0",
+    "zarr>=2.13.3",    # github.com/zarr-developers/zarr-python/issues/1362
+]
+
+[project.optional-dependencies]
+test = [
+    "moto",
+    "pytest",
+    "pytest-cov>=2.10.1",
+    "pytest-html!=2.1.0",
+    "pytest-metadata>=1.5.1",
+    "pytest-xdist",
+]
+
+[tool.setuptools]
+packages = ["activestorage"]
+include-package-data = true
+
+[tool.setuptools_scm]
+
+[tool.pytest.ini_options]
+addopts = """
+    -m 'not slow'
+    --ignore=old_code/
+    --ignore=tests/s3_exploratory
+    --ignore=bnl
+    --cov=activestorage
+    --cov-report=xml:test-reports/coverage.xml
+    --cov-report=html:test-reports/coverage_html
+    --html=test-reports/report.html
+"""
+markers = [
+    "slow: marks tests as slow (deselect with '-m \"not slow\"')",
+]
+
+[tool.coverage.run]
+parallel = true
+
+[tool.coverage.report]
+exclude_lines = [
+    "if __name__ == .__main__.:",
+]