etl: search recursively for input ndjson files

Author: mikix

cumulus_etl/common.py

@@ -78,7 +78,9 @@ def get_temp_dir(subdir: str) -> str:
 
 
 def ls_resources(root: store.Root, resources: set[str], warn_if_empty: bool = False) -> list[str]:
-    found_files = cfs.list_multiline_json_in_dir(root.path, resources, fsspec_fs=root.fs)
+    found_files = cfs.list_multiline_json_in_dir(
+        root.path, resources, fsspec_fs=root.fs, recursive=True
+    )
 
     if warn_if_empty:
         # Invert the {path: type} found_files dictionary into {type: [paths...]}

cumulus_etl/inliner/inliner.py

@@ -55,7 +55,7 @@ async def inliner(
     mimetypes = set(mimetypes)
 
     # Grab files to read for the given resources
-    found_files = cfs.list_multiline_json_in_dir(in_root.path, resources, fsspec_fs=in_root.fs)
+    found_files = common.ls_resources(in_root, resources)
 
     # Predict how much work we'll have to do by getting counts of lines and files
     if in_root.protocol == "file":

cumulus_etl/loaders/fhir/ndjson_loader.py

@@ -56,7 +56,9 @@ async def detect_resources(self) -> set[str] | None:
             # Returning None means "dunno" (i.e. "just accept whatever you eventually get").
             return None
 
-        found_files = cfs.list_multiline_json_in_dir(self.root.path, fsspec_fs=self.root.fs)
+        found_files = cfs.list_multiline_json_in_dir(
+            self.root.path, fsspec_fs=self.root.fs, recursive=True
+        )
         return {resource for resource in found_files.values() if resource}
 
     async def load_resources(self, resources: set[str]) -> base.LoaderResults:

pyproject.toml

@@ -4,7 +4,7 @@ requires-python = ">= 3.11"
 dependencies = [
     "aiobotocore < 2.22.0",  # FIXME: temp hotfix for dependency version madness - remove later
     "ctakesclient >= 5.1",
-    "cumulus-fhir-support >= 1.5",
+    "cumulus-fhir-support >= 1.6",
     "delta-spark >= 4, < 5",
     "fsspec[http,s3]",
     "httpx",

tests/etl/test_etl_cli.py

@@ -381,6 +381,21 @@ async def test_rejects_new_phi_dir(self):
             await self.run_etl(tasks=["patient"], phi_path=self.make_tempdir())
         self.assertEqual(cm.exception.code, errors.WRONG_PHI_FOLDER)
 
+    async def test_looks_recursively_for_input(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            os.makedirs(f"{tmpdir}/subdir1/subdir2")
+            with open(f"{tmpdir}/subdir1/subdir2/patients.ndjson", "w", encoding="utf8") as f:
+                json.dump({"resourceType": "Patient", "id": "pat1"}, f)
+            await self.run_etl(tasks=["patient"], input_path=tmpdir)
+
+        self.assertEqual(
+            common.read_json(f"{self.output_path}/patient/patient.000.ndjson"),
+            {
+                "resourceType": "Patient",
+                "id": "50ffe70a1bdf3b6e73adac15e4ab7f9d7e247466d7a6c395c2ae9098741a62bd",
+            },
+        )
+
 
 class TestEtlJobConfig(BaseEtlSimple):
     """Test case for the job config logging data"""