sqlmesh-dag-generator 0.9.0

Open-source Airflow DAG generator for SQLMesh projects

SQLMesh DAG Generator

Generate Apache Airflow DAGs from SQLMesh projects - no cloud dependencies required.

Transform your SQLMesh models into production-ready Airflow DAGs with full data lineage, automatically!

✨ Key Features

• 🔥 Dynamic DAG Generation (Default): Fire-and-forget - place DAG once, auto-discovers models at runtime
• 📅 Auto-Scheduling: Automatically detects DAG schedule from SQLMesh model intervals - no manual configuration!
• 🔐 Runtime Connection Parametrization: Pass database credentials via Airflow Connections - no hardcoded secrets!
• ✅ Full Lineage in Airflow: Each SQLMesh model = One Airflow task with proper dependencies
• 🌍 Multi-Environment Support: Use Airflow Variables + SQLMesh gateways for dev/staging/prod
• ⚡ Incremental Models: Proper handling with data_interval_start/end
• 🎯 Enhanced Error Handling: SQLMesh-specific error messages in Airflow logs
• 🛠️ Dual Mode: Dynamic (auto-discovery, default) or Static (full control)
• 🚫 No Vendor Lock-in: Open source, no cloud dependencies

🏢 Enterprise Features (NEW in v0.8.0)

• 🔔 Callbacks: on_failure_callback, on_success_callback, sla_miss_callback for alerting
• 🏷️ Tag-Based Filtering: include_tags, exclude_tags for Data Mesh team-specific DAGs
• 🎱 Pool Configuration: pool, pool_slots for resource management
• ⏩ Trigger Downstream DAGs: trigger_dag_id for ML pipeline integration
• 🎯 Pattern Filtering: model_pattern for regex-based model selection

⚠️ Important: Gateway vs Environment

SQLMesh uses "gateways" to switch between environments, NOT an "environment" parameter.

# ❌ WRONG - environment parameter is for SQLMesh virtual environments (testing)
generator = SQLMeshDAGGenerator(
    sqlmesh_project_path="/path/to/project",
    environment="prod",  # ERROR: Environment 'prod' was not found
)

# ✅ CORRECT - Use gateway to switch between dev/staging/prod
generator = SQLMeshDAGGenerator(
    sqlmesh_project_path="/path/to/project",
    gateway="prod",  # This is how you select your environment!
    # environment defaults to "" (empty string) - perfect for production
)

📚 Read Understanding SQLMesh Environments for the complete explanation of the difference between gateway and environment.

🚀 Quick Start (3 Steps)

1. Install

pip install sqlmesh-dag-generator  # (when published)
# OR
git clone <repo> && cd SQLMeshDAGGenerator && pip install -e .

2. Generate DAG (Dynamic Mode - Default!)

from sqlmesh_dag_generator import SQLMeshDAGGenerator

# Point to your SQLMesh project
generator = SQLMeshDAGGenerator(
    sqlmesh_project_path="/path/to/your/sqlmesh/project",
    dag_id="my_pipeline",
    schedule_interval="@daily"
)

# Generate dynamic DAG (default - fire and forget!)
dag_code = generator.generate_dynamic_dag()

# Save it
with open("my_pipeline.py", "w") as f:
    f.write(dag_code)

3. Deploy to Airflow

cp my_pipeline.py /opt/airflow/dags/

That's it! 🎉 Your SQLMesh models are now orchestrated by Airflow. The DAG will auto-discover models at runtime - no regeneration needed when models change!

💡 What You Get

Your SQLMesh Project:

my_project/
└── models/
    ├── raw_orders.sql
    ├── stg_orders.sql      # depends on raw_orders
    └── orders_summary.sql  # depends on stg_orders

Generated Airflow DAG:

Airflow Graph View:
  [raw_orders] → [stg_orders] → [orders_summary]
  
✅ Each model = separate task
✅ SQLMesh dependencies = Airflow dependencies  
✅ Full lineage visible in Airflow UI

📚 Documentation

Getting Started

• Quick Start Guide - Step-by-step tutorial (start here!)
• Quick Reference - One-page cheat sheet
• Examples - Code examples

Configuration & Features

• Auto-Scheduling Guide - Automatic schedule detection 📅
• Runtime Configuration - Pass credentials via Airflow Connections 🔐
• Multi-Environment Setup - Configure for dev/staging/prod ⚠️ IMPORTANT
• Dynamic DAGs - Fire-and-forget mode explained

Deployment & Production

