MEP v8.0

.astro/config.yaml

@@ -0,0 +1,2 @@
+project:
+    name: sparte

.astro/dag_integrity_exceptions.txt

@@ -0,0 +1 @@
+# Add dag files to exempt from parse test below. ex: dags/<test-file>

.astro/test_dag_integrity_default.py

@@ -0,0 +1,130 @@
+"""Test the validity of all DAGs. **USED BY DEV PARSE COMMAND DO NOT EDIT**"""
+
+import logging
+import os
+from contextlib import contextmanager
+
+import pytest
+from airflow.hooks.base import BaseHook
+from airflow.models import Connection, DagBag, Variable
+from airflow.utils.db import initdb
+
+# init airflow database
+initdb()
+
+# The following code patches errors caused by missing OS Variables, Airflow Connections, and Airflow Variables
+
+
+# =========== MONKEYPATCH BaseHook.get_connection() ===========
+def basehook_get_connection_monkeypatch(key: str, *args, **kwargs):
+    print(f"Attempted to fetch connection during parse returning an empty Connection object for {key}")
+    return Connection(key)
+
+
+BaseHook.get_connection = basehook_get_connection_monkeypatch
+# # =========== /MONKEYPATCH BASEHOOK.GET_CONNECTION() ===========
+
+
+# =========== MONKEYPATCH OS.GETENV() ===========
+def os_getenv_monkeypatch(key: str, *args, **kwargs):
+    default = None
+    if args:
+        default = args[0]  # os.getenv should get at most 1 arg after the key
+    if kwargs:
+        default = kwargs.get("default", None)  # and sometimes kwarg if people are using the sig
+
+    env_value = os.environ.get(key, None)
+
+    if env_value:
+        return env_value  # if the env_value is set, return it
+    if key == "JENKINS_HOME" and default is None:  # fix https://github.com/astronomer/astro-cli/issues/601
+        return None
+    if default:
+        return default  # otherwise return whatever default has been passed
+    return f"MOCKED_{key.upper()}_VALUE"  # if absolutely nothing has been passed - return the mocked value
+
+
+os.getenv = os_getenv_monkeypatch
+# # =========== /MONKEYPATCH OS.GETENV() ===========
+
+# =========== MONKEYPATCH VARIABLE.GET() ===========
+
+
+class magic_dict(dict):
+    def __init__(self, *args, **kwargs):
+        self.update(*args, **kwargs)
+
+    def __getitem__(self, key):
+        return {}.get(key, "MOCKED_KEY_VALUE")
+
+
+_no_default = object()  # allow falsey defaults
+
+
+def variable_get_monkeypatch(key: str, default_var=_no_default, deserialize_json=False):
+    print(f"Attempted to get Variable value during parse, returning a mocked value for {key}")
+
+    if default_var is not _no_default:
+        return default_var
+    if deserialize_json:
+        return magic_dict()
+    return "NON_DEFAULT_MOCKED_VARIABLE_VALUE"
+
+
+Variable.get = variable_get_monkeypatch
+# # =========== /MONKEYPATCH VARIABLE.GET() ===========
+
+
+@contextmanager
+def suppress_logging(namespace):
+    """
+    Suppress logging within a specific namespace to keep tests "clean" during build
+    """
+    logger = logging.getLogger(namespace)
+    old_value = logger.disabled
+    logger.disabled = True
+    try:
+        yield
+    finally:
+        logger.disabled = old_value
+
+
+def get_import_errors():
+    """
+    Generate a tuple for import errors in the dag bag, and include DAGs without errors.
+    """
+    with suppress_logging("airflow"):
+        dag_bag = DagBag(include_examples=False)
+
+        def strip_path_prefix(path):
+            return os.path.relpath(path, os.environ.get("AIRFLOW_HOME"))
+
+        # Initialize an empty list to store the tuples
+        result = []
+
+        # Iterate over the items in import_errors
+        for k, v in dag_bag.import_errors.items():
+            result.append((strip_path_prefix(k), v.strip()))
+
+        # Check if there are DAGs without errors
+        for file_path in dag_bag.dags:
+            # Check if the file_path is not in import_errors, meaning no errors
+            if file_path not in dag_bag.import_errors:
+                result.append((strip_path_prefix(file_path), "No import errors"))
+
+        return result
+
+
+@pytest.mark.parametrize("rel_path, rv", get_import_errors(), ids=[x[0] for x in get_import_errors()])
+def test_file_imports(rel_path, rv):
+    """Test for import errors on a file"""
+    if os.path.exists(".astro/dag_integrity_exceptions.txt"):
+        with open(".astro/dag_integrity_exceptions.txt", "r") as f:
+            exceptions = f.readlines()
+    print(f"Exceptions: {exceptions}")
+    if (rv != "No import errors") and rel_path not in exceptions:
+        # If rv is not "No import errors," consider it a failed test
+        raise Exception(f"{rel_path} failed to import with message \n {rv}")
+    else:
+        # If rv is "No import errors," consider it a passed test
+        print(f"{rel_path} passed the import test")

