Release 0.29.0 (#65)

* Fix missing init

* Dependencies fixes (#55)

* Fix missing init

* Fix test

* Fix precommit

Co-authored-by: p-pekala <piotr.pekala@getindata.com>

* Moving CC_TEST_REPORTER_ID from secrets to allow contribution (CC_TES… (#59)

* Moving CC_TEST_REPORTER_ID from secrets to allow contribution (CC_TEST_REPORTER_ID is just an ID not credential).

Co-authored-by: piotrpekala <piotr.pekala@getindata.com>

* feat: Allows passing config file to the operator (#57)

* feat: Allows passing config file to the operator

Co-authored-by: bartlomiej.olma.kitopi <bartlomiej.olma@kitopi.com>

* New documentation (#58)

* New documentation

Co-authored-by: PP <piotr.pekala@getindata.com>

* Moving CC_TEST_REPORTER_ID from secrets to allow contribution (CC_TEST_REPORTER_ID is just an ID not credential).

* Data 377 (#62)

* feat: add config envs if datahub recipe exists

* feat: add test to check if env are properly readed

* feat: add execution script parameter

* feat: add logic behind using execution script

* chore: pre-commit

* feat: add test for execution script and datahub env

* chore: update bash script name

* chore: parametrize script path

* chore: refactor args builder

* chore: refactor tests after changes in args

* chore: add defualt value to backward compatible

* chore: refactor tests

Co-authored-by: Piotr Sierkin <piotr.sierkin@getindata.com>

* Data 369 integrate scheduling airbyte from airflow in the same dag as transformations (#64)

* feat: [DATA-369] Add ingestion tasks.

* Data 412 add save points after each stage of transformations (#61)

* feat: [DATA-412] Abstract away task with mapping task from graph to airflow.

* FIX #65  - Bump version and CHANGELOG for release 0.29.0

Co-authored-by: p-pekala <piotr.pekala@getindata.com>
Co-authored-by: github-actions <github-actions@github.com>
Co-authored-by: Piotr Pękala <piotr_pekala@wp.pl>
Co-authored-by: bartlomiejolma <b.olma92@gmail.com>
Co-authored-by: bartlomiej.olma.kitopi <bartlomiej.olma@kitopi.com>
Co-authored-by: Piotr Sierkin <psierkin@gmail.com>
Co-authored-by: Piotr Sierkin <piotr.sierkin@getindata.com>
Co-authored-by: Paweł Kociński <pawel93kocinski@gmail.com>

Author: 8 people

.isort.cfg

@@ -4,4 +4,4 @@ known_first_party = dbt_airflow_factory
 default_section = THIRDPARTY
 
 [settings]
-known_third_party = airflow,jinja2,networkx,pytimeparse,setuptools,yaml
+known_third_party = airflow,jinja2,networkx,pytest,pytimeparse,setuptools,yaml

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## [Unreleased]
 
+## [0.29.0] - 2022-09-02
+
 ## [0.28.0] - 2022-07-19
 
 ## [0.27.0] - 2022-07-01
@@ -122,7 +124,9 @@ This version brings compatibility with `dbt 1.0`.
 
 -   Initial implementation of `dbt_airflow_manifest_parser` library.
 
-[Unreleased]: https://github.com/getindata/dbt-airflow-factory/compare/0.28.0...HEAD
+[Unreleased]: https://github.com/getindata/dbt-airflow-factory/compare/0.29.0...HEAD
+
+[0.29.0]: https://github.com/getindata/dbt-airflow-factory/compare/0.28.0...0.29.0
 
 [0.28.0]: https://github.com/getindata/dbt-airflow-factory/compare/0.27.0...0.28.0
 

dbt_airflow_factory/__init__.py

@@ -1 +1 @@
-version = "0.28.0"
+version = "0.29.0"

dbt_airflow_factory/airflow_dag_factory.py

@@ -6,6 +6,8 @@
 from airflow import DAG
 from airflow.models import BaseOperator
 
+from dbt_airflow_factory.ingestion import IngestionEngine, IngestionFactory
+
 if airflow.__version__.startswith("1."):
     from airflow.operators.dummy_operator import DummyOperator
 else:
@@ -14,7 +16,7 @@
 from pytimeparse import parse
 
 from dbt_airflow_factory.builder_factory import DbtAirflowTasksBuilderFactory
-from dbt_airflow_factory.config_utils import read_config
+from dbt_airflow_factory.config_utils import read_config, read_env_config
 from dbt_airflow_factory.notifications.handler import NotificationHandlersFactory
 from dbt_airflow_factory.tasks_builder.builder import DbtAirflowTasksBuilder
 
@@ -53,6 +55,8 @@ def __init__(
         dbt_config_file_name: str = "dbt.yml",
         execution_env_config_file_name: str = "execution_env.yml",
         airflow_config_file_name: str = "airflow.yml",
+        airbyte_config_file_name: str = "airbyte.yml",
+        ingestion_config_file_name: str = "ingestion.yml",
     ):
         self._notifications_handlers_builder = NotificationHandlersFactory()
         self.airflow_config = self._read_config(dag_path, env, airflow_config_file_name)
@@ -64,6 +68,18 @@ def __init__(
             execution_env_config_file_name,
         ).create()
         self.dag_path = dag_path
+        airbyte_config = read_env_config(
+            dag_path=dag_path, env=env, file_name=airbyte_config_file_name
+        )
+        self.ingestion_config = read_env_config(
+            dag_path=dag_path, env=env, file_name=ingestion_config_file_name
+        )
+        self.ingestion_tasks_builder_factory = IngestionFactory(
+            ingestion_config=airbyte_config,
+            name=IngestionEngine.value_of(
+                self.ingestion_config.get("engine", IngestionEngine.AIRBYTE.value)
+            ),
+        )
 
     def create(self) -> DAG:
         """
@@ -82,7 +98,14 @@ def create_tasks(self) -> None:
         """
         Parse ``manifest.json`` and create tasks based on the data contained there.
         """
+
+        ingestion_enabled = self.ingestion_config.get("enable", False)
+
         start = self._create_starting_task()
+        if ingestion_enabled and self.ingestion_tasks_builder_factory:
+            builder = self.ingestion_tasks_builder_factory.create()
+            ingestion_tasks = builder.build()
+            ingestion_tasks >> start
         end = DummyOperator(task_id="end")
         tasks = self._builder.parse_manifest_into_tasks(self._manifest_file_path())
         for starting_task in tasks.get_starting_tasks():

dbt_airflow_factory/builder_factory.py

@@ -10,6 +10,10 @@
 )
 from dbt_airflow_factory.operator import DbtRunOperatorBuilder
 from dbt_airflow_factory.tasks_builder.builder import DbtAirflowTasksBuilder
+from dbt_airflow_factory.tasks_builder.gateway import (
+    GatewayConfiguration,
+    TaskGraphConfiguration,
+)
 from dbt_airflow_factory.tasks_builder.parameters import TasksBuildingParameters
 
 
@@ -65,9 +69,16 @@ def create(self) -> DbtAirflowTasksBuilder:
         dbt_params = self._create_dbt_config()
         execution_env_type = self._read_execution_env_type()
         tasks_airflow_config = self._create_tasks_airflow_config()
+
         return DbtAirflowTasksBuilder(
             tasks_airflow_config,
             self._create_operator_builder(execution_env_type, dbt_params),
+            gateway_config=TaskGraphConfiguration(
+                GatewayConfiguration(
+                    separation_schemas=self.airflow_config.get("save_points", []),
+                    gateway_task_name="gateway",
+                )
+            ),
         )
 
     def _create_tasks_airflow_config(self) -> TasksBuildingParameters:

dbt_airflow_factory/ingestion.py

@@ -0,0 +1,54 @@
+from enum import Enum
+from typing import List
+
+from airflow.models import BaseOperator
+from airflow.providers.airbyte.operators.airbyte import AirbyteTriggerSyncOperator
+
+
+class IngestionEngine(Enum):
+    AIRBYTE = "airbyte"
+
+    @classmethod
+    def value_of(cls, value: str) -> "IngestionEngine":
+        return IngestionEngine(value)
+
+
+class IngestionTasksBuilder:
+    def build(self) -> List[BaseOperator]:
+        raise NotImplementedError("Should implement build method")
+
+
+class AirbyteIngestionTasksBuilder(IngestionTasksBuilder):
+    def __init__(self, config: dict):
+        self.ingestion_config = config
+
+    def build(self) -> List[BaseOperator]:
+        airflow_tasks = []
+        tasks = self.ingestion_config["tasks"]
+        for task in tasks:
+            airflow_tasks.append(
+                AirbyteTriggerSyncOperator(
+                    task_id=task["task_id"],
+                    airbyte_conn_id=self.ingestion_config["airbyte_connection_id"],
+                    connection_id=task["connection_id"],
+                    asynchronous=task["asyncrounous"],
+                    api_version=task["api_version"],
+                    wait_seconds=task["wait_seconds"],
+                    timeout=task["timeout"],
+                )
+            )
+
+        return airflow_tasks
+
+
+class IngestionFactory:
+    def __init__(self, ingestion_config: dict, name: IngestionEngine):
+        self.ingestion_config = ingestion_config
+        self.name = name
+
+    def create(
+        self,
+    ) -> IngestionTasksBuilder:
+        if self.name == IngestionEngine.AIRBYTE:
+            return AirbyteIngestionTasksBuilder(self.ingestion_config)
+        raise NotImplementedError(f"{self.name} is not supported !")

dbt_airflow_factory/tasks_builder/builder.py

@@ -5,6 +5,7 @@
 
 import airflow
 from airflow.models.baseoperator import BaseOperator
+from airflow.operators.dummy import DummyOperator
 from airflow.sensors.external_task_sensor import ExternalTaskSensor
 
 from dbt_airflow_factory.tasks_builder.node_type import NodeType
@@ -15,24 +16,31 @@
 
 from dbt_airflow_factory.operator import DbtRunOperatorBuilder, EphemeralOperator
 from dbt_airflow_factory.tasks import ModelExecutionTask, ModelExecutionTasks
+from dbt_airflow_factory.tasks_builder.gateway import TaskGraphConfiguration
 from dbt_airflow_factory.tasks_builder.graph import DbtAirflowGraph
 
 
 class DbtAirflowTasksBuilder:
     """
     Parses ``manifest.json`` into Airflow tasks.
 
+    :param airflow_config: DBT node operator.
+    :type airflow_config: TasksBuildingParameters
     :param operator_builder: DBT node operator.
     :type operator_builder: DbtRunOperatorBuilder
+    :param gateway_config: DBT node operator.
+    :type gateway_config: TaskGraphConfiguration
     """
 
     def __init__(
         self,
         airflow_config: TasksBuildingParameters,
         operator_builder: DbtRunOperatorBuilder,
+        gateway_config: TaskGraphConfiguration,
     ):
         self.operator_builder = operator_builder
         self.airflow_config = airflow_config
+        self.gateway_config = gateway_config
 
     def parse_manifest_into_tasks(self, manifest_path: str) -> ModelExecutionTasks:
         """
@@ -128,6 +136,8 @@ def _create_task_from_graph_node(
             )
         elif node["node_type"] == NodeType.SOURCE_SENSOR:
             return self._create_dag_sensor(node)
+        elif node["node_type"] == NodeType.MOCK_GATEWAY:
+            return self._create_dummy_task(node)
         else:
             return self._create_task_for_model(
                 node["select"],
@@ -157,7 +167,8 @@ def _make_dbt_tasks(self, manifest_path: str) -> ModelExecutionTasks:
         return tasks_with_context
 
     def _create_tasks_graph(self, manifest: dict) -> DbtAirflowGraph:
-        dbt_airflow_graph = DbtAirflowGraph()
+
+        dbt_airflow_graph = DbtAirflowGraph(self.gateway_config)
         dbt_airflow_graph.add_execution_tasks(manifest)
         if self.airflow_config.enable_dags_dependencies:
             dbt_airflow_graph.add_external_dependencies(manifest)
@@ -183,3 +194,7 @@ def _create_dag_sensor(self, node: Dict[str, Any]) -> ModelExecutionTask:
                 mode="reschedule",
             )
         )
+
+    @staticmethod
+    def _create_dummy_task(node: Dict[str, Any]) -> ModelExecutionTask:
+        return ModelExecutionTask(DummyOperator(task_id=node["select"]))

dbt_airflow_factory/tasks_builder/gateway.py

@@ -0,0 +1,112 @@
+from dataclasses import dataclass
+from typing import List
+
+from dbt_airflow_factory.tasks_builder.utils import is_model_run_task
+
+
+@dataclass
+class NodeProperties:
+    node_name: str
+    schema_name: str
+
+
+@dataclass
+class GatewayConfiguration:
+    separation_schemas: List[str]
+    gateway_task_name: str
+
+
+@dataclass
+class TaskGraphConfiguration:
+    gateway: GatewayConfiguration
+
+
+@dataclass
+class SeparationLayer:
+    left: str
+    right: str
+
+
+def is_gateway_valid_dependency(
+    separation_layer: SeparationLayer, dependency_node_properties: NodeProperties, node_schema: str
+) -> bool:
+    if is_model_run_task(dependency_node_properties.node_name):
+        dep_schema = dependency_node_properties.schema_name
+        if dep_schema == separation_layer.left and node_schema == separation_layer.right:
+            return False
+        return True
+    return True
+
+
+def get_gateway_dependencies(manifest: dict, separation_layer: SeparationLayer) -> List:
+    downstream_dependencies = _get_downstream_dependencies(
+        manifest=manifest, separation_layer_right=separation_layer.right
+    )
+
+    upstream_dependencies_connected_to_downstream = (
+        _get_upstream_dependencies_connected_to_downstream(
+            manifest=manifest,
+            separation_layer_left=separation_layer.left,
+            downstream_dependencies=downstream_dependencies,
+        )
+    )
+    dependencies = [
+        node_name
+        for node_name, values in manifest["nodes"].items()
+        if values["schema"] == separation_layer.left
+        and node_name in upstream_dependencies_connected_to_downstream
+    ]
+    return dependencies
+
+
+def _get_downstream_dependencies(manifest: dict, separation_layer_right: str) -> List:
+    downstream_dependencies = [
+        node_name
+        for node_name, values in manifest["nodes"].items()
+        if values["schema"] == separation_layer_right
+    ]
+    return downstream_dependencies
+
+
+def _get_upstream_dependencies_connected_to_downstream(
+    manifest: dict, separation_layer_left: str, downstream_dependencies: List[str]
+) -> List:
+    upstream_deps_connected_to_downstream: List[str] = []
+
+    for downstream_node in downstream_dependencies:
+        upstream_deps = manifest["nodes"][downstream_node]["depends_on"]["nodes"]
+        for dep in upstream_deps:
+            _add_upstream_dep_based_on_downstream(
+                dep=dep,
+                manifest=manifest,
+                separation_layer_left=separation_layer_left,
+                upstream_dependencies_connected_to_downstream=upstream_deps_connected_to_downstream,
+            )
+    return upstream_deps_connected_to_downstream
+
+
+def _add_upstream_dep_based_on_downstream(
+    dep: str,
+    manifest: dict,
+    separation_layer_left: str,
+    upstream_dependencies_connected_to_downstream: List[str],
+) -> None:
+    if is_model_run_task(dep) and manifest["nodes"][dep]["schema"] == separation_layer_left:
+        upstream_dependencies_connected_to_downstream.append(dep)
+
+
+def add_gateway_to_dependencies(
+    gateway_name: str, filtered_dependencies: List[str], filtered_records: List[str]
+) -> None:
+    if len(filtered_dependencies) < len(filtered_records):
+        filtered_dependencies.append(gateway_name)
+
+
+def create_gateway_name(separation_layer: SeparationLayer, gateway_task_name: str) -> str:
+    return f"{separation_layer.left}_{separation_layer.right}_{gateway_task_name}"
+
+
+def should_gateway_be_added(node_schema: str, separation_schemas: List[str]) -> bool:
+    valid_schemas_input_length = len(separation_schemas) >= 2
+    schema_is_in_given_schemas = node_schema in separation_schemas
+    return valid_schemas_input_length and schema_is_in_given_schemas