Bulk FHIR Gen 3 Authorization Tagging

.github/workflows/ci.yaml

@@ -20,7 +20,7 @@ jobs:
       matrix:
         # the versions for which we require the CI to pass on PRs can
         # be updated in the repo settings (branch rules)
-        version: ['3.9', '3.10', '3.11', '3.12', '3.13']
+        version: ['3.10', '3.11', '3.12', '3.13']
       fail-fast: false  # run for all versions even if one of them fails
     with:
         python-version: ${{ matrix.version }}

.gitignore

@@ -109,6 +109,7 @@ schema.json
 
 # pytest outputs
 tests/outputs
+tests/test_data/fhir_outputs
 *.sq3
 gen3/tools/indexing/tmp/
 

docs/howto/devTest.md

@@ -1,25 +1,11 @@
 ## Dev-Test
 
-### Set up Python Virtual Environment
-
-You can set up a Python development environment with a virtual environment:
-
-```bash
-python3 -m venv py3
-```
-
-Make sure that you have the virtual environment activated:
-
-```bash
-. py3/bin/activate
-```
-
 ### Install poetry
 
 To use the latest code in this repo (or to develop new features) you can clone this repo, install `poetry`:
 
 ```
-curl -sSL https://raw.githubusercontent.com/python-poetry/poetry/master/get-poetry.py | python -
+curl -sSL https://install.python-poetry.org | python3 -
 ```
 
 and then use `poetry` to install this package:
@@ -34,8 +20,7 @@ poetry install -vv
 Local development like this:
 
 ```
-poetry shell
-poetry install -vv
+eval $(poetry env activate)
 python3 -m pytest
 ```
 

docs/howto/fhir.md

