Update Application model to be more generic (#166)

databuilder/models/application.py

@@ -28,14 +28,17 @@ class Application(Neo4jCsvSerializable):
                  task_id,  # type: str
                  dag_id,  # type: str,
                  application_url_template,  # type: str
-                 exec_date,  # type: str
+                 db_name='hive',  # type: str
+                 schema_name='',  # type: str
+                 table_name='',  # type: str
+                 exec_date='',  # type: str
                  ):
         # type: (...) -> None
         self.task = task_id
 
         # todo: need to modify this hack
         self.application_url = application_url_template.format(dag_id=dag_id)
-        self.database, self.schema, self.table = task_id.split('.')
+        self.database, self.schema, self.table = db_name, schema_name, table_name
 
         self.dag = dag_id
 

example/sample_data/sample_application.csv

-task_id,dag_id,exec_date,application_url_template
-hive.test_schema.test_table1,event_test,"2018-05-31T00:00:00","https://airflow_host.net/admin/airflow/tree?dag_id={dag_id}"
+task_id,dag_id,exec_date,application_url_template,db_name,schema_name,table_name
+hive.test_schema.test_table1,event_test,"2018-05-31T00:00:00","https://airflow_host.net/admin/airflow/tree?dag_id={dag_id}",hive,test_schema,test_table1

example/scripts/sample_data_loader.py

@@ -229,17 +229,24 @@ def load_application_data_from_csv(file_name):
                     '(task_id VARCHAR(64) NOT NULL , '
                     'dag_id VARCHAR(64) NOT NULL , '
                     'exec_date VARCHAR(64) NOT NULL, '
-                    'application_url_template VARCHAR(128) NOT NULL)')
+                    'application_url_template VARCHAR(128) NOT NULL, '
+                    'db_name VARCHAR(64) NOT NULL, '
+                    'schema_name VARCHAR(64) NOT NULL, '
+                    'table_name VARCHAR(64) NOT NULL)')
         file_loc = 'example/sample_data/' + file_name
         with open(file_loc, 'r') as fin:
             dr = csv.DictReader(fin)
             to_db = [(i['task_id'],
                       i['dag_id'],
                       i['exec_date'],
-                      i['application_url_template']) for i in dr]
+                      i['application_url_template'],
+                      i['db_name'],
+                      i['schema_name'],
+                      i['table_name'],) for i in dr]
 
         cur.executemany("INSERT INTO test_application_metadata (task_id, dag_id, "
-                        "exec_date, application_url_template) VALUES (?, ?, ?, ?);", to_db)
+                        "exec_date, application_url_template, db_name, schema_name, table_name) "
+                        "VALUES (?, ?, ?, ?, ?, ?, ?);", to_db)
         conn.commit()
 
 

setup.py

@@ -2,7 +2,7 @@ import os
 from setuptools import setup, find_packages
 
 
-__version__ = '1.4.11'
+__version__ = '1.4.12'
 
 
 requirements_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'requirements.txt')

tests/unit/models/test_application.py

@@ -16,7 +16,8 @@ class TestApplication(unittest.TestCase):
 
         self.application = Application(task_id='hive.default.test_table',
                                        dag_id='event_test',
-                                       exec_date='2018-05-31T00:00:00',
+                                       schema_name='default',
+                                       table_name='test_table',
                                        application_url_template='airflow_host.net/admin/airflow/tree?dag_id={dag_id}')
 
         self.expected_node_result = {