feat: add dags ci and cd github actions

.github/workflows/dags_test.yml

@@ -0,0 +1,30 @@
+name: Dags Check
+on:
+
+    pull_request:
+        branches:
+        - main
+jobs:
+    test:
+        runs-on: ubuntu-latest
+        steps:
+        - uses: actions/checkout@v3
+        - name: Set up Python
+          uses: actions/setup-python@v4
+          with:
+            python-version: "3.9"
+
+        - name: Install dependencies
+          run: |
+            python -m pip install --upgrade pip
+            pip install -r dags/requirements.txt
+            pip check
+        - name: Lint dags with ruff
+          run: |
+            pip install ruff
+            ruff check --format=github ./dags
+        - name: Test with Pytest
+          run: |
+            pip install pytest apache-airflow
+            cd dags || exit
+            pytest tests.py -v

README.md

@@ -99,6 +99,16 @@ We use PR reviews to approve or reject, comment on, and request further iteratio
   - LOCALFOLDER
 
 
+## Dags
+
+Dags that use the rikolti modules in Airflow are defined in the `dags` folder.
+
+A very basic level of testing is done on the dags to ensure tey load as expected. 
+
+### Continuous Deployment
+
+On each merge to `main`, an `AWS CodeBuild` task will run, triggered by a webhook emitted from github that the `CodeBuild` project listens for.
+
 ## Airflow Development
 
 ### Set up `aws-mwaa-local-runner`

dags/.airflowignore

@@ -0,0 +1,3 @@
+tests.py
+test_dag.py
+README.md

dags/test_dag.py

@@ -0,0 +1,64 @@
+# DAG exhibiting task flow paradigm in airflow 2.0
+# https://airflow.apache.org/docs/apache-airflow/2.0.2/tutorial_taskflow_api.html
+# Modified for our use case
+
+import json
+from airflow.decorators import dag, task
+from airflow.utils.dates import days_ago
+# These args will get passed on to each operator
+# You can override them on a per-task basis during operator initialization
+default_args = {
+    'owner': 'airflow',
+}
+@dag(
+        default_args=default_args, 
+        schedule_interval="@daily", 
+        start_date=days_ago(2), 
+        tags=['test'])
+def dag_with_taskflow_api():
+    """
+    ### TaskFlow API Tutorial Documentation
+    This is a simple ETL data pipeline example which demonstrates the use of
+    the TaskFlow API using three simple tasks for Extract, Transform, and Load.
+    Documentation that goes along with the Airflow TaskFlow API tutorial is
+    located
+    [here](https://airflow.apache.org/docs/stable/tutorial_taskflow_api.html)
+    """
+    @task()
+    def extract():
+        """
+        #### Extract task
+        A simple Extract task to get data ready for the rest of the data
+        pipeline. In this case, getting data is simulated by reading from a
+        hardcoded JSON string.
+        """
+        data_string = '{"1001": 301.27, "1002": 433.21, "1003": 502.22}'
+
+        order_data_dict = json.loads(data_string)
+        return order_data_dict
+    @task(multiple_outputs=True)
+    def transform(order_data_dict: dict):
+        """
+        #### Transform task
+        A simple Transform task which takes in the collection of order data and
+        computes the total order value.
+        """
+        total_order_value = 0
+
+        for value in order_data_dict.values():
+            total_order_value += value
+
+        return {"total_order_value": total_order_value}
+    @task()
+    def load(total_order_value: float):
+        """
+        #### Load task
+        A simple Load task which takes in the result of the Transform task and
+        instead of saving it to end user review, just prints it out.
+        """
+
+        print("Total order value is: %.2f" % total_order_value)
+    order_data = extract()
+    order_summary = transform(order_data)
+    load(order_summary["total_order_value"])
+dag_with_taskflow_api = dag_with_taskflow_api()

dags/tests.py

@@ -0,0 +1,14 @@
+from unittest import TestCase
+
+from airflow.models import DagBag
+
+
+DAGS_FOLDER = "."
+
+
+class HarvestDagsTest(TestCase):
+    def dag_bag(self):
+        return DagBag(dag_folder=DAGS_FOLDER, include_examples=False)
+
+    def test_no_import_errors(self):
+        assert not self.dag_bag().import_errors