bhf_smarthealth.utils.workflow_utils

This module contains utility functions for common Snakemake workflow actions

Attributes

logger

Classes

ValidTargets	Class to hold valid date and type target combinations
SENTINEL_FLAGS	Configure naming of Snakemake sentinel flags

Functions

read_or_generate_run_id(→ str)	Check whether a run_id file has been generated by a snakemake pipeline run, else generate a new run_id file. This is required to ensure persistent run_id state throughout a pipeline run.
get_bucket_credentials(→ dict[str, str])	Retrieve credentials from the s3 object storage associated with the specified 'profile' as configured in the .aws/config file
validate_date_to_process(→ datetime.date)	Validates that the date specified in the Snakemake config.yaml file is a valid ISO date or the string "today".
generate_dates(→ list[str])	Generates dates for days before the given start date, as specified by the values in 'deltas'
get_valid_day_deltas(data_type, all_dates, targets, ...)
get_valid_clean_targets(→ list[ValidTargets])	Return a list of ValidTargets objects that comprise the valid table,date combinations determined from the clean table registry and configured day deltas for each stream.
generate_s3_key(wildcards)	Parses snakemake wildcards to generate file key depending on table partitioning

Module Contents

bhf_smarthealth.utils.workflow_utils.logger

class bhf_smarthealth.utils.workflow_utils.ValidTargets

Class to hold valid date and type target combinations

date: str

type: str

class bhf_smarthealth.utils.workflow_utils.SENTINEL_FLAGS

Bases: enum.StrEnum

Configure naming of Snakemake sentinel flags

PROCESSING_COMPLETE

UPLOAD_RAW_COMPLETE

UPLOAD_INTERMEDIATE_COMPLETE

DATABASE_LOAD_COMPLETE

DATASTREAM_SUCCESS

PIPELINE_ERROR

PIPELINE_SUCCESS

bhf_smarthealth.utils.workflow_utils.read_or_generate_run_id(filepath: pathlib.Path | str) → str

Check whether a run_id file has been generated by a snakemake pipeline run, else generate a new run_id file. This is required to ensure persistent run_id state throughout a pipeline run.

Parameters:

filepath (Path | str) – Path, or string of path, to the lockfile that contains, or should contain, the run_id

Returns:

String of the generated run_id

Return type:

str

Warning

The output file must be deleted by the pipeline onsuccess or onerror to ensure a new run_id is generated.

bhf_smarthealth.utils.workflow_utils.get_bucket_credentials(profile: str) → dict[str, str]

Retrieve credentials from the s3 object storage associated with the specified ‘profile’ as configured in the .aws/config file

Parameters:

profile (str) – String of profile name as configured through AWS CLI

Returns:

Dict containing bucket credentials

Return type:

dict[str, str]

Raises:

RuntimeError – Raised if error retrieveing credentials for the specified profile

bhf_smarthealth.utils.workflow_utils.validate_date_to_process(date_to_process: str | datetime.date) → datetime.date

Validates that the date specified in the Snakemake config.yaml file is a valid ISO date or the string “today”.

Parameters:

date_to_process (str | date) – The DATE_TO_PROCESS config specified in the Snakemake config file

Returns:

validated_date – A date object of the validated date to process

Return type:

date

Raises:

ValueError – Raised if the specified date is not a valid ISO 8601 date or ‘today’

bhf_smarthealth.utils.workflow_utils.generate_dates(start_date: datetime.date, deltas: list[int]) → list[str]

Generates dates for days before the given start date, as specified by the values in ‘deltas’ ‘deltas’ must contain values that are greater than 0, values less than or equal to 0 will be ignored Duplicate values in deltas will also be ignored.

Parameters:

• start_date (date) – Date to generate other dates relative to

• deltas (list[int]) – List of deltas representing the interval (in days) for dates prior to the start date

Returns:

List of ISO8601 formatted date strings

Return type:

list[str]

bhf_smarthealth.utils.workflow_utils.get_valid_day_deltas(data_type, all_dates, targets, wildcards, checkpoints, storage_provider)

1. Runs the checkpoint to list archive contents.

2. Parses the text file to find available dates.

3. Returns targets only for valid dates.

bhf_smarthealth.utils.workflow_utils.get_valid_clean_targets(config: dict[str, Any]) → list[ValidTargets]

Return a list of ValidTargets objects that comprise the valid table,date combinations determined from the clean table registry and configured day deltas for each stream.

Parameters:

config (dict[str, Any]) – Snakemake config dict (should contain the streams and their associated day deltas)

Returns:

List of ValidTargets

Return type:

list[ValidTargets]

bhf_smarthealth.utils.workflow_utils.generate_s3_key(wildcards)

Parses snakemake wildcards to generate file key depending on table partitioning