• Deployment Checklist - Step-by-step deployment guide ✅ NEW!
• Container Deployment - Docker, Kubernetes, ECS, Cloud Composer 🐳 NEW!
• Common Issues - Real-world problems and solutions 🔧 NEW!
• Troubleshooting - Debug guide
• Deployment Warnings - Critical production considerations

Reference

• Usage Guide - Complete reference
• Migration Guide - Fix common configuration issues
• Architecture - Technical details

🔥 Why Dynamic Mode (Default)?

Dynamic mode auto-discovers SQLMesh models at runtime:

dag_code = generator.generate_dynamic_dag()  # Default behavior!

Benefits:

• ✅ No regeneration needed when SQLMesh models change
• ✅ Always in sync - DAG updates automatically
• ✅ Multi-environment - Uses Airflow Variables
• ✅ Production-ready - Enhanced error handling

Want static mode instead? Just use generator.generate_dag() - see Usage Guide.

🎯 Simple Example

The simplest possible usage - just 3 lines of code:

from sqlmesh_dag_generator import SQLMeshDAGGenerator

generator = SQLMeshDAGGenerator(
    sqlmesh_project_path="/path/to/your/sqlmesh/project",
    dag_id="my_pipeline"
)

dag_code = generator.generate_dynamic_dag()

See examples/simple_generate.py for a complete runnable example.

🤝 Contributing

Contributions welcome! See CONTRIBUTING.md for guidelines.

📄 License

[Your License Here]

---

Built with ❤️ for the data engineering community

Configuration File

Create a dag_generator_config.yaml:

sqlmesh:
  project_path: "/path/to/sqlmesh/project"
  gateway: "prod"
  environment: ""  # Empty for production
  default_catalog: "my_catalog" # Optional: For 3-part naming support

airflow:
  dag_id: "sqlmesh_pipeline"
  schedule_interval: "0 0 * * *"
  default_args:
    owner: "data-team"
    retries: 3
    retry_delay_minutes: 5
  tags:
    - sqlmesh
    - analytics

generation:
  output_dir: "/path/to/airflow/dags"
  operator_type: "python"  # or "bash"
  include_tests: true
  parallel_tasks: true
  auto_replan_on_change: true  # Automatically run 'sqlmesh plan' if models change
  # Plan optimization options (for faster execution when no changes)
  skip_backfill: false   # Skip apply if backfill is required (use CI/CD for initial backfills)
  plan_only: false       # Generate plan without applying (for review/dry-run)
  log_plan_details: true # Log detailed plan information (context, plan, apply phases)
  return_value: true     # Enable/disable XCom return values

How It Works

1. Load SQLMesh Project: Reads your SQLMesh project configuration and models
2. Extract Dependencies: Analyzes SQL queries to build dependency graph
3. Generate Tasks: Creates Airflow tasks for each SQLMesh model
4. Set Dependencies: Connects tasks based on model dependencies
5. Apply Schedules: Preserves cron schedules and execution logic
6. Output DAG: Generates Python file ready for Airflow

Architecture

SQLMesh Project
    ↓
SQLMeshDAGGenerator
    ├── Context Loader (loads SQLMesh context)
    ├── Model Parser (extracts model metadata)
    ├── Dependency Resolver (builds dependency graph)
    └── DAG Builder (generates Airflow DAG)
    ↓
Airflow DAG File

Advanced Features

Custom Operators

from sqlmesh_dag_generator import SQLMeshDAGGenerator
from airflow.operators.python import PythonOperator

generator = SQLMeshDAGGenerator(
    sqlmesh_project_path="/path/to/project",
    custom_operator_class=PythonOperator,
    operator_kwargs={"provide_context": True}
)

Model Filtering

# Generate DAG for specific models only
generator = SQLMeshDAGGenerator(
    sqlmesh_project_path="/path/to/project",
    include_models=["model1", "model2"],
    exclude_models=["test_*"]
)

Dynamic Task Generation

# Generate tasks with dynamic parallelism
generator = SQLMeshDAGGenerator(
    sqlmesh_project_path="/path/to/project",
    enable_dynamic_tasks=True,
    max_parallel_tasks=10
)

⚠️ Important: Deployment Warnings

🔴 Distributed Airflow Requires Shared Volume

If you're using KubernetesExecutor, CeleryExecutor, or any distributed Airflow setup:

Your SQLMesh project MUST be accessible to all workers!

Solutions:

• Option 1 (Recommended): Mount project on shared volume (EFS/NFS/Filestore)
• Option 2: Bake project into Docker image (loses fire-and-forget benefit)

See full guide: docs/DEPLOYMENT_WARNINGS.md

🟡 Operator Type Limitations

• Dynamic Mode: Python operator only (current limitation)
• Static Mode: Supports Python, Bash, and Kubernetes operators

