Sync back

.github/workflows/ci-data.yml

@@ -17,10 +17,10 @@ jobs:
         with:
           persist-credentials: false
 
-      - name: Set up Python 3.7
+      - name: Set up Python 3.12
         uses: actions/setup-python@v2
         with:
-          python-version: 3.7
+          python-version: 3.12
 
       - name: Install dependencies
         run: |

.github/workflows/claude.yml

@@ -0,0 +1,49 @@
+name: Claude Code
+
+on:
+  issue_comment:
+    types: [created]
+  pull_request_review_comment:
+    types: [created]
+  issues:
+    types: [opened, assigned]
+  pull_request_review:
+    types: [submitted]
+
+jobs:
+  claude:
+    if: |
+      (github.event_name == 'issue_comment' && contains(github.event.comment.body, '@claude')) ||
+      (github.event_name == 'pull_request_review_comment' && contains(github.event.comment.body, '@claude')) ||
+      (github.event_name == 'pull_request_review' && contains(github.event.review.body, '@claude')) ||
+      (github.event_name == 'issues' && (contains(github.event.issue.body, '@claude') || contains(github.event.issue.title, '@claude')))
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      pull-requests: read
+      issues: read
+      id-token: write
+      actions: read # Required for Claude to read CI results on PRs
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 1
+
+      - name: Run Claude Code
+        id: claude
+        uses: anthropics/claude-code-action@v1
+        with:
+          anthropic_api_key: ${{ secrets.ANTHROPIC_API_KEY }}
+
+          # This is an optional setting that allows Claude to read CI results on PRs
+          additional_permissions: |
+            actions: read
+
+          # Optional: Give a custom prompt to Claude. If this is not specified, Claude will perform the instructions specified in the comment that tagged it.
+          # prompt: 'Update the pull request description to include a summary of changes.'
+
+          # Optional: Add claude_args to customize behavior and configuration
+          # See https://github.com/anthropics/claude-code-action/blob/main/docs/usage.md
+          # or https://code.claude.com/docs/en/cli-reference for available options
+          # claude_args: '--allowed-tools Bash(gh pr:*)'

CLAUDE.md

