dagctl

CLI tool for managing SQLMesh projects with dagctl-hosted state databases.

Features

• 🔐 Secure Authentication - OAuth-based authentication with JWT tokens
• 🔄 Auto-Refresh - JWT tokens automatically refresh when expired
• 🚀 Zero Config - SQLMesh state connection configured automatically
• 🏢 Multi-Org - Support for multiple organizations and projects
• 🔒 Secure - No long-lived credentials, JWT-based database authentication

Installation

pip install dagctl

Quick Start

# 1. Authenticate
dagctl auth login --org your-org

# 2. Set project context
dagctl use-project my-project

# 3. Run SQLMesh
cd my-sqlmesh-project
sqlmesh plan

How It Works

dagctl provides a Python function get_state_connection() that SQLMesh calls to get state database credentials. This function:

1. Reads your authentication from ~/.dagctl/
2. Automatically refreshes JWT tokens if expired
3. Returns PostgreSQL connection details for the dagctl-hosted state database
4. Uses JWT tokens for passwordless authentication via pg-proxy

Usage

1. Authentication

Authenticate to your dagctl organization via browser:

dagctl auth login --org acme

# Output:
# dagctl Authentication
# 
#   Opening browser for authentication...
#   Listening on http://localhost:8080
# 
#   Waiting for authentication... ✓
#   Exchanging code for tokens... ✓
#   Verifying organization membership... ✓
#   Fetching environment configuration... ✓
# 
# ✓ Authenticated as user@company.com
# ✓ Organization: acme (ID: 4710a6f2-dbcf-4966-9427-b91784327d2d)
# 
# Next step: dagctl use-project <project-name>

For development environments with self-signed certificates:

dagctl auth login --org acme --api-url https://api.dagctl.internal -k

2. Set Project Context

Tell dagctl which project you're working on:

dagctl use-project my-project

# Output:
# Setting project context: my-project
# 
#   Verifying project exists... ✓
# 
# ✓ Project set: my-project
# 
# Next step:
#   dagctl config generate

For development:

dagctl use-project snowflake -k

3. Generate SQLMesh Config

If you have a config.yaml, generate a Python config:

dagctl config generate

This converts your config.yaml to config.py with automatic state connection:

Before (config.yaml):

gateways:
  snowflake:
    connection:
      type: snowflake
      account: abc123.us-east-1
      user: "${SNOWFLAKE_USER}"
      password: "${SNOWFLAKE_PASSWORD}"

After (config.py):

from dagctl import get_state_connection
from sqlmesh.core.config import Config

gateways = {
    "snowflake": {
        "connection": {
            "type": "snowflake",
            "account": os.environ.get("SNOWFLAKE_ACCOUNT"),
            "user": os.environ.get("SNOWFLAKE_USER"),
            # ...
        },
        "state_connection": get_state_connection(),  # ← Auto-configured!
    },
}

config = Config(
    gateways=gateways,
    default_gateway="snowflake",
)

4. Use SQLMesh

Just run SQLMesh normally! The state connection is handled automatically:

sqlmesh plan
sqlmesh run
sqlmesh fetchdf "SELECT * FROM my_model"

Manual Config (No config.yaml)

If you don't have a config.yaml, create config.py manually:

"""SQLMesh configuration with dagctl state connection."""

import os
from dagctl import get_state_connection
from sqlmesh.core.config import Config

gateways = {
    "my_gateway": {
        "connection": {
            "type": "snowflake",
            "account": os.environ.get("SNOWFLAKE_ACCOUNT"),
            "user": os.environ.get("SNOWFLAKE_USER"),
            "authenticator": "snowflake_jwt",
            "private_key_path": os.environ.get("SNOWFLAKE_PRIVATE_KEY_PATH"),
            "warehouse": os.environ.get("SNOWFLAKE_WAREHOUSE"),
            "database": os.environ.get("SNOWFLAKE_DATABASE"),
            "role": os.environ.get("SNOWFLAKE_ROLE"),
        },
        "state_connection": get_state_connection(),
    },
}

config = Config(
    gateways=gateways,
    default_gateway="my_gateway",
)

Commands

Authentication

# Login
dagctl auth login --org <organization>
dagctl auth login --org acme --api-url https://api.staging.dagctl.io

# Logout
dagctl auth logout

# Check status
dagctl auth status

Project Management

# Set project
dagctl use-project <project-name>
dagctl use-project my-project -k  # Skip SSL verification (dev)

# Generate config
dagctl config generate

# View current context
dagctl config current

API Reference

get_state_connection()

Returns a dictionary with PostgreSQL connection configuration for SQLMesh.

from dagctl import get_state_connection

# Returns:
# {
#     "type": "postgres",
#     "host": "pg-proxy.dev-us-1.dagctl.internal",
#     "port": 5432,
#     "database": "org_acme_myproject",
#     "user": "acme/myproject",
#     "password": "<jwt-token>"
# }

Features:

• ✅ Auto-refreshes JWT tokens when expired
• ✅ Uses current org/project from ~/.dagctl/config.yaml
• ✅ Thread-safe and cached
• ✅ Works with SQLMesh's config system

Parameters:

• gateway (optional): Gateway name (uses current project if not specified)
• insecure (optional): Skip SSL verification for development

get_fresh_token()

Get a fresh JWT access token, refreshing if necessary.

from dagctl import get_fresh_token

token = get_fresh_token()
# Returns: "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9..."

How Authentication Works

1. Login Flow:

   • dagctl auth login opens browser for OAuth
   • You authenticate with Auth0
   • dagctl receives JWT tokens (access_token, refresh_token, id_token)
   • Tokens stored in ~/.dagctl/auth.json (0600 permissions)

2. State Connection:

   • SQLMesh calls get_state_connection()
   • dagctl checks if JWT is expired
   • If expired, automatically refreshes using refresh_token
   • Returns connection with JWT as password

3. pg-proxy Authentication:

   • Client connects to pg-proxy with JWT as password
   • pg-proxy validates JWT using Auth0 JWKS
   • If valid, pg-proxy fetches org credentials from management API
   • pg-proxy connects to CoreDB and proxies traffic

Configuration Directory

dagctl stores configuration in ~/.dagctl/:

~/.dagctl/
├── config.yaml      # Current org/project context
├── auth.json        # Auth tokens (0600 permissions)
└── credentials/     # Cached credentials (deprecated, not used)

Security

• JWT-Based Auth: No passwords stored, only JWT tokens
• Auto-Refresh: Tokens automatically refresh when expired
• Short-Lived: Access tokens expire after 1 hour
• Secure Storage: All files in ~/.dagctl/ have 0600 permissions
• No Shared Secrets: Each user authenticates individually

Development

For development with self-signed certificates:

# Login with SSL verification disabled
dagctl auth login --org acme -k

# Set project with SSL verification disabled  
dagctl use-project my-project -k

The -k flag is equivalent to curl -k and skips SSL certificate verification.

Troubleshooting

"Not authenticated"

dagctl auth login --org your-org

"No project set"

dagctl use-project your-project

"Authentication failed: invalid token"

Your token may have expired. Re-authenticate:

dagctl auth logout
dagctl auth login --org your-org

"State backend connection failed"

Check that:

1. You're authenticated: dagctl auth status
2. Project is set: dagctl config current
3. pg-proxy is running and accessible

License

Apache-2.0

Support

• Documentation: https://docs.dagctl.io
• Email: support@scalecraft.dev