Jpuerto/example8 dag

src/ingest-pipeline/airflow/dags/generate_bdbag.py

@@ -53,9 +53,7 @@
     hubmapbags_id_namespace as id_namespace,
     hubmapbags_biosample_in_collection as biosample_in_collection,
     hubmapbags_file_in_collection as file_in_collection,
-    hubmapbags_primary_dcc_contact as primary_dcc_contact,
     hubmapbags_biosample as biosample,
-    hubmapbags_projects as projects,
     hubmapbags_collection as collection,
     hubmapbags_anatomy as anatomy,
     hubmapbags_file as files,
@@ -284,9 +282,6 @@ def generate_bdbag(**kwargs):
             print('Making biosample_in_collection.tsv')
             biosample_in_collection.create_manifest(biosample_id, hubmap_id, output_directory)
 
-            print('Making project.tsv')
-            projects.create_manifest(data_provider, output_directory)
-
             print('Making project_in_project.tsv')
             project_in_project.create_manifest(data_provider, output_directory)
 
@@ -305,9 +300,6 @@ def generate_bdbag(**kwargs):
             print('Making file_describes_collection.tsv')
             file_describes_collection.create_manifest(hubmap_id, data_directory, output_directory)
 
-            print('Making dcc.tsv')
-            primary_dcc_contact.create_manifest(output_directory)
-
             print('Making id_namespace.tsv')
             id_namespace.create_manifest(output_directory)
 

src/ingest-pipeline/airflow/dags/hubmapbags/generate_today_json.py

@@ -0,0 +1,75 @@
+import os
+import sys
+from datetime import datetime, timedelta
+
+from airflow.operators.python import PythonOperator
+from airflow.configuration import conf as airflow_conf
+
+from utils import (
+    HMDAG,
+    get_queue_resource,
+    get_auth_tok,
+    encrypt_tok,
+    )
+
+class add_path:
+    """
+    Add an element to sys.path using a context.
+    Thanks to Eugene Yarmash https://stackoverflow.com/a/39855753
+    """
+
+    def __init__(self, path):
+        self.path = path
+
+    def __enter__(self):
+        sys.path.insert(0, self.path)
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        try:
+            sys.path.remove(self.path)
+        except ValueError:
+            pass
+
+with add_path(airflow_conf.as_dict()["connections"]["SRC_PATH"].strip("'").strip('"')):
+    from submodules import hubmapbags
+
+# Following are defaults which can be overridden later on
+default_args = {
+    'owner': 'hubmap',
+    'depends_on_past': False,
+    'start_date': datetime.now(),
+    'email': ['[email protected]'],
+    'email_on_failure': False,
+    'email_on_retry': False,
+    'retries': 1,
+    'retry_delay': timedelta(minutes=1),
+    'xcom_push': True,
+    'queue': get_queue_resource('generate_today_json'),
+}
+
+
+with HMDAG('generate_today_json',
+           schedule_interval="0 0 * * *",
+           is_paused_upon_creation=False,
+           default_args=default_args,
+           user_defined_macros={
+           }) as dag:
+
+    def generate_report(**kwargs):
+        token = get_auth_tok(**kwargs)
+        hubmapbags.utilities.clean()
+        hubmapbags.reports.daily(token)
+        print('Finished')
+
+    t_generate_report = PythonOperator(
+        task_id='generate_report',
+        python_callable=generate_report,
+        provide_context=True,
+        op_kwargs={
+            "crypt_auth_tok": (
+                encrypt_tok(airflow_conf.as_dict()["connections"]["APP_CLIENT_SECRET"]).decode()
+            ),
+        },
+        )
+
+    t_generate_report

src/ingest-pipeline/airflow/dags/rebuild_multiple_metadata.py

@@ -1,32 +1,18 @@
-import os
-import yaml
-import utils
 from pprint import pprint
 
-from airflow.operators.bash import BashOperator
 from airflow.operators.python import PythonOperator
 from airflow.operators.trigger_dagrun import TriggerDagRunOperator
-from airflow.exceptions import AirflowException
 from airflow.configuration import conf as airflow_conf
-from datetime import datetime, timedelta
+from datetime import datetime
+from airflow import DAG
 
 from utils import (
-    HMDAG,
     get_queue_resource,
     get_preserve_scratch_resource,
     get_soft_data,
-    create_dataset_state_error_callback,
-    pythonop_md_consistency_tests,
-    make_send_status_msg_function,
     get_tmp_dir_path,
-    localized_assert_json_matches_schema as assert_json_matches_schema,
-    pythonop_get_dataset_state,
     encrypt_tok,
-)
-
-from hubmap_operators.common_operators import (
-    CreateTmpDirOperator,
-    CleanupTmpDirOperator,
+    pythonop_get_dataset_state,
 )
 
 
@@ -37,30 +23,11 @@ def get_uuid_for_error(**kwargs):
     return None
 
 