@@ -0,0 +1,98 @@
+# CLAUDE.md
+
+This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
+
+## Project Overview
+
+Dagger is a configuration-driven framework that transforms YAML definitions into Apache Airflow DAGs. It uses dataset lineage (matching inputs/outputs) to automatically build dependency graphs across workflows.
+
+## Common Commands
+
+### Development Setup
+```bash
+make install-dev      # Create venv, install package in editable mode with dev/test deps
+source venv/bin/activate
+```
+
+### Testing
+```bash
+make test             # Run all tests with coverage (sets AIRFLOW_HOME automatically)
+
+# Run a single test file
+AIRFLOW_HOME=$(pwd)/tests/fixtures/config_finder/root/ ENV=local pytest -s tests/path/to/test_file.py
+
+# Run a specific test
+AIRFLOW_HOME=$(pwd)/tests/fixtures/config_finder/root/ ENV=local pytest -s tests/path/to/test_file.py::test_function_name
+```
+
+### Linting
+```bash
+make lint             # Run flake8 on dagger and tests directories
+black dagger tests    # Format code
+```
+
+### Local Airflow Testing
+```bash
+make test-airflow     # Build and start Airflow in Docker (localhost:8080, user: dev_user, pass: dev_user)
+make stop-airflow     # Stop Airflow containers
+```
+
+### CLI
+```bash
+dagger --help
+dagger list-tasks     # Show available task types
+dagger list-ios       # Show available IO types
+dagger init-pipeline  # Create a new pipeline.yaml
+dagger init-task --type=<task_type>  # Add a task configuration
+dagger init-io --type=<io_type>      # Add an IO definition
+dagger print-graph    # Visualize dependency graph
+```
+
+## Architecture
+
+### Core Flow
+1. **ConfigFinder** discovers pipeline directories (each with `pipeline.yaml` + task YAML files)
+2. **ConfigProcessor** loads YAML configs with environment variable support
+3. **TaskFactory/IOFactory** use reflection to instantiate task/IO objects from YAML
+4. **TaskGraph** builds a 3-layer graph: Pipeline → Task → Dataset nodes
+5. **DagCreator** traverses the graph and generates Airflow DAGs using **OperatorFactory**
+
+### Key Directories
+- `dagger/pipeline/tasks/` - Task type definitions (DbtTask, SparkTask, AthenaTransformTask, etc.)
+- `dagger/pipeline/ios/` - IO type definitions (S3, Redshift, Athena, Databricks, etc.)
+- `dagger/dag_creator/airflow/operator_creators/` - One creator per task type, translates tasks to Airflow operators
+- `dagger/graph/` - Graph construction from task inputs/outputs
+- `dagger/config_finder/` - YAML discovery and loading
+- `tests/fixtures/config_finder/root/dags/` - Example DAG configurations for testing
+
+### Adding a New Task Type
+1. Create task definition in `dagger/pipeline/tasks/` (subclass of Task)
+2. Create any needed IOs in `dagger/pipeline/ios/` (if new data sources)
+3. Create operator creator in `dagger/dag_creator/airflow/operator_creators/`
+4. Register in `dagger/dag_creator/airflow/operator_factory.py`
+
+### Configuration Files
+- `pipeline.yaml` - Pipeline metadata (owner, schedule, alerts, airflow_parameters)
+- `[taskname].yaml` - Task configs (type, inputs, outputs, task-specific params)
+- `dagger_config.yaml` - System config (Neo4j, Elasticsearch, Spark settings)
+
+### Key Patterns
+- **Factory Pattern**: TaskFactory/IOFactory auto-discover types via reflection
+- **Strategy Pattern**: OperatorCreator subclasses handle task-specific operator creation
+- **Dataset Aliasing**: IO `alias()` method enables automatic dependency detection across pipelines
+
+## Coding Standards
+
+### Avoid getattr
+Do not use `getattr` for accessing task or IO properties. Instead, define explicit properties on the class. This ensures:
+- Type safety and IDE autocompletion
+- Clear interface contracts
+- Easier debugging and testing
+
+```python
+# Bad - avoid this pattern
+value = getattr(self._task, 'some_property', default)
+
+# Good - use explicit properties
+value = self._task.some_property  # Property defined on task class
+```

Makefile

@@ -96,17 +96,16 @@ 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.12 venv; \
 	source venv/bin/activate; \
-	python -m pip install --upgrade pip; \
-	python setup.py install; \
+	python -m pip install --upgrade pip setuptools wheel; \
 	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.12 venv; \
 	source venv/bin/activate; \
-	python -m pip install --upgrade pip; \
+	python -m pip install --upgrade pip setuptools wheel; \
 	pip install -r reqs/test.txt -r reqs/base.txt
 
 install-ui: clean ## install the package to the active Python's site-packages

README.md