.dockerignore

@@ -0,0 +1,8 @@
+astro
+.git
+.env
+airflow_settings.yaml
+logs/
+.venv
+airflow.db
+airflow.cfg

.github/workflows/deploy_airflow.yml

@@ -0,0 +1,19 @@
+name: Deploy to production Airflow
+on:
+  push:
+    branches:
+      - "staging"
+
+jobs:
+  deploy:
+    name: Deploy
+    runs-on: ubuntu-latest
+    steps:
+      - name: executing remote ssh git pull
+        uses: appleboy/ssh-action@master
+        with:
+          host: ${{ secrets.AIRFLOW_SSH_HOST }}
+          username: ${{ secrets.AIRFLOW_SSH_USER }}
+          key: ${{ secrets.AIRFLOW_SSH_KEY }}
+          port: ${{ secrets.AIRFLOW_SSH_PORT }}
+          script: cd ~/sparte && git pull

.github/workflows/pr.yml

@@ -36,7 +36,7 @@ jobs:
       - uses: actions/checkout@v4
       - uses: pre-commit/[email protected]
         env:
-          SKIP: ggshield
+          SKIP: ggshield,sqlfluff-lint,sqlfluff-fix
   test:
     runs-on: ubuntu-latest
     steps:

.pre-commit-config.yaml

@@ -27,11 +27,6 @@ repos:
   hooks:
     - id: black
       language_version: python3
-# - repo: https://github.com/pycqa/bandit
-#   rev: 1.7.0
-#   hooks:
-#     - id: bandit
-#       args: ['-iii', '-ll']
 -   repo: https://github.com/PyCQA/autoflake
     rev: v2.2.1
     hooks:
@@ -41,6 +36,19 @@ repos:
     hooks:
     -   id: flake8
         args: ['--config', 'flake8']
+-   repo: https://github.com/sqlfluff/sqlfluff
+    rev: 3.1.1
+    hooks:
+      - id: sqlfluff-lint
+        additional_dependencies: [
+          'dbt-postgres==1.8.2',
+          'sqlfluff-templater-dbt'
+        ]
+      - id: sqlfluff-fix
+        additional_dependencies: [
+          'dbt-postgres==1.8.2',
+          'sqlfluff-templater-dbt'
+        ]
 -   repo: https://github.com/gitguardian/ggshield
     rev: v1.24.0
     hooks:

.sqlfluff

@@ -0,0 +1,25 @@
+[sqlfluff]
+templater = dbt
+dialect = postgres
+
+[sqlfluff:templater:jinja]
+apply_dbt_builtins = True
+
+
+[sqlfluff:templater:dbt]
+project_dir = airflow/include/sql/sparte
+profiles_dir = ~/.dbt
+
+
+[sqlfluff:layout:type:alias_expression]
+# We want non-default spacing _before_ the alias expressions.
+spacing_before = align
+# We want to align them within the next outer select clause.
+# This means for example that alias expressions within the FROM
+# or JOIN clause would _not_ be aligned with them.
+align_within = select_clause
+# The point at which to stop searching outward for siblings, which
+# in this example would likely be the boundary of a CTE. Stopping
+# when we hit brackets is usually a good rule of thumb for this
+# configuration.
+align_scope = bracketed

airflow/.astro/config.yaml

@@ -0,0 +1,8 @@
+airflow:
+  expose_port: true
+project:
+    name: airflow
+postgres:
+    port: 5433
+webserver:
+    port: 9090

airflow/.astro/dag_integrity_exceptions.txt

@@ -0,0 +1 @@
+# Add dag files to exempt from parse test below. ex: dags/<test-file>

airflow/.astro/test_dag_integrity_default.py