-def get_dataset_uuid(**kwargs):
-    return kwargs['dag_run'].conf['uuid']
-
-
-def get_dataset_lz_path(**kwargs):
-    ctx = kwargs['dag_run'].conf
-    return ctx['lz_path']
-
-
 default_args = {
-    'owner': 'hubmap',
-    'depends_on_past': False,
     'start_date': datetime(2019, 1, 1),
-    'email': ['[email protected]'],
-    'email_on_failure': False,
-    'email_on_retry': False,
-    'retries': 1,
-    'retry_delay': timedelta(minutes=1),
-    'xcom_push': True,
-    'queue': get_queue_resource('rebuild_metadata'),
-    'on_failure_callback': create_dataset_state_error_callback(get_uuid_for_error)
 }
 
-with HMDAG('rebuild_multiple_metadata',
+with DAG('rebuild_multiple_metadata',
            schedule_interval=None,
            is_paused_upon_creation=False,
            default_args=default_args,
@@ -77,15 +44,30 @@ def build_dataset_lists(**kwargs):
         pprint(kwargs['dag_run'].conf)
         for uuid in kwargs['dag_run'].conf['uuids']:
             soft_data = get_soft_data(uuid, **kwargs)
-            if soft_data.get('primary'):
-                kwargs['dag_run'].conf['primary_datasets'].append(uuid)
+
+            # If we got nothing back from soft_data, then let's try to determine using entity_api
+            if soft_data:
+                if soft_data.get('primary'):
+                    kwargs['dag_run'].conf['primary_datasets'].append(uuid)
+                else:
+                    kwargs['dag_run'].conf['processed_datasets'].append(uuid)
             else:
-                kwargs['dag_run'].conf['processed_datasets'].append(uuid)
+                print(f'No matching soft data returned for {uuid}')
+                ds_rslt = pythonop_get_dataset_state(dataset_uuid_callable=lambda **kwargs: uuid, **kwargs)
+                if ds_rslt.get("dataset_info"):
+                    # dataset_info should only be populated for processed_datasets
+                    print(ds_rslt.get("dataset_info"))
+                    kwargs['dag_run'].conf['processed_datasets'].append(uuid)
+                else:
+                    kwargs['dag_run'].conf['primary_datasets'].append(uuid)
+
+
 
     t_build_dataset_lists = PythonOperator(
         task_id='build_dataset_lists',
         python_callable=build_dataset_lists,
         provide_context=True,
+        queue= get_queue_resource('rebuild_metadata'),
         op_kwargs={
             'crypt_auth_tok': encrypt_tok(airflow_conf.as_dict()
                                           ['connections']['APP_CLIENT_SECRET']).decode(),
@@ -97,7 +79,8 @@ def get_primary_dataset_uuids(**kwargs):
 
     t_get_primary_dataset_uuids = PythonOperator(
         task_id='get_primary_dataset_uuids',
-        python_callable=get_primary_dataset_uuids(),
+        python_callable=get_primary_dataset_uuids,
+        queue=get_queue_resource('rebuild_metadata'),
         provide_context=True
     )
 
@@ -106,22 +89,27 @@ def get_processed_dataset_uuids(**kwargs):
 
     t_get_processed_dataset_uuids = PythonOperator(
         task_id='get_processed_dataset_uuids',
-        python_callable=get_processed_dataset_uuids(),
+        python_callable=get_processed_dataset_uuids,
+        queue=get_queue_resource('rebuild_metadata'),
         provide_context=True
     )
 
-    t_launch_rebuild_primary_dataset_metadata = TriggerDagRunOperator().partial(
+    t_launch_rebuild_primary_dataset_metadata = TriggerDagRunOperator.partial(
         task_id="trigger_rebuild_primary_dataset_metadata",
         trigger_dag_id="rebuild_primary_dataset_metadata",
+        queue=get_queue_resource('rebuild_metadata'),
     ).expand(
         conf=t_get_primary_dataset_uuids.output
     )
 
-    t_launch_rebuild_processed_dataset_metadata = TriggerDagRunOperator().partial(
+    t_launch_rebuild_processed_dataset_metadata = TriggerDagRunOperator.partial(
         task_id="trigger_rebuild_processed_dataset_metadata",
         trigger_dag_id="rebuild_processed_dataset_metadata",
+        queue=get_queue_resource('rebuild_metadata'),
     ).expand(
         conf=t_get_processed_dataset_uuids.output
     )
 
-    t_build_dataset_lists >> t_get_primary_dataset_uuids >> t_get_processed_dataset_uuids >> t_launch_rebuild_primary_dataset_metadata >> t_launch_rebuild_processed_dataset_metadata
+    t_build_dataset_lists >> [t_get_primary_dataset_uuids, t_get_processed_dataset_uuids]
+    t_get_primary_dataset_uuids >> t_launch_rebuild_primary_dataset_metadata
+    t_get_processed_dataset_uuids >> t_launch_rebuild_processed_dataset_metadata

src/ingest-pipeline/airflow/dags/rebuild_primary_dataset_metadata.py

@@ -58,7 +58,7 @@ def get_dataset_lz_path(**kwargs):
     'on_failure_callback': create_dataset_state_error_callback(get_uuid_for_error)
 }
 
-with HMDAG('rebuild_metadata',
+with HMDAG('rebuild_primary_dataset_metadata',
            schedule_interval=None,
            is_paused_upon_creation=False,
            default_args=default_args,