issue #41 - data version fuzzy matching

README.md

@@ -34,9 +34,9 @@ This project provides data tools and low friction access to versioned datasets w
    df.glimpse()
    ```
 
-Read the [Dataset User Guide](docs/dataset_user.md) for more information about accessing datasets.
+Read the [Dataset User Guide](docs/data_user_guide.md) for more information about accessing datasets.
 
-Read the [Dataset Developer Guide](docs/dataset_developer.md) for information about how to run ETL to add new versions of an existing datasets and about how to create new datasets.
+Read the [Dataset Developer Guide](docs/data_developer_guide.md) for information about how to run ETL to add new versions of an existing datasets and about how to create new datasets.
 
 ## Project admins
 

cfa/dataops/__init__.py

@@ -5,9 +5,13 @@
 will run on all configurations.
 """
 
-from importlib.metadata import version
+try:
+    from importlib.metadata import version
+
+    __version__ = version(__name__)
+except ImportError:
+    __version__ = "unknown"
 
 from .catalog import datacat, reportcat
 
-__version__ = version(__name__)
 __all__ = [__version__, datacat, reportcat]

cfa/dataops/catalog.py

@@ -27,7 +27,13 @@
     ValidationError,
 )
 from .reporting.catalog import report_dict_to_sn
-from .utils import get_dataset_dot_path, get_date, get_timestamp, get_user
+from .utils import (
+    get_dataset_dot_path,
+    get_date,
+    get_timestamp,
+    get_user,
+    version_matcher,
+)
 
 _here = os.path.abspath(os.path.dirname(__file__))
 _config = ConfigParser()
@@ -222,14 +228,19 @@ def write_blob(
         self.ledger_entry(action="write")
         # print(f"file written to: {full_path}")
 
-    def read_blobs(self, version: str = "latest") -> List[bytes]:
+    def read_blobs(
+        self, version: str = "latest", newest: bool = True
+    ) -> List[bytes]:
         """Read a blob in as bytes so it can be loaded into a dataframe
 
         Args:
             path_after_prefix (str): The path to the data (e.g.,
             {timestamp}/dataset.csv)
+            version (str, optional): the version of the data to read.
+                Defaults to "latest".
+            newest (bool, optional): whether to get the newest matching version. Defaults to True.
         """
-        blobs = self._get_version_blobs(version)
+        blobs = self._get_version_blobs(version, newest=newest)
         blob_bytes = [
             read_blob_stream(
                 blob_url=i["name"],
@@ -279,41 +290,75 @@ def get_file_ext(self) -> str:
         Returns:
             str: the file extension
         """
-        return self._get_version_blobs()[0]["name"].split(".")[-1]
+        return self._get_version_blobs(print_version=False)[0]["name"].split(
+            "."
+        )[-1]
 
-    def _get_version_blobs(self, version: str = "latest") -> list:
-        if version == "latest" and not self.is_ledger:
-            version = self.get_versions()[0]
+    def _get_version_blobs(
+        self, version: str = "latest", newest=True, print_version=True
+    ) -> list:
         if not self.is_ledger:
-            version = version.removesuffix("/")
-            walk_path = f"{self.prefix}/{version}/"
+            available_versions = self.get_versions()
+            version = version_matcher(
+                version, available_versions, newest=newest
+            )
+            if not version:
+                raise ValueError(
+                    f"Version {version} not found in available versions: {available_versions}"
+                )
+            if print_version:
+                print(f"Using version(s): {version}")
+            if isinstance(version, list):
+                walk_path = [f"{self.prefix}/{v}/" for v in version]
+            else:
+                walk_path = f"{self.prefix}/{version}/"
         else:
             walk_path = f"{self.prefix.removesuffix('/')}/"
-        return sorted(
-            list(
-                walk_blobs_in_container(
-                    name_starts_with=walk_path,
-                    account_name=self.account,
-                    container_name=self.container,
+        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,
+                    )
                 )
-            ),
-            key=lambda x: x["creation_time"],
-        )
+            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"],
+            )
 
     def download_version_to_local(
-        self, local_path: str, version: str = "latest", force: bool = False
+        self,
+        local_path: str,
+        version: str = "latest",
+        force: bool = False,
+        newest: bool = True,
     ) -> bool:
         """Download a specific version of the data to a local path
 
         Args:
             local_path (str): the local path to download to
             version (str, optional): the version to download. Defaults to "latest".
             force (bool, optional): whether to force re-download if local.
+            newest (bool, optional): whether to get the newest matching version. Defaults to True.
         Returns:
             bool: whether any files were written
         """
         written = False
-        blobs = self._get_version_blobs(version)
+        blobs = self._get_version_blobs(version, newest=newest)
         for blob in blobs:
             blob_data = read_blob_stream(
                 blob_url=blob["name"],
@@ -339,7 +384,11 @@ def download_version_to_local(
         return written
 
     def get_dataframe(
-        self, output="pandas", version="latest", pl_lazy: bool = False
+        self,
+        output="pandas",
+        version="latest",
+        pl_lazy: bool = False,
+        newest: bool = True,
     ) -> pd.DataFrame | pl.DataFrame:
         """Get the data as a pandas or polars dataframe
 
@@ -350,6 +399,8 @@ def get_dataframe(
                 Defaults to "latest".
             pl_lazy (bool, optional): whether to return a lazy polars dataframe.
                 Defaults to False.
+            newest (bool, optional): whether to get the newest matching version. Defaults to True.
+                False returns the oldest matching version.
 
         Raises:
             ValueError: if output is not 'pandas' or 'polars'
@@ -361,7 +412,7 @@ def get_dataframe(
             raise ValueError(
                 f"Output {output} needs to be 'pandas', 'polars', 'pd, or 'pl'."
             )
-        blobs = self.read_blobs(version)
+        blobs = self.read_blobs(version, newest=newest)
         file_ext = self.get_file_ext()
         blob_bytes = [blob.content_as_bytes() for blob in blobs]
         blob_files = [BytesIO(pq) for pq in blob_bytes]

cfa/dataops/command.py

@@ -131,6 +131,18 @@ def save_data_locally():
     parser.add_argument(
         "--force", "-f", help="force re-download of data", action="store_true"
     )
+    parser.add_argument(
+        "--oldest",
+        "-o",
+        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
@@ -142,8 +154,14 @@ def save_data_locally():
         versions = _get_versions_list(dataset, stage)
         version = versions[0]
     local_path = os.path.abspath(args.location)
+    if args.oldest:
+        newest = False
+    elif args.full_range:
+        newest = None
+    else:
+        newest = True
     written = eval(
-        f"datacat.{dataset}.{stage}.download_version_to_local('{local_path}', version='{version}', force={args.force})"
+        f"datacat.{dataset}.{stage}.download_version_to_local('{local_path}', version='{version}', force={args.force}, newest={newest})"
     )
     if not written:
         Console().print(

cfa/dataops/utils.py

@@ -3,6 +3,7 @@
 import getpass
 import glob
 import os
+import re
 from datetime import datetime
 from itertools import islice
 from pathlib import Path
@@ -189,3 +190,82 @@ def inner(dir_path: Path, prefix: str = "", level=-1):
         + f"\n{directories} directories"
         + (f", {files} files" if files else "")
     )
+
+
+def version_matcher(
+    version: str,
+    available_versions: list[str],
+    newest: Optional[bool] = True,
+    and_sep=",",
+) -> str | list[str]:
+    """Match a version string to the closest available version.
+
+    Args:
+        version (str): The version string to match (e.g., '1.2').
+        available_versions (list[str]): A list of available version strings (e.g., ['1.0', '1.1', '1.2', '2.0']).
+        newest (Optional[bool]): Whether to return the newest matching version, returns oldest if False (default is True).
+        and_sep (str): The separator for multiple version conditions (default is ',').
+
+    Returns:
+        list[str]: The matched version string if found, otherwise or empty list if non found.
+
+    Example:
+        >>> available_versions = ['1.0', '1.1', '1.2', '2.0']
+        >>> version_matcher('>=1.1,<2.0', available_versions)
+        '1.2'
+        >>> version_matcher('>=1.1,<2.0', available_versions, newest=False)
+        '1.1'
+        >>> version_matcher('latest', available_versions)
+        '2.0'
+        >>> version_matcher('~=1', available_versions)
+        '1.2'
+        >>> version_matcher('>=1.1,<2.0', available_versions, newest=None)
+        ['1.2', '1.1']
+    """
+    if version == "latest":
+        return sorted(available_versions, reverse=True)[0]
+    version = re.sub(r"\s", "", version)
+    v_ands_parsed = []
+    v_ands = version.split(and_sep)
+    for v in v_ands:
+        cond = re.match(r"[\>\<\=\~\!]+", v)
+        if cond and cond.span(0)[0] == 0:
+            v_cond = cond.group(0)
+        else:
+            v_cond = "=="
+        v_parts = re.findall(r"\d+", v)
+        v_ands_parsed.append((v_cond, ".".join(v_parts)))
+    av_parsed = {}
+    for avail_version in sorted(available_versions, reverse=True):
+        avail_parts = re.findall(r"\d+", avail_version)
+        av_parsed[avail_version] = ".".join(avail_parts)
+    v_match = []
+    logic_vals = []
+    for idx, (v_cond, v_p) in enumerate(v_ands_parsed):
+        logic_vals.append([])
+        for av, av_p in av_parsed.items():
+            if v_cond == "==" and v_p == av_p:
+                logic_vals[idx].append(True)
+            elif v_cond in [">=", "=>"] and v_p <= av_p:
+                logic_vals[idx].append(True)
+            elif v_cond in ["<=", "=<"] and v_p >= av_p:
+                logic_vals[idx].append(True)
+            elif v_cond == ">" and v_p < av_p:
+                logic_vals[idx].append(True)
+            elif v_cond == "<" and v_p > av_p:
+                logic_vals[idx].append(True)
+            elif v_cond == "!=" and v_p != av_p:
+                logic_vals[idx].append(True)
+            elif v_cond == "~=" and v_p == av_p[: len(v_p)]:
+                logic_vals[idx].append(True)
+            else:
+                logic_vals[idx].append(False)
+            v_match.append(av)
+    keep = [all(i) for i in zip(*logic_vals)]
+    if isinstance(newest, bool):
+        if newest:
+            return max([i for i, j in zip(v_match, keep) if j])
+        else:
+            return min([i for i, j in zip(v_match, keep) if j])
+    else:
+        return [i for i, j in zip(v_match, keep) if j]

changelog.md

@@ -8,6 +8,36 @@ The versioning pattern is `YYYY.MM.DD.micro(a/b/{none if release})
 
 ---
 
