Feature/data 1791 defferable sensors

Makefile

@@ -96,15 +96,15 @@ install: clean ## install the package to the active Python's site-packages
 
 
 install-dev: clean ## install the package to the active Python's site-packages
-	virtualenv -p python3 venv; \
+	virtualenv -p python3.9 venv; \
 	source venv/bin/activate; \
 	python -m pip install --upgrade pip; \
 	python setup.py install; \
 	pip install -e . ; \
-	pip install -r reqs/dev.txt -r reqs/test.txt
+	SYSTEM_VERSION_COMPAT=0 CFLAGS='-std=c++20' pip install -r reqs/dev.txt -r reqs/test.txt
 
 install-test: clean ## install the package to the active Python's site-packages
-	virtualenv -p python3 venv; \
+	virtualenv -p python3.9 venv; \
 	source venv/bin/activate; \
 	python -m pip install --upgrade pip; \
 	pip install -r reqs/test.txt -r reqs/base.txt

dagger/conf.py

@@ -21,9 +21,7 @@
 # Airflow parameters
 airflow_config = config.get('airflow', None) or {}
 WITH_DATA_NODES = airflow_config.get('with_data_nodes', False)
-EXTERNAL_SENSOR_POKE_INTERVAL = airflow_config.get('external_sensor_poke_interval', 600)
-EXTERNAL_SENSOR_TIMEOUT = airflow_config.get('external_sensor_timeout', 28800)
-EXTERNAL_SENSOR_MODE = airflow_config.get('external_sensor_mode', 'reschedule')
+EXTERNAL_SENSOR_DEFAULT_ARGS = airflow_config.get('external_sensor_default_args', {})
 IS_DUMMY_OPERATOR_SHORT_CIRCUIT = airflow_config.get('is_dummy_operator_short_circuit', False)
 
 # Neo4j parameters

dagger/dag_creator/airflow/dag_creator.py