@@ -0,0 +1,130 @@
+"""Test the validity of all DAGs. **USED BY DEV PARSE COMMAND DO NOT EDIT**"""
+
+import logging
+import os
+from contextlib import contextmanager
+
+import pytest
+from airflow.hooks.base import BaseHook
+from airflow.models import Connection, DagBag, Variable
+from airflow.utils.db import initdb
+
+# init airflow database
+initdb()
+
+# The following code patches errors caused by missing OS Variables, Airflow Connections, and Airflow Variables
+
+
+# =========== MONKEYPATCH BaseHook.get_connection() ===========
+def basehook_get_connection_monkeypatch(key: str, *args, **kwargs):
+    print(f"Attempted to fetch connection during parse returning an empty Connection object for {key}")
+    return Connection(key)
+
+
+BaseHook.get_connection = basehook_get_connection_monkeypatch
+# # =========== /MONKEYPATCH BASEHOOK.GET_CONNECTION() ===========
+
+
+# =========== MONKEYPATCH OS.GETENV() ===========
+def os_getenv_monkeypatch(key: str, *args, **kwargs):
+    default = None
+    if args:
+        default = args[0]  # os.getenv should get at most 1 arg after the key
+    if kwargs:
+        default = kwargs.get("default", None)  # and sometimes kwarg if people are using the sig
+
+    env_value = os.environ.get(key, None)
+
+    if env_value:
+        return env_value  # if the env_value is set, return it
+    if key == "JENKINS_HOME" and default is None:  # fix https://github.com/astronomer/astro-cli/issues/601
+        return None
+    if default:
+        return default  # otherwise return whatever default has been passed
+    return f"MOCKED_{key.upper()}_VALUE"  # if absolutely nothing has been passed - return the mocked value
+
+
+os.getenv = os_getenv_monkeypatch
+# # =========== /MONKEYPATCH OS.GETENV() ===========
+
+# =========== MONKEYPATCH VARIABLE.GET() ===========
+
+
+class magic_dict(dict):
+    def __init__(self, *args, **kwargs):
+        self.update(*args, **kwargs)
+
+    def __getitem__(self, key):
+        return {}.get(key, "MOCKED_KEY_VALUE")
+
+
+_no_default = object()  # allow falsey defaults
+
+
+def variable_get_monkeypatch(key: str, default_var=_no_default, deserialize_json=False):
+    print(f"Attempted to get Variable value during parse, returning a mocked value for {key}")
+
+    if default_var is not _no_default:
+        return default_var
+    if deserialize_json:
+        return magic_dict()
+    return "NON_DEFAULT_MOCKED_VARIABLE_VALUE"
+
+
+Variable.get = variable_get_monkeypatch
+# # =========== /MONKEYPATCH VARIABLE.GET() ===========
+
+
+@contextmanager
+def suppress_logging(namespace):
+    """
+    Suppress logging within a specific namespace to keep tests "clean" during build
+    """
+    logger = logging.getLogger(namespace)
+    old_value = logger.disabled
+    logger.disabled = True
+    try:
+        yield
+    finally:
+        logger.disabled = old_value
+
+
+def get_import_errors():
+    """
+    Generate a tuple for import errors in the dag bag, and include DAGs without errors.
+    """
+    with suppress_logging("airflow"):
+        dag_bag = DagBag(include_examples=False)
+
+        def strip_path_prefix(path):
+            return os.path.relpath(path, os.environ.get("AIRFLOW_HOME"))
+
+        # Initialize an empty list to store the tuples
+        result = []
+
+        # Iterate over the items in import_errors
+        for k, v in dag_bag.import_errors.items():
+            result.append((strip_path_prefix(k), v.strip()))
+
+        # Check if there are DAGs without errors
+        for file_path in dag_bag.dags:
+            # Check if the file_path is not in import_errors, meaning no errors
+            if file_path not in dag_bag.import_errors:
+                result.append((strip_path_prefix(file_path), "No import errors"))
+
+        return result
+
+
+@pytest.mark.parametrize("rel_path, rv", get_import_errors(), ids=[x[0] for x in get_import_errors()])
+def test_file_imports(rel_path, rv):
+    """Test for import errors on a file"""
+    if os.path.exists(".astro/dag_integrity_exceptions.txt"):
+        with open(".astro/dag_integrity_exceptions.txt", "r") as f:
+            exceptions = f.readlines()
+    print(f"Exceptions: {exceptions}")
+    if (rv != "No import errors") and rel_path not in exceptions:
+        # If rv is not "No import errors," consider it a failed test
+        raise Exception(f"{rel_path} failed to import with message \n {rv}")
+    else:
+        # If rv is "No import errors," consider it a passed test
+        print(f"{rel_path} passed the import test")

airflow/.dockerignore

@@ -0,0 +1,8 @@
+astro
+.git
+.env
+airflow_settings.yaml
+logs/
+.venv
+airflow.db
+airflow.cfg

airflow/.gitignore

@@ -0,0 +1,11 @@
+.git
+.env
+.DS_Store
+airflow_settings.yaml
+__pycache__/
+astro
+.venv
+airflow-webserver.pid
+webserver_config.py
+airflow.cfg
+airflow.db

airflow/Dockerfile

@@ -0,0 +1,3 @@
+FROM quay.io/astronomer/astro-runtime:11.8.0
+RUN mkdir /home/astro/.dbt
+COPY ./dbt_profile.yml /home/astro/.dbt/profiles.yml