cloe-delta-table-manager 0.2.1

A new cool package.

cloe-delta-table-manager

Owner: initions

Usage

Run dtm from the directory containing project.yaml to execute the migration pipeline. Use plan to generate a deployment plan without executing SQL, or deploy to run the full pipeline including SQL execution.

You can also initialize a dtm project by using the init command, which will create a sample project structure with example SQL files and a project.yaml configuration.

uv run dtm --target dev plan
uv run dtm --target dev deploy

Migration file metadata

User migration SQL files can declare metadata in a leading comment. The first comment of the file is parsed as YAML and is currently used to express execution-order dependencies between migrations.

Supported keys:

• depends_on — one or more migration files that must run before this one. Accepts a single path or a list. Paths are relative to the migration's own directory.

/* depends_on:
  - 001_create_table.sql
  - 002_alter_table.sql
*/
CREATE INDEX idx_email ON users(email);

A single dependency can also be given as a scalar:

/* depends_on: 001_create_table.sql */
ALTER TABLE users ADD COLUMN email VARCHAR(255);

Files without a leading comment are treated as having no metadata and no dependencies.

Known limitations

• Migration metadata must be the first comment. The metadata block (see Migration file metadata) is read from the first comment in the file, which must be valid YAML. Any other content as the first comment causes a parsing error and the migration fails. Generic, non-metadata leading comments are not yet supported.

Connection environment variables

The Databricks connector reads credentials from environment variables. Set the workspace URL plus one authentication method.

export CLOE_DBX_WORKSPACE_URL="https://adb-xxxx.azuredatabricks.net"
export CLOE_DBX_SQL_WAREHOUSE_ID=""   # SQL warehouse / serverless compute used for execution

Authentication — choose one:

# Personal Access Token
export CLOE_DBX_PAT=""
# PAT can be derived from Databricks CLI
$(databricks auth token --host "$CLOE_DBX_WORKSPACE_URL" 2>/dev/null \
  | python3 -c "import json,sys; print(json.load(sys.stdin)['access_token'])" 2>/dev/null)

# Databricks-managed service principal
export CLOE_DBX_CLIENT_ID=""
export CLOE_DBX_CLIENT_SECRET=""

# Azure Entra ID service principal
export CLOE_AZURE_TENANT_ID=""
export CLOE_AZURE_CLIENT_ID=""
export CLOE_AZURE_CLIENT_SECRET=""

Optional defaults for SQL execution:

export CLOE_DBX_DEFAULT_CATALOG=""
export CLOE_DBX_DEFAULT_SCHEMA=""
export CLOE_DBX_STATEMENT_TIMEOUT="50s"   # Databricks synchronous max is 50s

Devcontainer

If the claude code devcontainer-feature is used, it expect you to have a local .credentials.json:

${localEnv:HOME}/.claude/.credentials.json

If you are setting this up for the first time, you need to create this file manually but it can be empty.

mkdir -p ~/.claude/
touch ~/.claude/.credentials.json

Testing

Snapshot Testing

This project uses inline-snapshot for test assertions. Snapshots capture the complete structure of test outputs, making tests more comprehensive and easier to maintain.

Running Tests

# Run all tests
uv run pytest

# Run specific test file
uv run pytest tests/state/test_state.py

Updating Snapshots

When you modify code that changes test outputs, you'll need to update the snapshots:

# Review and approve snapshot changes interactively
uv run pytest --inline-snapshot=review

# Automatically fix all snapshots (use with caution)
uv run pytest --inline-snapshot=fix

# Create new snapshots for tests with empty snapshot() calls
uv run pytest --inline-snapshot=create

Writing Tests with Snapshots

Use .model_dump() to convert Pydantic models to dictionaries before snapshotting:

from inline_snapshot import snapshot
from dirty_equals import IsStr, IsDatetime, IsUUID

def test_example(self):
    result = create_database_state()

    assert result.model_dump() == snapshot({
        "objects": {
            "db1": {
                "id": IsUUID(4),  # Dynamic UUID field
                "name": "db1",
                "created_at": IsDatetime(),  # Dynamic datetime field
            }
        },
        "metadata": {
            "version": 1,
            "delta_table_manager_version": IsStr(),
        }
    })

Use dirty-equals matchers (IsStr(), IsDatetime(), IsUUID(), etc.) for fields that change between test runs.