@@ -72,12 +72,7 @@ def _get_external_task_sensor(self, from_task_id: str, to_task_id: str, follow_e
 
         to_pipe_id = self._task_graph.get_node(to_task_id).obj.pipeline.name
 
-
-        extra_args = {
-            'mode': conf.EXTERNAL_SENSOR_MODE,
-            'poke_interval': conf.EXTERNAL_SENSOR_POKE_INTERVAL,
-            'timeout': conf.EXTERNAL_SENSOR_TIMEOUT,
-        }
+        extra_args = conf.EXTERNAL_SENSOR_DEFAULT_ARGS.copy()
         extra_args.update(follow_external_dependency)
 
         return ExternalTaskSensor(
@@ -141,6 +136,7 @@ def _create_edge_without_data(self, from_task_id: str, to_task_ids: list, node:
             to_task_ids: The IDs of the tasks to which the edge connects.
             node: The current node in a task graph.
         """
+
         from_pipe = (
             self._task_graph.get_node(from_task_id).obj.pipeline_name if from_task_id else None
         )

dagger/dag_creator/airflow/operator_creators/redshift_load_creator.py

@@ -2,9 +2,8 @@
 from typing import Optional
 
 from dagger.dag_creator.airflow.operator_creator import OperatorCreator
-from dagger.dag_creator.airflow.operators.redshift_sql_operator import (
-    RedshiftSQLOperator,
-)
+from dagger.dag_creator.airflow.operators.redshift_sql_operator import RedshiftSQLOperator
+
 
 
 class RedshiftLoadCreator(OperatorCreator):

dagger/dag_creator/airflow/operator_creators/redshift_transform_creator.py

@@ -1,7 +1,7 @@
 from os.path import join
 
 from dagger.dag_creator.airflow.operator_creator import OperatorCreator
-from dagger.dag_creator.airflow.operators.postgres_operator import PostgresOperator
+from dagger.dag_creator.airflow.operators.redshift_sql_operator import RedshiftSQLOperator
 
 
 class RedshiftTransformCreator(OperatorCreator):
@@ -22,11 +22,11 @@ def _read_sql(directory, file_path):
     def _create_operator(self, **kwargs):
         sql_string = self._read_sql(self._task.pipeline.directory, self._task.sql_file)
 
-        redshift_op = PostgresOperator(
+        redshift_op = RedshiftSQLOperator(
             dag=self._dag,
             task_id=self._task.name,
             sql=sql_string,
-            postgres_conn_id=self._task.postgres_conn_id,
+            redshift_conn_id=self._task.postgres_conn_id,
             params=self._template_parameters,
             **kwargs,
         )

dagger/dag_creator/airflow/operator_creators/redshift_unload_creator.py

@@ -1,7 +1,7 @@
 from os.path import join
 
 from dagger.dag_creator.airflow.operator_creator import OperatorCreator
-from dagger.dag_creator.airflow.operators.postgres_operator import PostgresOperator
+from dagger.dag_creator.airflow.operators.redshift_sql_operator import RedshiftSQLOperator
 
 REDSHIFT_UNLOAD_CMD = """
 unload ('{sql_string}')
@@ -58,12 +58,13 @@ def _create_operator(self, **kwargs):
 
         unload_cmd = self._get_unload_command(sql_string)
 
-        redshift_op = PostgresOperator(
+        redshift_op = RedshiftSQLOperator(
             dag=self._dag,
             task_id=self._task.name,
             sql=unload_cmd,
-            postgres_conn_id=self._task.postgres_conn_id,
+            redshift_conn_id=self._task.postgres_conn_id,
             params=self._template_parameters,
+            autocommit=True,
             **kwargs,
         )
 

dagger/dag_creator/airflow/operators/postgres_operator.py

@@ -1,6 +1,6 @@
 from typing import Iterable, Mapping, Optional, Union
 
-from airflow.hooks.postgres_hook import PostgresHook
+from airflow.providers.postgres.hooks.postgres import PostgresHook
 from airflow.utils.decorators import apply_defaults
 from dagger.dag_creator.airflow.operators.dagger_base_operator import DaggerBaseOperator
 

dagger/dagger_config.yaml

@@ -1,8 +1,10 @@
 airflow:
+  external_sensor_default_args:
+    poll_interval: 30
+    timeout: 28800
+    mode: reschedule
+    deferrable: true
   with_data_node: false
-  external_sensor_poke_interval: 600
-  external_sensor_timeout: 28800
-  external_sensor_mode: reschedule
   is_dummy_operator_short_circuit: false
 
 

reqs/dev.txt

@@ -1,19 +1,18 @@
-apache-airflow[amazon,postgres,s3,statsd]==2.3.4
+pip==24.0
+apache-airflow[amazon,postgres,s3,statsd]==2.9.0 --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.9.0/constraints-3.9.txt"
 black==22.10.0
-boto3==1.26.16
+boto3==1.34.82
 bumpversion==0.6.0
-coverage==6.5.0
-elasticsearch==7.17.7
-flake8==5.0.4
-neo4j==5.2.1
-numpydoc==1.5.0
-pip==22.3.1
+coverage==7.4.4
+#elasticsearch==7.17.7
+flake8==7.0.0
+#neo4j==5.19.0
+numpydoc==1.7.0
 pre-commit==2.20.0
-sphinx-rtd-theme==1.1.1
-Sphinx==4.3.2
-SQLAlchemy==1.4.44
-tox==3.27.1
-twine==4.0.1
-watchdog==2.1.9
-Werkzeug==2.2.2
-wheel==0.38.4
+sphinx-rtd-theme==2.0.0
+Sphinx==7.2.6
+SQLAlchemy
+tox==4.14.2
+twine==5.0.0
+watchdog==4.0.0
+Werkzeug

reqs/test.txt

@@ -1,3 +1,3 @@
-apache-airflow[amazon,postgres,s3,statsd]==2.3.4
+apache-airflow[amazon,postgres,s3,statsd]==2.9.0
 pytest-cov==4.0.0
 pytest==7.2.0

setup.py

@@ -45,8 +45,7 @@ def reqs(*f):
     classifiers=[
         "Development Status :: 2 - Pre-Alpha",
         "Intended Audience :: Developers",
-        "Programming Language :: Python :: 3.6",
-        "Programming Language :: Python :: 3.7",
+        "Programming Language :: Python :: 3.9",
     ],
     description="Config Driven ETL",
     entry_points={"console_scripts": ["dagger=dagger.main:cli"]},