Rr 114 expose metadata

cfa/dataops/catalog.py

@@ -6,6 +6,7 @@
 import pkgutil
 from collections.abc import Sequence
 from configparser import ConfigParser
+from dataclasses import dataclass
 from importlib import import_module
 from io import BytesIO
 from pathlib import PurePosixPath
@@ -44,6 +45,18 @@
 _config.read(os.path.join(_here, "config.ini"))
 
 logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class VersionMetadata:
+    """Result of resolving a version specification against available versions."""
+
+    version: str | None
+    blob_url: str | None
+    version_spec: str | None
+    selection: str
+
+
 if not logger.handlers:
     logger.addHandler(logging.NullHandler())
 
@@ -235,18 +248,18 @@ def write_blob(
     def read_blobs(
         self,
         version_spec: str | None = None,
-        selection: Literal["newest", "oldest", "all"] = "newest",
+        selection: Literal["newest", "oldest"] = "newest",
         print_version: bool = True,
     ) -> list[bytes]:
         """Read a blob in as bytes so it can be loaded into a dataframe
 
         Args:
             version_spec (str | None, optional): the version of the data to read.
                 Defaults to "latest".
-            selection (Literal["newest", "oldest", "all"], optional): whether to get the newest, oldest, or all matching versions. Defaults to "newest".
+            selection (Literal["newest", "oldest"], optional): whether to get the newest or oldest matching versions. Defaults to "newest".
             print_version (bool, optional): whether to print the version being used. Defaults to True.
         """
-        blobs = self._get_version_blobs(
+        blobs, _ = self._get_version_blobs(
             version_spec=version_spec, selection=selection, print_version=print_version
         )
         blob_bytes = [
@@ -296,48 +309,51 @@ def get_versions(self) -> list:
     def get_file_ext(
         self,
         version_spec: str | None = None,
-        selection: Literal["newest", "oldest", "all"] = "newest",
+        selection: Literal["newest", "oldest"] = "newest",
     ) -> str:
         """returns the file extension for handy routing of read byte types for
         DataFrame reading
 
         Args:
             version_spec (str | None, optional): the version specifier to get.
-            selection (Literal["newest", "oldest", "all"], optional): which version to select. Defaults to "newest".
+            selection (Literal["newest", "oldest"], optional): which version to select. Defaults to "newest".
         Returns:
             str: the file extension
         """
-        return self._get_version_blobs(
+        blobs, _ = self._get_version_blobs(
             version_spec=version_spec, selection=selection, print_version=False
-        )[0]["name"].split(".")[-1]
+        )
+        return blobs[0]["name"].split(".")[-1]
 
     def _get_version_blobs(
         self,
         version_spec: str | None = None,
-        selection: Literal["newest", "oldest", "all"] = "newest",
+        selection: Literal["newest", "oldest"] = "newest",
         print_version=True,
-    ) -> list:
+    ) -> tuple[list, str | None]:
         """Return blob metadata for the requested version selection.
 
         Args:
             version_spec (str | None, optional): Version specifier to pass through to
                 ``version_matcher``. Defaults to ``None``.
-            selection (Literal["newest", "oldest", "all"], optional): When matching multiple versions, choose the
-                newest matching version when ``"newest"``, the oldest matching version
-                when ``"oldest"``, or all matching versions when ``"all"``.
+            selection (Literal["newest", "oldest"], optional): When matching multiple versions, choose the
+                newest matching version when ``"newest"``, or the oldest matching version
+                when ``"oldest"``.
             print_version (bool, optional): Whether to print the resolved version
                 before fetching blobs.
 
         Returns:
             list: Blob metadata dictionaries sorted by creation time for the
-            resolved version or versions.
+            resolved version.
+            str | None: version string for resolved version
 
         Raises:
             ValueError: If the requested version cannot be resolved.
         """
         # check credential access
         if not check_ext_env():
             raise RuntimeError("No EXT access configured.")
+        version = None
         if not self.is_ledger:
             available_versions = self.get_versions()
             version = version_matcher(
@@ -350,58 +366,42 @@ def _get_version_blobs(
             logger.info(f"Using version: {version}")
             if print_version:
                 print(f"Using version: {version}")
-            if isinstance(version, list):
-                walk_path = [f"{self.prefix}/{v}/" for v in version]
-            else:
-                walk_path = f"{self.prefix}/{version}/"
+            walk_path = f"{self.prefix}/{version}/"
         else:
             walk_path = f"{self.prefix.removesuffix('/')}/"
-        if isinstance(walk_path, list):
-            all_blobs = []
-            for wp in walk_path:
-                all_blobs.extend(
-                    walk_blobs_in_container(
-                        name_starts_with=wp,
-                        account_name=self.account,
-                        container_name=self.container,
-                    )
+        return sorted(
+            list(
+                walk_blobs_in_container(
+                    name_starts_with=walk_path,
+                    account_name=self.account,
+                    container_name=self.container,
                 )
-            return sorted(
-                all_blobs,
-                key=lambda x: x["creation_time"],
-            )
-        else:
-            return sorted(
-                list(
-                    walk_blobs_in_container(
-                        name_starts_with=walk_path,
-                        account_name=self.account,
-                        container_name=self.container,
-                    )
-                ),
-                key=lambda x: x["creation_time"],
-            )
+            ),
+            key=lambda x: x["creation_time"],
+        ), version
 
     def download_version_to_local(
         self,
         local_path: str,
         version_spec: str | None = None,
         force: bool = False,
-        selection: Literal["newest", "oldest", "all"] = "newest",
+        selection: Literal["newest", "oldest"] = "newest",
     ) -> bool:
         """Download a specific version of the data to a local path
 
         Args:
             local_path (str): the local path to download to
             version_spec (str | None, optional): the version specifier to download. Defaults to None.
             force (bool, optional): whether to force re-download if local.
-            selection (Literal["newest", "oldest", "all"], optional): which version to select. Defaults to "newest".
+            selection (Literal["newest", "oldest"], optional): which version to select. Defaults to "newest".
         Returns:
             bool: whether any files were written
         """
 
         written = False
-        blobs = self._get_version_blobs(version_spec=version_spec, selection=selection)
+        blobs, _ = self._get_version_blobs(
+            version_spec=version_spec, selection=selection
+        )
         for blob in blobs:
             blob_data = read_blob_stream(
                 blob_url=blob["name"],
@@ -467,8 +467,8 @@ def get_dataframe(
                 either 'pandas' or 'polars' or 'pl_lazy'. Defaults to "pandas".
             version_spec (str, optional): the version of the data to get.
                 Defaults to "latest".
-            selection (Literal["newest", "oldest", "all"], optional): whether to get the newest, oldest, or all matching versions. Defaults to "newest".
-            print_version (bool, optional): whether to log the resolved version information. Defaults to False.
+            selection (Literal["newest", "oldest"], optional): whether to get the newest or oldest matching versions. Defaults to "newest".
+            print_version (bool, optional): whether to print the version being used. Defaults to False.
 
         Raises:
             ValueError: if output is not one of
@@ -483,20 +483,23 @@ def get_dataframe(
             raise ValueError(
                 f"Output {output} needs to be 'pandas', 'polars', 'pd', 'pl', 'pl_lazy', or 'lazy'."
             )
+
         # Fetch version blobs once and validate before deriving file extension.
-        version_blobs = self._get_version_blobs(
+        version_blobs, _ = self._get_version_blobs(
             version_spec=version_spec, selection=selection, print_version=print_version
         )
+        version_meta = self.resolve_version(
+            version_spec=version_spec, selection=selection
+        )
         if not version_blobs:
             raise ValueError(
                 f"No blobs found for version '{version_spec}' in container '{self.container}'."
             )
-        name = version_blobs[0]["name"]
-        file_ext = PurePosixPath(name).suffix.lstrip(".").lower()
+
+        file_ext = version_meta.blob_url.split(".")[-1].lower()
+        fullpath = version_meta.blob_url
         if output in ["pl_lazy", "lazy"]:
             if file_ext in ["parquet", "parq"]:
-                path = str(PurePosixPath(name).parent / f"*.{file_ext}")
-                fullpath = f"az://{self.container}/{path}"
                 df = pl.scan_parquet(
                     fullpath,
                     storage_options={"account_name": self.account},
@@ -507,8 +510,6 @@ def get_dataframe(
                 # self.ledger_entry(action="read")
                 return df
             elif file_ext == "csv":
-                path = str(PurePosixPath(name).parent / f"*.{file_ext}")
-                fullpath = f"az://{self.container}/{path}"
                 df = pl.scan_csv(
                     fullpath,
                     infer_schema_length=None,
@@ -520,8 +521,6 @@ def get_dataframe(
                 ##self.ledger_entry(action="read")
                 return df
             elif file_ext == "ndjson" or file_ext == "jsonl":
-                path = str(PurePosixPath(name).parent / f"*.{file_ext}")
-                fullpath = f"az://{self.container}/{path}"
                 df = pl.scan_ndjson(
                     fullpath,
                     infer_schema_length=None,
@@ -615,6 +614,50 @@ def ledger_entry(self, action: str) -> None:
             overwrite=False,
         )
 
+    def resolve_version(
+        self,
+        version_spec: str | None = None,
+        selection: Literal["newest", "oldest"] = "newest",
+    ) -> VersionMetadata:
+        """Resolve the version of the dataset based on the version specification and selection criteria.
+
+        Args:
+            version_spec (str | None): the version specification to resolve
+            selection (Literal["newest", "oldest"]): whether to select the newest or oldest version
+
+        Returns:
+            VersionMetadata: Resolution containing resolved version, blob URL, and selection details.
+        """
+        try:
+            version_blobs, version = self._get_version_blobs(
+                version_spec=version_spec, selection=selection, print_version=False
+            )
+        except ValueError:
+            return VersionMetadata(
+                version=None,
+                blob_url=None,
+                version_spec=version_spec,
+                selection=selection,
+            )
+
+        if not version_blobs:
+            return VersionMetadata(
+                version=None,
+                blob_url=None,
+                version_spec=version_spec,
+                selection=selection,
+            )
+        name = version_blobs[0]["name"]
+        file_ext = PurePosixPath(name).suffix.lstrip(".").lower()
+        path = str(PurePosixPath(name).parent / f"*.{file_ext}")
+        fullpath = f"az://{self.container}/{path}"
+        return VersionMetadata(
+            version=version,
+            blob_url=fullpath,
+            version_spec=version_spec,
+            selection=selection,
+        )
+
     def save_dataframe(
         self,
         df: pd.DataFrame | pl.DataFrame,

cfa/dataops/command.py

@@ -133,12 +133,6 @@ def save_data_locally():
         help="download the oldest version of data instead of the newest",
         action="store_true",
     )
-    parser.add_argument(
-        "--full_range",
-        "-r",
-        help="download the full range of data versions that meet the version criteria",
-        action="store_true",
-    )
     args = parser.parse_args()
     dataset = args.dataset
     stage = args.stage
@@ -152,8 +146,6 @@ def save_data_locally():
     local_path = os.path.abspath(args.location)
     if args.oldest:
         selection = "oldest"
-    elif args.full_range:
-        selection = "all"
     else:
         selection = "newest"
     written = eval(

cfa/dataops/stub_templates/catalog.pyi.mako

@@ -36,22 +36,22 @@ class BlobEndpoint:
     def read_blobs(
         self,
         version_spec: str | None = None,
-        selection: Literal["newest", "oldest", "all"] = "newest",
+        selection: Literal["newest", "oldest"] = "newest",
         print_version: bool = True,
     ) -> list[bytes]: ...
     def read_csv(self, suffix: str) -> pd.DataFrame: ...
     def get_versions(self) -> list[str]: ...
     def get_file_ext(
         self,
         version_spec: str | None = None,
-        selection: Literal["newest", "oldest", "all"] = "newest",
+        selection: Literal["newest", "oldest"] = "newest",
     ) -> str: ...
     def download_version_to_local(
         self,
         local_path: str,
         version_spec: str | None = None,
         force: bool = False,
-        selection: Literal["newest", "oldest", "all"] = "newest",
+        selection: Literal["newest", "oldest"] = "newest",
     ) -> bool: ...
     @overload
     def get_dataframe(

cfa/dataops/utils.py

@@ -215,30 +215,31 @@ def construct_version_spec(version: str | None) -> str | None:
 def version_matcher(
     version_spec: str | None,
     available_versions: list[str],
-    selection: Literal["newest", "oldest", "all"] = "newest",
-) -> str | list[str] | None:
+    selection: Literal["newest", "oldest"] = "newest",
+) -> str | None:
     """Select version strings from a list using an optional specifier.
 
     Args:
         version_spec (str | None): Optional packaging-compatible version specifier such as
             ``"==2025-12-15"`` or ``">=2025-01-01,<2026-01-01"``. If ``None``,
             all available versions are considered.
         available_versions (list[str]): Version strings to evaluate.
-        selection (Literal["newest", "oldest", "all"]): Controls the return shape.
-            ``"newest"`` returns the newest matching version, ``"oldest"`` returns
-            the oldest matching version, and ``"all"`` returns all matching versions
-            in descending order.
+        selection (Literal["newest", "oldest"]): Controls which matching
+            version is returned.
 
     Returns:
-        str | list[str] | None: A single matching version, all matching versions in
-        descending order, or ``None`` if no match is found.
+        str | None: A single matching version, or ``None`` if no match is found.
 
     Raises:
+        ValueError: If ``selection`` is not one of ``"newest"`` or ``"oldest"``.
         packaging.specifiers.InvalidSpecifier: If ``version_spec`` is not ``None`` and is not
             a valid packaging specifier after normalization.
         packaging.version.InvalidVersion: If any version string cannot be parsed by
             ``packaging.version.Version``.
     """
+    if selection not in {"newest", "oldest"}:
+        raise ValueError("selection must be 'newest' or 'oldest'")
+
     versions = sorted(
         (Version(normalize(version)), version) for version in available_versions
     )
@@ -251,10 +252,6 @@ def version_matcher(
 
     originals = [original for _, original in versions]
 
-    match selection:
-        case "newest":
-            return originals[-1] if originals else None
-        case "oldest":
-            return originals[0] if originals else None
-        case "all":
-            return originals[::-1]
+    if selection == "newest":
+        return originals[-1] if originals else None
+    return originals[0] if originals else None