+## [2025.11.13.0a]
+
+### Added
+
+- `cfa.dataops.utils.py:version_matcher` utility function and testing via doctests
+
+### Updated
+
+- Data versions (when retrieved/`get_`) can now use conditional logic:
+  - Examples:
+    ```python
+    >>> from cfa.dataops import datacat
+    >>> datacat.public.my_dataset.load.get_versions()
+    ['2025-10-31',
+     '2025-09-19',
+     '2025-06-01',
+     '2024-12-08',
+     '2024-11-21']
+    >>> df = datacat.public.my_dataset.load.get_dataframe(version=">2024.12.01,<2025.08")
+    Using version: 2025-06-01
+    >>> df = datacat.public.my_dataset.load.get_dataframe(version=">2024-12.01,<2025.08", newest=False)
+    Using version: 2024-12-08
+    >>> df = datacat.public.my_dataset.load.get_dataframe(version="~=2024/11")
+    Using version: 2024-11-21
+    >>> df = datacat.public.my_dataset.load.get_dataframe(version="latest")
+    Using version: 2025-10-31
+    ```
+- Links in `README.md`
+- pytest adopts in `pyproject.toml` to include `cfa/dataops/utils.py` for doctests
+
 ## [2025.10.31.0a]
 
 ### Added

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "cfa.dataops"
-version = "2025.10.31.0a"
+version = "2025.11.13.0a"
 description = "Data cataloging, ETL, modeling, verification, and validation for CFA"
 authors = [
     { name = "Phil Rogers", email = "ap66@cdc.gov" },
@@ -67,4 +67,4 @@ dataops_save = "cfa.dataops.command:save_data_locally"
 
 
 [tool.pytest.ini_options]
-addopts = "tests --doctest-modules -vv --cov=cfa --cov-report html --cov-report term"
+addopts = "tests cfa/dataops/utils.py --doctest-modules -vv --cov=cfa --cov-report html --cov-report term"