From 70774b16e3abe76790dbdc383c8fa360f6d45930 Mon Sep 17 00:00:00 2001 From: Avik Datta Date: Tue, 6 Feb 2024 13:16:44 +0000 Subject: [PATCH] added dummy task for single sample branch --- ...g36_cellranger_arc_scRNA_multiome_utils.py | 33 +++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/igf_airflow/utils/dag36_cellranger_arc_scRNA_multiome_utils.py b/igf_airflow/utils/dag36_cellranger_arc_scRNA_multiome_utils.py index ce35c9b1..efe19502 100644 --- a/igf_airflow/utils/dag36_cellranger_arc_scRNA_multiome_utils.py +++ b/igf_airflow/utils/dag36_cellranger_arc_scRNA_multiome_utils.py @@ -376,6 +376,39 @@ def configure_cellranger_arc_aggr( raise ValueError( f"Failed to configure cellranger aggr run, error: {e}") + +## TASK +@task( + task_id="dummy_task_for_single_sample", + retry_delay=timedelta(minutes=5), + retries=4, + queue='hpc_8G') +def dummy_task_for_single_sample( + main_work_dir: str) -> str: + try: + return main_work_dir + except Exception as e: + context = get_current_context() + log.error(e) + log_file_path = [ + os.environ.get('AIRFLOW__LOGGING__BASE_LOG_FOLDER'), + f"dag_id={context['ti'].dag_id}", + f"run_id={context['ti'].run_id}", + f"task_id={context['ti'].task_id}", + f"attempt={context['ti'].try_number}.log"] + message = \ + f"Error: {e}, Log: {os.path.join(*log_file_path)}" + send_log_to_channels( + slack_conf=SLACK_CONF, + ms_teams_conf=MS_TEAMS_CONF, + task_id=context['task'].task_id, + dag_id=context['task'].dag_id, + project_id=None, + comment=message, + reaction='fail') + raise ValueError(e) + + ## TASK @task( task_id="run_single_sample_scanpy_for_arc",