@@ -87,6 +87,41 @@ flowchart TD;
 
 ```
 
+Plugins for dagger
+-------
+
+### Overview
+Dagger now supports a plugin system that allows users to extend its functionality by adding custom Python classes. These plugins are integrated into the Jinja2 templating engine, enabling dynamic rendering of task configuration templates.
+### Purpose
+The plugin system allows users to define Python classes that can be loaded into the Jinja2 environment. When functions from these classes are invoked within a task configuration template, they are rendered dynamically using Jinja2. This feature enhances the flexibility of task configurations by allowing custom logic to be embedded directly in the templates.
+
+### Usage
+1. **Creating a Plugin:** To create a new plugin, define a Python class in a folder(for example `plugins/sample_plugin/sample_plugin.py`) with the desired methods. For example:
+```python
+class MyCustomPlugin:
+    def generate_input(self, branch_name):
+        return [{"name": f"{branch_name}", "type": "dummy"}]
+```
+This class defines a `generate_input` method that takes the branch_name from the module config and returns a dummy dagger task.
+
+2. **Loading the Plugin into Dagger:** To load this plugin into Dagger's Jinja2 environment, you need to register it in your `dagger_config.yaml`:
+```yaml
+# pipeline.yaml
+plugin:
+  paths: 
+    - plugins # all Python classes within this path will be loaded into the Jinja environment
+```
+
+3. **Using Plugin Methods in Templates:** Once the plugin is loaded, you can call its methods from within any Jinja2 template in your task configurations:
+```yaml
+# task_configuration.yaml
+type: batch
+description: sample task
+inputs:                        # format: list | Use dagger init-io cli
+  {{ MyCustomPlugin.generate_input("dummy_input") }}
+```
+
+
 
 Credits
 -------

dagger/cli/module.py

@@ -1,17 +1,49 @@
+import json
+
 import click
+import yaml
+
 from dagger.utilities.module import Module
 from dagger.utils import Printer
 
 
+def parse_key_value(ctx, param, value):
+    """Parse key=value pairs where value is a path to JSON or YAML file.
+
+    Args:
+        ctx: Click context.
+        param: Click parameter.
+        value: List of key=value pairs.
+
+    Returns:
+        Dictionary mapping variable names to parsed file contents.
+    """
+    if not value:
+        return {}
+    key_value_dict = {}
+    for pair in value:
+        try:
+            key, val_file_path = pair.split('=', 1)
+            with open(val_file_path, 'r') as f:
+                if val_file_path.endswith(('.yaml', '.yml')):
+                    val = yaml.safe_load(f)
+                else:
+                    val = json.load(f)
+            key_value_dict[key] = val
+        except ValueError:
+            raise click.BadParameter(f"Key-value pair '{pair}' is not in the format key=value")
+    return key_value_dict
+
 @click.command()
 @click.option("--config_file", "-c", help="Path to module config file")
 @click.option("--target_dir", "-t", help="Path to directory to generate the task configs to")
-def generate_tasks(config_file: str, target_dir: str) -> None:
+@click.option("--jinja_parameters", "-j", callback=parse_key_value, multiple=True, default=None, help="Jinja parameters file in the format: <var_name>=<path to json/yaml file>")
+def generate_tasks(config_file: str, target_dir: str, jinja_parameters: dict) -> None:
     """
     Generating tasks for a module based on config
     """
 
-    module = Module(config_file, target_dir)
+    module = Module(config_file, target_dir, jinja_parameters)
     module.generate_task_configs()
 
     Printer.print_success("Tasks are successfully generated")

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
@@ -100,4 +98,23 @@
 # Alert parameters
 alert_config = config.get('alert', None) or {}
 SLACK_TOKEN = alert_config.get('slack_token', None)
-DEFAULT_ALERT = alert_config.get('default_alert', {"type": "slack", "channel": "#airflow-jobs", "mentions": None})
+DEFAULT_ALERT = alert_config.get('default_alert', {"type": "slack", "channel": "#airflow-jobs", "mentions": None})
+
+# Plugin parameters
+plugin_config = config.get('plugin', None) or {}
+PLUGIN_DIRS = [os.path.join(AIRFLOW_HOME, path) for path in plugin_config.get('paths', [])]
+logging.info(f"All Python classes will be loaded as plugins from the following directories: {PLUGIN_DIRS}")
+
+# ReverseETL parameters
+reverse_etl_config = config.get('reverse_etl', None) or {}
+REVERSE_ETL_DEFAULT_JOB_NAME = reverse_etl_config.get('default_job_name', None)
+REVERSE_ETL_DEFAULT_EXECUTABLE_PREFIX = reverse_etl_config.get('default_executable_prefix', None)
+REVERSE_ETL_DEFAULT_EXECUTABLE = reverse_etl_config.get('default_executable', None)
+
+# Soda parameters
+soda_config = config.get('soda', None) or {}
+SODA_DEFAULT_JOB_NAME = soda_config.get('default_job_name', None)
+SODA_DEFAULT_EXECUTABLE_PREFIX = soda_config.get('default_executable_prefix', None)
+SODA_DEFAULT_EXECUTABLE = soda_config.get('default_executable', None)
+SODA_DEFAULT_OUTPUT_TABLE = soda_config.get('default_output_table', None)
+SODA_DEFAULT_OUTPUT_S3_PATH = soda_config.get('default_output_s3_path', None)

dagger/dag_creator/airflow/dag_creator.py

@@ -58,7 +58,7 @@ def _get_external_task_sensor_name_dict(self, from_task_id: str) -> dict:
             "external_sensor_name": f"{from_pipeline_name}-{from_task_name}-sensor",
         }
 
-    def _get_external_task_sensor(self, from_task_id: str, to_task_id: str) -> ExternalTaskSensor:
+    def _get_external_task_sensor(self, from_task_id: str, to_task_id: str, follow_external_dependency: dict) -> ExternalTaskSensor:
         """
         create an object of external task sensor for a specific from_task_id and to_task_id
         """
@@ -72,6 +72,9 @@ def _get_external_task_sensor(self, from_task_id: str, to_task_id: str) -> Exter
 
         to_pipe_id = self._task_graph.get_node(to_task_id).obj.pipeline.name
 
+        extra_args = conf.EXTERNAL_SENSOR_DEFAULT_ARGS.copy()
+        extra_args.update(follow_external_dependency)
+
         return ExternalTaskSensor(
             dag=self._dags[to_pipe_id],
             task_id=external_sensor_name,
@@ -80,9 +83,7 @@ def _get_external_task_sensor(self, from_task_id: str, to_task_id: str) -> Exter
             execution_date_fn=self._get_execution_date_fn(
                 from_pipeline_schedule, to_pipeline_schedule
             ),
-            mode=conf.EXTERNAL_SENSOR_MODE,
-            poke_interval=conf.EXTERNAL_SENSOR_POKE_INTERVAL,
-            timeout=conf.EXTERNAL_SENSOR_TIMEOUT,
+            **extra_args
         )
 
     def _create_control_flow_task(self, pipe_id, dag):
@@ -135,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
         )
@@ -143,7 +145,7 @@ def _create_edge_without_data(self, from_task_id: str, to_task_ids: list, node:
             to_pipe = self._task_graph.get_node(to_task_id).obj.pipeline_name
             if from_pipe and from_pipe == to_pipe:
                 self._tasks[from_task_id] >> self._tasks[to_task_id]
-            elif from_pipe and from_pipe != to_pipe and edge_properties.follow_external_dependency:
+            elif from_pipe and from_pipe != to_pipe and edge_properties.follow_external_dependency is not None:
                 from_schedule = self._task_graph.get_node(from_task_id).obj.pipeline.schedule
                 to_schedule = self._task_graph.get_node(to_task_id).obj.pipeline.schedule
                 if not from_schedule.startswith("@") and not to_schedule.startswith("@"):
@@ -155,15 +157,17 @@ def _create_edge_without_data(self, from_task_id: str, to_task_ids: list, node:
                         not in self._sensor_dict.get(to_pipe, dict()).keys()
                     ):
                         external_task_sensor = self._get_external_task_sensor(
-                            from_task_id, to_task_id
+                            from_task_id, to_task_id, edge_properties.follow_external_dependency
                         )
-                        self._sensor_dict[to_pipe] = {
+
+                        if self._sensor_dict.get(to_pipe) is None:
+                            self._sensor_dict[to_pipe] = {}
+
+                        self._sensor_dict[to_pipe].update({
                             external_task_sensor_name: external_task_sensor
-                        }
-                        (
-                            self._tasks[self._get_control_flow_task_id(to_pipe)]
-                            >> external_task_sensor
-                        )
+                        })
+
+                        self._tasks[self._get_control_flow_task_id(to_pipe)] >> external_task_sensor
                     self._sensor_dict[to_pipe][external_task_sensor_name] >> self._tasks[to_task_id]
             else:
                 self._tasks[self._get_control_flow_task_id(to_pipe)] >> self._tasks[to_task_id]