diff --git a/README.md b/README.md index 36d4b8c..fe93dcb 100644 --- a/README.md +++ b/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 diff --git a/cfa/dataops/__init__.py b/cfa/dataops/__init__.py index 969c37a..ebcde05 100644 --- a/cfa/dataops/__init__.py +++ b/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] diff --git a/cfa/dataops/catalog.py b/cfa/dataops/catalog.py index 29431bf..538ccf1 100644 --- a/cfa/dataops/catalog.py +++ b/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,29 +290,62 @@ 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 @@ -309,11 +353,12 @@ def download_version_to_local( 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] diff --git a/cfa/dataops/command.py b/cfa/dataops/command.py index 4b01046..40b0090 100644 --- a/cfa/dataops/command.py +++ b/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( diff --git a/cfa/dataops/utils.py b/cfa/dataops/utils.py index 6a4aa20..af7166c 100644 --- a/cfa/dataops/utils.py +++ b/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] diff --git a/changelog.md b/changelog.md index d360438..97198ad 100644 --- a/changelog.md +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index 25a2383..8a4bc65 100644 --- a/pyproject.toml +++ b/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"