windmill

What this skill does

# Windmill Workflow Automation Skill

Master Windmill for developer-first workflow automation that transforms scripts into production workflows with auto-generated UIs. This skill covers script authoring in Python/TypeScript/Go/Bash, flow orchestration, approval flows, schedules, and enterprise deployment patterns.

## When to Use This Skill

### USE when:
- Developers prefer writing code over visual tools
- Need auto-generated UIs for script parameters
- Building internal tools with minimal frontend work
- Python, TypeScript, Go, or Bash are primary languages
- Combining workflow automation with internal tools
- Need code review and version control for automations
- Require approval flows with audit trails
- Self-hosting for data sovereignty

### DON'T USE when:
- Non-developers need to build workflows (use n8n, Activepieces)
- Need 400+ pre-built integrations (use n8n)
- Complex DAG orchestration with dependencies (use Airflow)
- CI/CD pipelines tightly coupled with git (use GitHub Actions)
- Simple visual automation preferred (use Activepieces)

## Prerequisites

### Installation Options

**Option 1: Docker Compose (Recommended)**
```yaml
# docker-compose.yml
version: '3.8'

services:
  windmill:
    image: ghcr.io/windmill-labs/windmill:main
    restart: always
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgres://windmill:${POSTGRES_PASSWORD}@postgres:5432/windmill?sslmode=disable
      - MODE=standalone
      - BASE_URL=http://localhost:8000
      - RUST_LOG=info
      - NUM_WORKERS=4
      - DISABLE_SERVER=false
      - DISABLE_WORKERS=false
    depends_on:
      postgres:
        condition: service_healthy
    volumes:
      - worker_dependency_cache:/tmp/windmill/cache

  postgres:
    image: postgres:15
    restart: always
    environment:
      - POSTGRES_USER=windmill
      - POSTGRES_PASSWORD=${POSTGRES_PASSWORD}
      - POSTGRES_DB=windmill
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U windmill"]
      interval: 10s
      timeout: 5s
      retries: 5

  lsp:
    image: ghcr.io/windmill-labs/windmill-lsp:latest
    restart: always
    ports:
      - "3001:3001"
    volumes:
      - lsp_cache:/root/.cache

volumes:
  postgres_data:
  worker_dependency_cache:
  lsp_cache:
```

```bash
# Generate password
export POSTGRES_PASSWORD=$(openssl rand -hex 32)

# Start services
docker compose up -d

# Access UI at http://localhost:8000
# Default credentials: [email protected] / changeme
```

**Option 2: Kubernetes with Helm**
```bash
# Add Windmill Helm repo
helm repo add windmill https://windmill-labs.github.io/windmill-helm-charts
helm repo update

# Create namespace
kubectl create namespace windmill

# Create secrets
kubectl create secret generic windmill-secrets \
  --namespace windmill \
  --from-literal=postgres-password=$(openssl rand -hex 32)

# Install Windmill
helm install windmill windmill/windmill \
  --namespace windmill \
  --set postgresql.auth.existingSecret=windmill-secrets \
  --set windmill.baseUrl=https://windmill.example.com \
  --set windmill.workers.replicas=3

# Get LoadBalancer IP
kubectl get svc -n windmill windmill-app
```

**Option 3: Local Development**
```bash
# Install Windmill CLI
npm install -g windmill-cli

# Or with pip
pip install wmill

# Login to instance
wmill workspace add my-workspace https://windmill.example.com
wmill workspace switch my-workspace

# Initialize project
wmill init

# Sync local scripts to Windmill
wmill sync push
```

### Development Setup
```bash
# Install language-specific dependencies

# Python development
pip install wmill pandas numpy requests

# TypeScript/Deno development
# Windmill uses Deno runtime for TypeScript
deno --version

# Go development
go install github.com/windmill-labs/windmill-go-client@latest

# Bash scripts work out of the box
```

## Core Capabilities

### 1. Python Scripts

```python
# scripts/data_processing/fetch_and_transform.py
"""
Fetch data from API and transform for analysis.
Auto-generates UI with input fields for all parameters.
"""

import wmill
from datetime import datetime, timedelta
import requests
import pandas as pd


def main(
    api_endpoint: str,
    date_range_days: int = 7,
    include_metadata: bool = True,
    output_format: str = "json",  # Dropdown: json, csv, parquet
    filters: dict = None,
):
    """
    Fetch and transform data from external API.

    Args:
        api_endpoint: The API endpoint URL to fetch data from
        date_range_days: Number of days of data to fetch (default: 7)
        include_metadata: Whether to include metadata in response
        output_format: Output format - json, csv, or parquet
        filters: Optional filters to apply to the data

    Returns:
        Transformed data in specified format
    """
    # Get API credentials from Windmill resources
    api_credentials = wmill.get_resource("u/admin/api_credentials")

    # Calculate date range
    end_date = datetime.now()
    start_date = end_date - timedelta(days=date_range_days)

    # Fetch data
    headers = {
        "Authorization": f"Bearer {api_credentials['api_key']}",
        "Content-Type": "application/json"
    }

    params = {
        "start_date": start_date.isoformat(),
        "end_date": end_date.isoformat(),
    }

    if filters:
        params.update(filters)

    response = requests.get(
        f"{api_endpoint}/data",
        headers=headers,
        params=params,
        timeout=30
    )
    response.raise_for_status()
    data = response.json()

    # Transform with pandas
    df = pd.DataFrame(data["records"])

    # Apply transformations
    if "timestamp" in df.columns:
        df["timestamp"] = pd.to_datetime(df["timestamp"])
        df["date"] = df["timestamp"].dt.date
        df["hour"] = df["timestamp"].dt.hour

    if "value" in df.columns:
        df["value_normalized"] = (df["value"] - df["value"].min()) / (
            df["value"].max() - df["value"].min()
        )

    # Generate summary statistics
    summary = {
        "total_records": len(df),
        "date_range": {
            "start": str(start_date.date()),
            "end": str(end_date.date())
        },
        "statistics": df.describe().to_dict() if not df.empty else {}
    }

    # Format output
    if output_format == "json":
        result = df.to_dict(orient="records")
    elif output_format == "csv":
        result = df.to_csv(index=False)
    else:
        # For parquet, return as dict (Windmill handles serialization)
        result = df.to_dict(orient="records")

    if include_metadata:
        return {
            "data": result,
            "metadata": summary,
            "format": output_format,
            "generated_at": datetime.now().isoformat()
        }

    return result
```

```python
# scripts/integrations/sync_crm_to_database.py
"""
Sync CRM contacts to internal database with deduplication.
"""

import wmill
from typing import Optional
import psycopg2
from psycopg2.extras import execute_values


def main(
    crm_list_id: str,
    batch_size: int = 100,
    dry_run: bool = False,
    update_existing: bool = True,
):
    """
    Sync CRM contacts to PostgreSQL database.

    Args:
        crm_list_id: The CRM list ID to sync
        batch_size: Number of records per batch
        dry_run: If True, don't actually write to database
        update_existing: If True, update existing records

    Returns:
        Sync statistics
    """
    # Get resources
    crm_api = wmill.get_resource("u/admin/crm_api")
    db_conn = wmill.get_resource("u/admin/postgres_warehouse")

    # Fetch contacts from CRM
    import requests
    contacts = []
    page = 1

    while True:
        response = requests.get(
            f"{crm_api['base_url']}/lists/{crm_list_id}/contacts",
            headers={"Authorization": f"Bearer {crm_api['api_key']}"},
            params={"page": page, "per_page": batch_size}
        )
        response.raise_for_status()