For Bash/Kubernetes in dynamic mode, use static generation for now.

🟢 Kubernetes Operator Support

To use operator_type: kubernetes:

generation:
  operator_type: kubernetes
  docker_image: "your-registry/sqlmesh:v1.0"  # REQUIRED
  namespace: "data-pipelines"

📖 Full Documentation: docs/DEPLOYMENT_WARNINGS.md

Requirements

• Python >= 3.8
• Apache Airflow >= 2.0
• SQLMesh >= 0.20.0

Development

# Clone the repository
git clone https://github.com/yourusername/sqlmesh-dag-generator.git
cd sqlmesh-dag-generator

# Install in development mode
pip install -e ".[dev]"

# Run tests
pytest

# Run linter
black .
ruff check .

🔧 Troubleshooting

Common Issues

1. TypeError: Object of type CompletionStatus is not JSON serializable

Cause: Airflow cannot serialize the SQLMesh CompletionStatus enum returned by tasks. Fix: Upgrade to v0.7.2+. This version automatically converts the status to a string. Alternatively, use return_value=False in your generator config:

generator = SQLMeshDAGGenerator(
    ...,
    return_value=False  # Disable XCom return values
)

2. ConfigError: A query is required and must be a SELECT statement (External Models)

Cause: Defining external models in .sql files is not supported by the generator. Fix: Define external models in external_models.yaml instead.

# external_models.yaml
models:
  - name: raw.users
    kind: EXTERNAL
    columns:
      id: INT
      name: TEXT

3. "Environment 'prod' was not found"

Cause: You are using environment="prod" instead of gateway="prod". Fix: Set gateway="prod" and leave environment as empty string (default).

generator = SQLMeshDAGGenerator(
    ...,
    gateway="prod",
    environment=""  # Correct for production
)

4. Changes to models are not picked up

Cause: SQLMesh requires a plan/apply step to register changes. Fix: Enable auto_replan_on_change in your generator config:

generator = SQLMeshDAGGenerator(
    ...,
    auto_replan_on_change=True  # Automatically runs 'sqlmesh plan --auto-apply'
)

Or run sqlmesh plan --auto-apply manually.

5. sqlmesh_plan_apply task takes 7+ minutes even with no changes

Cause: When auto_replan_on_change=True, the plan task may run full backfills even when there are no model changes. Fix (v0.9.0+): The plan optimization now:

• ✅ Detects "no changes" scenarios and skips apply (~15 seconds instead of 7+ minutes)
• ✅ Logs detailed timing for each phase (context load, plan compute, apply)

For additional control, use these options:

generator = SQLMeshDAGGenerator(
    ...,
    auto_replan_on_change=True,
    skip_backfill=True,    # Skip apply if backfill is required (use CI/CD for initial backfills)
    plan_only=True,        # Generate plan without applying (for review/dry-run)
    log_plan_details=True  # Log detailed plan information (default: True)
)

Recommended CI/CD approach:

• Run sqlmesh plan --auto-apply in CI/CD on model changes (handles backfills)
• Set skip_backfill=True in Airflow DAG (only applies non-backfill changes)
• This gives you fast DAG runs (~15s) while ensuring changes are still applied

5. 3-Part Table Naming Issues (Cross-Database)

Cause: SQLMesh generates queries with 3-part names (catalog.schema.table) which might fail if the catalog name doesn't match. Fix: Set default_catalog in your config to tell SQLMesh which catalog to omit or use.

generator = SQLMeshDAGGenerator(
    ...,
    default_catalog="my_catalog"
)

Advanced Features

Health Check Task

Add a pre-flight check to verify database connectivity before running models:

generator = SQLMeshDAGGenerator(
    ...,
    enable_health_check=True
)

Skip Audits

For faster development iterations, you can skip audit checks:

generator = SQLMeshDAGGenerator(
    ...,
    skip_audits=True
)

Partial DAG Runs

You can generate tasks for a subset of models (useful for testing):

# In your DAG file
generator.create_tasks_in_dag(dag, models=["model_a", "model_b"])

Contributing

Contributions are welcome! Please read our Contributing Guide for details.

License

MIT License - see LICENSE file for details.

Comparison with Tobiko Cloud

Feature	Tobiko Cloud	SQLMesh DAG Generator
Cost	Paid	Free & Open Source
Deployment	Cloud-based	Self-hosted
Customization	Limited	Fully Customizable
Privacy	External	On-premise
Dependencies	Cloud connection	None

Support

• 📖 Documentation
• 🐛 Issue Tracker
• 💬 Discussions