@@ -0,0 +1,48 @@
+## FHIR 
+
+This integration aims to enhance the data ingestion capabilities of Gen3 by integrating a Fast Healthcare Interoperability Resources (FHIR) data ingestion tools. FHIR is an important standard for working with Electronic Health Records (EHR) and we have started development of a Gen3 FHIR Proxy service. 
+
+Gen3 is working on adding support for FHIR and these tools will help with data preparation.
+
+
+The fhir commands can be invoked as follows
+
+`gen3 fhir COMMAND [ARGS] [OPTIONS]`
+
+For a list of commands and options run
+
+`gen3 fhir --help`
+
+For example, the following tags the 'Patient.ndjson' file with Gen3 authorization and outputs 'gen3_Patient.ndjson' using the authorization rules from 'config.yaml'. 
+
+`gen3 fhir transform Patient.ndjson gen3_Patient.ndjson config.yaml --batch_size 10000`
+
+INPUT_FILE:
+There can only be one resource type per NDJSON file. 
+
+OUTPUT_FILE:
+Requires a distinct name/path from the input file to prevent overwriting the input file.
+
+CONFIG.YAML:
+The authorization configuration file has to be in yaml format and can have multiple conditions, as well as a `global_authz` which overrides all other rules. Conditions use [https://hl7.org/fhirpath/](FHIRPath) syntax.
+
+A resource matching two rules will throw an error and will require reconfiguration of the YAML file. 
+
+Example rules:
+
+```yaml
+rules:
+ - resource_type: "Patient"
+   condition: "Patient.managingOrganization.reference = 'Organization/site-alpha'"
+   authz: "/programs/Alpha/projects/Main"
+
+#Example with multiple conditions
+ - resource_type: "Specimen"
+   condition: "Specimen.status = 'available' and Specimen.Type = 'Blood specimen (specimen)'"
+   authz: "/programs/Alpha/projects/Biobank"
+```
+
+And example config.yaml file can be found in [fhir_config.yaml](../../tests/test_data/fhir_config.yaml)
+To run the fhir transfrom with synthetic test data:
+
+`poetry run gen3 -vv fhir transform ./tests/test_data/Patient.ndjson ./test_Patient.ndjson ./tests/test_data/fhir_config.yaml`

gen3/cli/__main__.py

@@ -33,7 +33,7 @@ def get(self):
         return self._cache
 
 
-@click.group()
+@click.group(epilog="FHIR commands require extras. See: poetry install --all-extras")
 @click.option(
     "--auth",
     "auth_config",
@@ -145,4 +145,27 @@ def main(
 main.add_command(nih.nih)
 main.add_command(users.users)
 main.add_command(wrap.run)
+
+# optional fhir subcommand dependent on whether user installed FHIR extras
+try:
+    import gen3.cli.fhir as fhir
+
+    main.add_command(fhir.fhir)
+except ImportError:
+
+    @click.group(
+        name="fhir",
+        epilog="Requires FHIR packages which aren't installed by default. Install the 'fhir' extras: poetry install --all-extras",
+    )
+
+    @click.argument("args", nargs=-1, type=click.UNPROCESSED)
+    def fhir(args):
+        """Commands for FHIR data processing: transform & cleanup"""
+        raise click.ClickException(
+            "FHIR commands require the 'fhir' extras: poetry install --all-extras"
+        )
+
+    main.add_command(fhir)
+
+
 main()

gen3/cli/fhir.py

@@ -0,0 +1,122 @@
+import click
+import os
+import pathlib
+from cdislogging import get_logger
+from gen3.fhir import *
+
+logging = get_logger(__name__)
+
+
+@click.group()
+def fhir():
+    """Commands for FHIR data processing"""
+    pass
+
+
+@click.command(
+    context_settings={"help_option_names": ["-h", "--help"]},
+    help="""Tag Bulk FHIR data with Gen3 compatible authorization tags.
+
+    \b 
+    input_file (str): Input .ndjson file with Bulk FHIR data, MUST be one resource type per file
+    output_file (str): Output file name, also an .ndjson file
+    config (str): .yaml file with authorization rules, see docs/howto/fhir.md for more details on formatting
+    """,
+)
+@click.argument(
+    "input_file",
+    type=click.Path(exists=True, dir_okay=False, readable=True),
+    metavar="input_file",
+)
+@click.argument(
+    "output_file", type=click.Path(dir_okay=False, writable=True), metavar="output_file"
+)
+@click.argument(
+    "config",
+    type=click.Path(exists=True, dir_okay=False, readable=True),
+    metavar="config",
+)
+@click.option(
+    "--work_dir",
+    type=click.Path(),
+    metavar="work_dir",
+    help=f"Specify which working directory to clean, if not specified the default ({DEFAULT_WORK_DIR}) will be used. Can also be set as an environment variable: GEN3_FHIR_WORK_DIR",
+)
+@click.option(
+    "-b",
+    "--batch_size",
+    type=click.IntRange(min=1),
+    default=10000,
+    show_default=True,
+    metavar="batch_size",
+    help="batch size for chunking",
+)
+@click.option(
+    "--force",
+    is_flag=True,
+    help="Remove all intermediate files for this run before exiting even if run crashes",
+)
+def cli(
+    input_file: str | os.PathLike[str],
+    output_file: str | os.PathLike[str],
+    config: str | os.PathLike[str],
+    work_dir: str | os.PathLike[str] | None,
+    batch_size: int,
+    force: bool,
+):
+    """
+    CLI implementation of tag_fhir_resources_with_authz.
+
+    Args:
+        input_file (str): Input .ndjson file
+        output_file (str): Output file name
+        config (str): .yaml file with authorization rules
+        work_dir (str): Working directory to save intermediate files for each run
+        batch_size (int): number of lines per chunk
+        force (bool): remove all intermediate files for this run before exiting even if it crashes
+    """
+    tag_fhir_resources_with_authz(
+        input_file=input_file,
+        output_file=output_file,
+        config=config,
+        batch_size=batch_size,
+        work_dir=work_dir,
+        force=force,
+    )
+
+
+@click.command(
+    context_settings={"help_option_names": ["-h", "--help"]},
+    help="Remove all intermediate files in the tmp folder from previous runs",
+)
+@click.option(
+    "--work_dir",
+    type=click.Path(),
+    metavar="work_dir",
+    help=f"Specify which working directory to clean, if not specified the default ({DEFAULT_WORK_DIR}) will be cleaned. Can also be set as an environment variable: GEN3_FHIR_WORK_DIR",
+)
+@click.option(
+    "--dry-run",
+    is_flag=True,
+    help="Report what would be deleted without deleting the files",
+)
+@click.option(
+    "--force",
+    is_flag=True,
+    help="Remove temporary directory ignoring status of each directory",
+)
+def cleanup(work_dir, dry_run: bool, force: bool):
+    """
+    Remove all intermediate files in the tmp folder from previous runs
+
+    Args:
+        work_dir (str): Working directory to save intermediate files for each run
+        dry_run (bool): If True, list the files that would be removed, but not actually remove them
+        force (bool): Delete all intermediate directories disregarding the status
+    """
+
+    cleanup_fhir_transform_artifacts(work_dir=work_dir, dry_run=dry_run, force=force)
+
+
+fhir.add_command(cli, name="transform")
+fhir.add_command(cleanup, name="cleanup")