Ap66-add_cfasodpy_and_respnet_etl (#58)

* refactoring namespace for multiple groups

* changes for namespace refactor

* updates to fix tests after refactor

* adding and updating soda code

* respnet wip

* added json raw reading

* including get_data in easy importing path

* new respnet etl

* all items complete

* removed personal app token

* version bump

* Update coverage badge

* casting weekly_rate to float

* Update coverage badge

* left out one important change worth mentioning

* Update coverage badge

* forgot to add this one __init__.py

* Update coverage badge

* updated docs to reflect changes in refactor

* Update coverage badge

---------

Co-authored-by: cdc-ap66 <cdc-ap66@users.noreply.github.com>

Author: cdc-ap66

README.md

@@ -33,9 +33,9 @@ To use this repository:
    ```
 
 To add a new dataset:
-1. Create a new TOML configuration file in `cfa/scenarios/dataops/datasets/`
-2. Create a new ETL script in `cfa/scenarios/dataops/etl/`
-3. Add SQL transformation templates in `cfa/scenarios/dataops/etl/transform_templates/`
+1. Create a new TOML configuration file in `cfa/scenarios/dataops/datasets/{team_dir}/`
+2. Create a new ETL script in `cfa/scenarios/dataops/etl/{team_dir}/`
+3. Add SQL transformation templates in `cfa/scenarios/dataops/etl/transform_templates/{team_dir}/` (is using SQL for transforms). These are [Mako templates](https://www.makotemplates.org/)
 
 ## Accessing Datasets
 
@@ -45,13 +45,13 @@ When the ETL pipelines are run, the data sources (raw and/or transformed) are st
 - type: either 'raw' or 'transformed'. Default is 'transformed'.
 - output: the type of dataframe to output, either 'pandas' or 'polars'. Default is 'pandas'.
 
-The available datasets can be found by running `list_datasets()`, which can be found in the `cfa.scenarios.dataops.datasets.catalog` submodule.
+The available datasets can be found by running `list_datasets()`, which can be found in the `cfa.scenarios.dataops.catalog` submodule.
 
 An example for getting the polars dataframes for the latest raw versions of the covid19vax_trends and seroprevalence datasets is below:
 ```python
-from cfa.scenarios.dataops.datasets.catalog import get_data
-vax_df = get_data("covid19vax_trends", type = "transformed", output = "polars")
-sero_df = get_data("seroprevalence", type = "transformed", output = "polars")
+from cfa.scenarios.dataops import get_data
+vax_df = get_data("scenarios.covid19vax_trends", type = "transformed", output = "polars")
+sero_df = get_data("scenarios.seroprevalence", type = "transformed", output = "polars")
 ```
 
 ## Running Workflows

cfa/scenarios/dataops/__init__.py

@@ -0,0 +1,10 @@
+"""Simplified importing of most commonly used objects
+
+All data configurations should reside in this directory and be in the toml
+format. Use the existing configuration files as a starting point. Validations
+will run on all configurations.
+"""
+
+from .catalog import datacat, get_data
+
+__all__ = [datacat, get_data]

cfa/scenarios/dataops/datasets/catalog.py cfa/scenarios/dataops/catalog.pycfa/scenarios/dataops/datasets/catalog.py renamed to cfa/scenarios/dataops/catalog.py

@@ -18,17 +18,27 @@
 from .config_validator import validate_dataset_config, verify_no_repeats
 
 _here_dir = os.path.split(os.path.abspath(__file__))[0]
-_dataset_config_paths = glob.glob(os.path.join(_here_dir, "*.toml"))
+_dataset_config_paths = glob.glob(
+    os.path.join(_here_dir, "datasets", "**", "*.toml")
+)
 
-dataset_configs = []
+name_paths = []
+dataset_configs = {}
 for cp_i in _dataset_config_paths:
+    ns_pre = cp_i.split(f"datasets{os.sep}")[-1].split(os.sep)[:-1]
     with open(cp_i, "rb") as f:
         config = tomli.load(f)
         config["_metadata"] = dict(filename=os.path.split(cp_i)[1])
         validate_dataset_config(config)
-        dataset_configs.append(config)
+        current = dataset_configs
+        for part in ns_pre:
+            if part not in current:
+                current[part] = {}
+            current = current[part]
+        current[config["properties"]["name"]] = config
+        name_paths.append(".".join(ns_pre + [config["properties"]["name"]]))
 
-verify_no_repeats(dataset_configs)
+verify_no_repeats(name_paths)
 
 
 class BlobEndpoint:
@@ -169,10 +179,7 @@ def dict_to_sn(d: Any) -> SimpleNamespace:
     return x
 
 
-datasets = SimpleNamespace()
-
-for i in dataset_configs:
-    setattr(datasets, i["properties"]["name"], dict_to_sn(i))
+datacat = dict_to_sn(dataset_configs)
 
 
 def get_data(
@@ -194,11 +201,13 @@ def get_data(
         pd.DataFrame | pl.DataFrame: pandas or polars dataframe
     """
     # check data exists
-    available_datasets = [x for x in datasets.__dict__.keys()]
-    if name not in available_datasets:
-        print(f"{name} not in available datasets.")
-        print("Available datasets:", available_datasets)
-        raise ValueError(f"{name} not in available datasets.")
+    try:
+        config = eval(f"datacat.{name}")
+    except AttributeError as e:
+        raise ValueError(
+            f"{name} not in available datasets."
+            f" Available datasets: {list_datasets()}"
+        ) from e
 
     # validate type, raise error if not raw or transformed
     if type not in ["raw", "transformed"]:
@@ -213,7 +222,7 @@ def get_data(
     # continue workflow based on raw or transformed
     if type == "raw":
         # get the BlobEndpoint for the raw data
-        blob_endpoint = datasets.__dict__[name].extract
+        blob_endpoint = config.extract
         # check version exists, raise error if not
         if version != "latest":
             v_list = blob_endpoint.get_versions()
@@ -225,17 +234,29 @@ def get_data(
                     f"Version {version} not in available versions."
                 )
         # get blobs and convert to correct df
-        blobs = blob_endpoint.read_blobs()
-        if output in ["pandas", "pd"]:
-            df = pd.concat([pd.read_csv(blob) for blob in blobs])
-        else:
-            df = pl.concat(
-                [pl.read_csv(blob.content_as_bytes()) for blob in blobs],
-                how="vertical_relaxed",
-            )
-        return df
+        if config.extract.get_file_ext() == "csv":
+            blobs = blob_endpoint.read_blobs()
+            if output in ["pandas", "pd"]:
+                df = pd.concat([pd.read_csv(blob) for blob in blobs])
+            else:
+                df = pl.concat(
+                    [pl.read_csv(blob.content_as_bytes()) for blob in blobs],
+                    how="vertical_relaxed",
+                )
+            return df
+        elif config.extract.get_file_ext() == "json":
+            blobs = blob_endpoint.read_blobs()
+            if output in ["pandas", "pd"]:
+                df = pd.concat(
+                    [pd.read_json(blob.content_as_bytes()) for blob in blobs]
+                )
+            else:
+                df = pl.concat(
+                    [pl.read_json(blob.content_as_bytes()) for blob in blobs],
+                )
+            return df
     else:
-        blob_endpoint = datasets.__dict__[name].load
+        blob_endpoint = config.load
         # check version exists, raise error if not
         if version != "latest":
             v_list = blob_endpoint.get_versions()
@@ -268,8 +289,8 @@ def list_datasets() -> list[str]:
 
     Examples:
         >>> datasets = list_datasets()
-        >>> 'covid19vax_trends' in datasets
+        >>> 'scenarios.covid19vax_trends' in datasets
         True
 
     """
-    return [x for x in datasets.__dict__.keys()]
+    return name_paths

…ios/dataops/datasets/config_validator.py cfa/scenarios/dataops/config_validator.pycfa/scenarios/dataops/datasets/config_validator.py renamed to cfa/scenarios/dataops/config_validator.py cfa/scenarios/dataops/datasets/config_validator.py renamed to cfa/scenarios/dataops/config_validator.py

@@ -64,18 +64,10 @@ def verify_no_repeats(configs: List[dict]) -> None:
     Raises:
         AttributeError: When the names repeat.
     """
-    name_counts = Counter([i["properties"]["name"] for i in configs])
+    name_counts = Counter(configs)
     repeats = [k for k, v in name_counts.items() if v > 1]
     if repeats:
-        names_in_files = []
-        for n_i in set(repeats):
-            names_in_files += [
-                f'  - {n_i} in {config["_metadata"]["filename"]}'
-                for config in configs
-                if config["properties"]["name"] == n_i
-            ]
-        error_file_strings = ("\n").join(names_in_files)
         raise AttributeError(
-            "More than one config shares a the same name property: "
-            f"{error_file_strings}"
+            "More than one config shares a the same name. Here are the repeats: "
+            f"{', '.join(repeats)}"
         )

cfa/scenarios/dataops/datasets/__init__.py

@@ -1,10 +0,0 @@
-"""Simplified importing of most commonly used objects
-
-All data configurations should reside in this directory and be in the toml
-format. Use the existing configuration files as a starting point. Validations
-will run on all configurations.
-"""
-
-from .catalog import datasets
-
-__all__ = [datasets]

cfa/scenarios/dataops/datasets/mcmv/__init__.py

@@ -0,0 +1,32 @@
+# This dataset is static, and thus only needs to be extracted in it's raw form
+# once and saved to blob storage. For a human viewable we format, see:
+# https://data.cdc.gov/Vaccinations/COVID-19-Vaccination-Age-and-Sex-Trends-in-the-Uni/5i5k-6cmh/data_preview
+# Raw (extracted) and transformed (loaded) data will be stored in azure blob storage
+# which requires the account, container and path for access
+
+[properties]
+
+name = "respnet"
+automate = false
+transform_template = ""
+schema = ""
+
+[source]
+
+domain = "data.cdc.gov"
+id = "kvib-3txy"
+limit = 10000
+
+[extract]
+
+account = "cfadatalakeprd"
+container = "cfapredict"
+prefix = "dataops/mcmv/raw/respnet"
+
+[load]
+
+account = "cfadatalakeprd"
+container = "cfapredict"
+prefix = "dataops/mcmv/transformed/respnet"
+
+# TODO: add some data schema validation

cfa/scenarios/dataops/datasets/mcmv/respnet.toml

@@ -7,8 +7,8 @@
 [properties]
 name = "covid19hospitalizations"
 automate = false
-transform_template = "covid19hospitalizations.sql"
-schema = "datasets/schemas/covid19hospitalizations.py"
+transform_template = "scenarios/covid19hospitalizations.sql"
+schema = "datasets/scenarios/schemas/covid19hospitalizations.py"
 
 [source]
 url = "https://healthdata.gov/resource/sgxm-t72h.csv"

cfa/scenarios/dataops/datasets/scenarios/__init__.py

@@ -8,8 +8,8 @@
 
 name = "covid19vax_trends"
 automate = false
-transform_template = "covid19vax_trends.sql"
-schema = "datasets/schemas/covid19vax_trends.py"
+transform_template = "scenarios/covid19vax_trends.sql"
+schema = "datasets/scenarios/schemas/covid19vax_trends.py"
 
 [source]