Imported from diegosouzapw/awesome-omni-skill (skills/tools/windmill-majiayu000/SKILL.md). Install upstream with npx skills add diegosouzapw/awesome-omni-skill --skill windmill-majiayu000. Copyright stays with the author.

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)

# 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:

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

# Start services
docker compose up -d

# Access UI at http://localhost:8000
# Default credentials: admin@windmill.dev / changeme

Option 2: Kubernetes with Helm

# 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

# 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

# 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

# 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

# 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()
        data = response.json()

        contacts.extend(data["contacts"])

        if not data.get("has_more"):
            break
        page += 1

    print(f"Fetched {len(contacts)} contacts from CRM")

    if dry_run:
        return {
            "mode": "dry_run",
            "contacts_fetched": len(contacts),
            "sample": contacts[:5]
        }

    # Connect to database
    conn = psycopg2.connect(
        host=db_conn["host"],
        port=db_conn["port"],
        database=db_conn["database"],
        user=db_conn["user"],
        password=db_conn["password"]
    )

    stats = {"inserted": 0, "updated": 0, "skipped": 0, "errors": []}

    try:
        with conn.cursor() as cur:
            for contact in contacts:
                try:
                    # Check if exists
                    cur.execute(
                        "SELECT id FROM contacts WHERE email = %s",
                        (contact["email"],)
                    )
                    existing = cur.fetchone()

                    if existing:
                        if update_existing:
                            cur.execute("""
                                UPDATE contacts
                                SET name = %s, company = %s, phone = %s,
                                    updated_at = NOW(), crm_id = %s
                                WHERE email = %s
                            """, (
                                contact["name"],
                                contact.get("company"),
                                contact.get("phone"),
                                contact["id"],
                                contact["email"]
                            ))
                            stats["updated"] += 1
                        else:
                            stats["skipped"] += 1
                    else:
                        cur.execute("""
                            INSERT INTO contacts
                            (email, name, company, phone, crm_id, created_at)
                            VALUES (%s, %s, %s, %s, %s, NOW())
                        """, (
                            contact["email"],
                            contact["name"],
                            contact.get("company"),
                            contact.get("phone"),
                            contact["id"]
                        ))
                        stats["inserted"] += 1

                except Exception as e:
                    stats["errors"].append({
                        "contact": contact["email"],
                        "error": str(e)
                    })

            conn.commit()

    finally:
        conn.close()

    return {
        "mode": "live",
        "contacts_fetched": len(contacts),
        **stats
    }

2. TypeScript/Deno Scripts

// scripts/api/webhook_handler.ts
/**
 * Handle incoming webhooks with validation and routing.
 * Uses Deno runtime with TypeScript support.
 */

import * as wmill from "npm:windmill-client@1";

// Define input types for auto-generated UI
type WebhookPayload = {
  event_type: string;
  data: Record<string, unknown>;
  timestamp: string;
  signature?: string;
};

type HandlerConfig = {
  validate_signature: boolean;
  allowed_events: string[];
  forward_to_slack: boolean;
};

export async function main(
  payload: WebhookPayload,
  config: HandlerConfig = {
    validate_signature: true,
    allowed_events: ["order.created", "order.updated", "payment.completed"],
    forward_to_slack: true,
  }
): Promise<{
  processed: boolean;
  event_type: string;
  actions_taken: string[];
}> {
  const actions: string[] = [];

  // Get webhook secret from resources
  const webhookSecret = await wmill.getResource("u/admin/webhook_secret");

  // Validate signature if required
  if (config.validate_signature && payload.signature) {
    const crypto = await import("node:crypto");
    const expectedSignature = crypto
      .createHmac("sha256", webhookSecret.secret)
      .update(JSON.stringify(payload.data))
      .digest("hex");

    if (payload.signature !== expectedSignature) {
      throw new Error("Invalid webhook signature");
    }
    actions.push("signature_validated");
  }

  // Check if event is allowed
  if (!config.allowed_events.includes(payload.event_type)) {
    return {
      processed: false,
      event_type: payload.event_type,
      actions_taken: ["event_filtered"],
    };
  }

  // Route based on event type
  switch (payload.event_type) {
    case "order.created":
      await handleOrderCreated(payload.data);
      actions.push("order_processed");
      break;

    case "order.updated":
      await handleOrderUpdated(payload.data);
      actions.push("order_updated");
      break;

    case "payment.completed":
      await handlePaymentCompleted(payload.data);
      actions.push("payment_recorded");
      break;
  }

  // Forward to Slack if configured
  if (config.forward_to_slack) {
    const slackWebhook = await wmill.getResource("u/admin/slack_webhook");
    await fetch(slackWebhook.url, {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({
        text: `Webhook received: ${payload.event_type}`,
        blocks: [
          {
            type: "section",
            text: {
              type: "mrkdwn",
              text: `*Event:* ${payload.event_type}\n*Timestamp:* ${payload.timestamp}`,
            },
          },
        ],
      }),
    });
    actions.push("slack_notified");
  }

  return {
    processed: true,
    event_type: payload.event_type,
    actions_taken: actions,
  };
}

async function handleOrderCreated(data: Record<string, unknown>) {
  console.log("Processing new order:", data);
  // Implementation
}

async function handleOrderUpdated(data: Record<string, unknown>) {
  console.log("Processing order update:", data);
  // Implementation
}

async function handlePaymentCompleted(data: Record<string, unknown>) {
  console.log("Processing payment:", data);
  // Implementation
}

// scripts/data/aggregate_metrics.ts
/**
 * Aggregate metrics from multiple sources into unified dashboard data.
 */

import * as wmill from "npm:windmill-client@1";

type MetricsSource = "database" | "api" | "cache";
type AggregationPeriod = "hourly" | "daily" | "weekly" | "monthly";

interface MetricConfig {
  sources: MetricsSource[];
  period: AggregationPeriod;
  include_comparisons: boolean;
  custom_dimensions?: string[];
}

interface AggregatedMetrics {
  period: string;
  total_revenue: number;
  total_orders: number;
  avg_order_value: number;
  unique_customers: number;
  top_products: Array<{ name: string; revenue: number; quantity: number }>;
  by_dimension: Record<string, Record<string, number>>;
  comparisons?: {
    previous_period: Record<string, number>;
    change_percent: Record<string, number>;
  };
}

export async function main(
  start_date: string,
  end_date: string,
  config: MetricConfig = {
    sources: ["database"],
    period: "daily",
    include_comparisons: true,
  }
): Promise<AggregatedMetrics> {
  // Get database connection
  const dbConfig = await wmill.getResource("u/admin/analytics_db");

  // Dynamic import for database client
  const { Client } = await import("npm:pg@8");
  const client = new Client(dbConfig);
  await client.connect();

  try {
    // Fetch base metrics
    const metricsQuery = `
      SELECT
        DATE_TRUNC('${config.period}', created_at) as period,
        COUNT(*) as total_orders,
        SUM(total_amount) as total_revenue,
        COUNT(DISTINCT customer_id) as unique_customers
      FROM orders
      WHERE created_at >= $1 AND created_at < $2
      GROUP BY DATE_TRUNC('${config.period}', created_at)
      ORDER BY period
    `;

    const metricsResult = await client.query(metricsQuery, [
      start_date,
      end_date,
    ]);

    // Aggregate across periods
    const totalRevenue = metricsResult.rows.reduce(
      (sum, r) => sum + parseFloat(r.total_revenue || 0),
      0
    );
    const totalOrders = metricsResult.rows.reduce(
      (sum, r) => sum + parseInt(r.total_orders || 0),
      0
    );
    const uniqueCustomers = metricsResult.rows.reduce(
      (sum, r) => sum + parseInt(r.unique_customers || 0),
      0
    );

    // Fetch top products
    const topProductsQuery = `
      SELECT
        p.name,
        SUM(oi.quantity) as quantity,
        SUM(oi.quantity * oi.unit_price) as revenue
      FROM order_items oi
      JOIN products p ON oi.product_id = p.id
      JOIN orders o ON oi.order_id = o.id
      WHERE o.created_at >= $1 AND o.created_at < $2
      GROUP BY p.id, p.name
      ORDER BY revenue DESC
      LIMIT 10
    `;

    const topProductsResult = await client.query(topProductsQuery, [
      start_date,
      end_date,
    ]);

    const result: AggregatedMetrics = {
      period: `${start_date} to ${end_date}`,
      total_revenue: totalRevenue,
      total_orders: totalOrders,
      avg_order_value: totalOrders > 0 ? totalRevenue / totalOrders : 0,
      unique_customers: uniqueCustomers,
      top_products: topProductsResult.rows.map((r) => ({
        name: r.name,
        revenue: parseFloat(r.revenue),
        quantity: parseInt(r.quantity),
      })),
      by_dimension: {},
    };

    // Add dimension breakdowns
    if (config.custom_dimensions) {
      for (const dimension of config.custom_dimensions) {
        const dimensionQuery = `
          SELECT
            ${dimension},
            SUM(total_amount) as revenue
          FROM orders
          WHERE created_at >= $1 AND created_at < $2
          GROUP BY ${dimension}
        `;
        const dimResult = await client.query(dimensionQuery, [
          start_date,
          end_date,
        ]);
        result.by_dimension[dimension] = Object.fromEntries(
          dimResult.rows.map((r) => [r[dimension], parseFloat(r.revenue)])
        );
      }
    }

    // Add period comparisons
    if (config.include_comparisons) {
      const periodDays = Math.ceil(
        (new Date(end_date).getTime() - new Date(start_date).getTime()) /
          (1000 * 60 * 60 * 24)
      );
      const prevStart = new Date(
        new Date(start_date).getTime() - periodDays * 24 * 60 * 60 * 1000
      )
        .toISOString()
        .split("T")[0];
      const prevEnd = start_date;

      const prevQuery = `
        SELECT
          COUNT(*) as total_orders,
          COALESCE(SUM(total_amount), 0) as total_revenue,
          COUNT(DISTINCT customer_id) as unique_customers
        FROM orders
        WHERE created_at >= $1 AND created_at < $2
      `;

      const prevResult = await client.query(prevQuery, [prevStart, prevEnd]);
      const prev = prevResult.rows[0];

      const prevRevenue = parseFloat(prev.total_revenue || 0);
      const prevOrders = parseInt(prev.total_orders || 0);

      result.comparisons = {
        previous_period: {
          revenue: prevRevenue,
          orders: prevOrders,
        },
        change_percent: {
          revenue: prevRevenue > 0 ? ((totalRevenue - prevRevenue) / prevRevenue) * 100 : 0,
          orders: prevOrders > 0 ? ((totalOrders - prevOrders) / prevOrders) * 100 : 0,
        },
      };
    }

    return result;
  } finally {
    await client.end();
  }
}

3. Go Scripts

// scripts/performance/batch_processor.go
// High-performance batch processing using Go.

package inner

import (
	"context"
	"encoding/json"
	"fmt"
	"sync"
	"time"

	wmill "github.com/windmill-labs/windmill-go-client"
)

type BatchConfig struct {
	BatchSize     int    `json:"batch_size"`
	Concurrency   int    `json:"concurrency"`
	RetryAttempts int    `json:"retry_attempts"`
	TimeoutSecs   int    `json:"timeout_secs"`
}

type ProcessResult struct {
	TotalItems     int            `json:"total_items"`
	Successful     int            `json:"successful"`
	Failed         int            `json:"failed"`
	ProcessingTime float64        `json:"processing_time_seconds"`
	Errors         []ProcessError `json:"errors,omitempty"`
}

type ProcessError struct {
	ItemID string `json:"item_id"`
	Error  string `json:"error"`
}

func Main(
	items []map[string]interface{},
	config BatchConfig,
) (*ProcessResult, error) {
	startTime := time.Now()

	// Set defaults
	if config.BatchSize == 0 {
		config.BatchSize = 100
	}
	if config.Concurrency == 0 {
		config.Concurrency = 4
	}
	if config.RetryAttempts == 0 {
		config.RetryAttempts = 3
	}
	if config.TimeoutSecs == 0 {
		config.TimeoutSecs = 30
	}

	// Get API credentials
	ctx := context.Background()
	resource, err := wmill.GetResource("u/admin/processing_api")
	if err != nil {
		return nil, fmt.Errorf("failed to get resource: %w", err)
	}

	apiKey := resource["api_key"].(string)

	result := &ProcessResult{
		TotalItems: len(items),
	}

	var mu sync.Mutex
	var wg sync.WaitGroup

	// Create worker pool
	itemChan := make(chan map[string]interface{}, config.BatchSize)
	errorChan := make(chan ProcessError, len(items))

	// Start workers
	for i := 0; i < config.Concurrency; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for item := range itemChan {
				err := processItem(ctx, item, apiKey, config)
				mu.Lock()
				if err != nil {
					result.Failed++
					errorChan <- ProcessError{
						ItemID: fmt.Sprintf("%v", item["id"]),
						Error:  err.Error(),
					}
				} else {
					result.Successful++
				}
				mu.Unlock()
			}
		}()
	}

	// Send items to workers
	for _, item := range items {
		itemChan <- item
	}
	close(itemChan)

	// Wait for all workers to complete
	wg.Wait()
	close(errorChan)

	// Collect errors
	for err := range errorChan {
		result.Errors = append(result.Errors, err)
	}

	result.ProcessingTime = time.Since(startTime).Seconds()

	return result, nil
}

func processItem(
	ctx context.Context,
	item map[string]interface{},
	apiKey string,
	config BatchConfig,
) error {
	var lastErr error

	for attempt := 1; attempt <= config.RetryAttempts; attempt++ {
		ctx, cancel := context.WithTimeout(ctx, time.Duration(config.TimeoutSecs)*time.Second)
		defer cancel()

		// Process item (simplified - real implementation would make API calls)
		select {
		case <-ctx.Done():
			lastErr = ctx.Err()
		default:
			// Simulate processing
			time.Sleep(10 * time.Millisecond)

			// Validate item
			if _, ok := item["id"]; !ok {
				return fmt.Errorf("missing required field: id")
			}

			return nil
		}

		// Exponential backoff before retry
		if attempt < config.RetryAttempts {
			time.Sleep(time.Duration(1<<attempt) * time.Second)
		}
	}

	return fmt.Errorf("failed after %d attempts: %w", config.RetryAttempts, lastErr)
}

4. Bash Scripts

#!/bin/bash
# scripts/devops/deploy_service.sh
# Deploy a service with health checks and rollback capability.

# Windmill automatically provides these as environment variables
# SERVICE_NAME, VERSION, ENVIRONMENT, DRY_RUN

set -euo pipefail

# Get secrets from Windmill resources
DEPLOY_KEY=$(curl -s -H "Authorization: Bearer $WM_TOKEN" \
  "$BASE_INTERNAL_URL/api/w/$WM_WORKSPACE/resources/get/u/admin/deploy_key" | jq -r '.value.key')

AWS_REGION=$(curl -s -H "Authorization: Bearer $WM_TOKEN" \
  "$BASE_INTERNAL_URL/api/w/$WM_WORKSPACE/resources/get/u/admin/aws_config" | jq -r '.value.region')

log() {
  echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1"
}

deploy_service() {
  local service=$1
  local version=$2
  local env=$3

  log "Deploying $service version $version to $env"

  # Update ECS service
  if [[ "$DRY_RUN" == "true" ]]; then
    log "[DRY RUN] Would update ECS service $service to $version"
    return 0
  fi

  aws ecs update-service \
    --cluster "${env}-cluster" \
    --service "$service" \
    --force-new-deployment \
    --region "$AWS_REGION"

  log "Deployment initiated"
}

wait_for_healthy() {
  local service=$1
  local env=$2
  local max_wait=300
  local interval=10
  local elapsed=0

  log "Waiting for $service to become healthy..."

  while [[ $elapsed -lt $max_wait ]]; do
    local running_count=$(aws ecs describe-services \
      --cluster "${env}-cluster" \
      --services "$service" \
      --region "$AWS_REGION" \
      --query 'services[0].runningCount' \
      --output text)

    local desired_count=$(aws ecs describe-services \
      --cluster "${env}-cluster" \
      --services "$service" \
      --region "$AWS_REGION" \
      --query 'services[0].desiredCount' \
      --output text)

    if [[ "$running_count" == "$desired_count" ]]; then
      log "Service healthy: $running_count/$desired_count tasks running"
      return 0
    fi

    log "Waiting... ($running_count/$desired_count tasks running)"
    sleep $interval
    elapsed=$((elapsed + interval))
  done

  log "ERROR: Service did not become healthy within ${max_wait}s"
  return 1
}

rollback() {
  local service=$1
  local env=$2

  log "Rolling back $service in $env"

  # Get previous task definition
  local prev_task_def=$(aws ecs describe-services \
    --cluster "${env}-cluster" \
    --services "$service" \
    --region "$AWS_REGION" \
    --query 'services[0].deployments[1].taskDefinition' \
    --output text)

  if [[ -z "$prev_task_def" || "$prev_task_def" == "None" ]]; then
    log "ERROR: No previous task definition found for rollback"
    return 1
  fi

  aws ecs update-service \
    --cluster "${env}-cluster" \
    --service "$service" \
    --task-definition "$prev_task_def" \
    --region "$AWS_REGION"

  log "Rollback initiated to $prev_task_def"
}

# Main execution
main() {
  log "=== Deployment Started ==="
  log "Service: $SERVICE_NAME"
  log "Version: $VERSION"
  log "Environment: $ENVIRONMENT"
  log "Dry Run: $DRY_RUN"

  # Deploy
  if ! deploy_service "$SERVICE_NAME" "$VERSION" "$ENVIRONMENT"; then
    log "ERROR: Deployment failed"
    exit 1
  fi

  # Wait for health (skip in dry run)
  if [[ "$DRY_RUN" != "true" ]]; then
    if ! wait_for_healthy "$SERVICE_NAME" "$ENVIRONMENT"; then
      log "Deployment failed health check, initiating rollback"
      rollback "$SERVICE_NAME" "$ENVIRONMENT"
      exit 1
    fi
  fi

  log "=== Deployment Completed Successfully ==="

  # Output result as JSON for Windmill
  cat <<EOF
{
  "status": "success",
  "service": "$SERVICE_NAME",
  "version": "$VERSION",
  "environment": "$ENVIRONMENT",
  "timestamp": "$(date -u +%Y-%m-%dT%H:%M:%SZ)"
}
EOF
}

main

5. Flow Orchestration

# flows/order_processing_flow.yaml
summary: Order Processing Pipeline
description: |
  End-to-end order processing with validation, payment, and fulfillment.
  Includes approval for high-value orders.

value:
  modules:
    - id: validate_order
      value:
        type: script
        path: f/order_processing/validate_order
        input_transforms:
          order:
            type: javascript
            expr: flow_input.order

    - id: check_inventory
      value:
        type: script
        path: f/inventory/check_availability
        input_transforms:
          items:
            type: javascript
            expr: results.validate_order.items

    - id: route_by_value
      value:
        type: branchone
        branches:
          - summary: High Value Order
            expr: results.validate_order.total > 5000
            modules:
              - id: request_approval
                value:
                  type: approval
                  timeout: 86400  # 24 hours
                  approvers:
                    - admin@example.com
                    - manager@example.com

              - id: process_approved
                value:
                  type: script
                  path: f/payments/process_payment
                  input_transforms:
                    order_id:
                      type: javascript
                      expr: results.validate_order.order_id
                    amount:
                      type: javascript
                      expr: results.validate_order.total

          - summary: Normal Order
            expr: results.validate_order.total <= 5000
            modules:
              - id: process_normal
                value:
                  type: script
                  path: f/payments/process_payment
                  input_transforms:
                    order_id:
                      type: javascript
                      expr: results.validate_order.order_id
                    amount:
                      type: javascript
                      expr: results.validate_order.total

    - id: create_shipment
      value:
        type: script
        path: f/fulfillment/create_shipment
        input_transforms:
          order_id:
            type: javascript
            expr: results.validate_order.order_id
          shipping_address:
            type: javascript
            expr: results.validate_order.shipping_address

    - id: send_confirmation
      value:
        type: script
        path: f/notifications/send_order_confirmation
        input_transforms:
          email:
            type: javascript
            expr: results.validate_order.customer_email
          order_details:
            type: javascript
            expr: |
              {
                order_id: results.validate_order.order_id,
                total: results.validate_order.total,
                tracking_number: results.create_shipment.tracking_number
              }

schema:
  $schema: https://json-schema.org/draft/2020-12/schema
  type: object
  properties:
    order:
      type: object
      properties:
        customer_email:
          type: string
          format: email
        items:
          type: array
          items:
            type: object
        shipping_address:
          type: object
      required:
        - customer_email
        - items
        - shipping_address
  required:
    - order

6. Schedule Management

# scripts/scheduling/dynamic_scheduler.py
"""
Dynamically manage schedules based on business rules.
"""

import wmill
from datetime import datetime, timedelta
from typing import List, Optional


def main(
    schedule_configs: List[dict],
    dry_run: bool = True,
):
    """
    Update Windmill schedules based on configuration.

    Args:
        schedule_configs: List of schedule configurations
        dry_run: If True, only report what would change

    Returns:
        Summary of schedule changes
    """
    client = wmill.Client()
    workspace = wmill.get_workspace()

    changes = []

    for config in schedule_configs:
        schedule_path = config["path"]
        enabled = config.get("enabled", True)
        cron = config.get("cron")
        timezone = config.get("timezone", "UTC")

        # Check business hours constraint
        if config.get("business_hours_only", False):
            # Modify cron to only run during business hours (9-17)
            if cron:
                cron_parts = cron.split()
                if len(cron_parts) >= 5:
                    cron_parts[1] = "9-17"  # Hours
                    cron_parts[4] = "1-5"   # Weekdays only
                    cron = " ".join(cron_parts)

        # Check maintenance window constraint
        if config.get("skip_maintenance_windows", False):
            maintenance = get_maintenance_windows()
            now = datetime.now()
            in_maintenance = any(
                m["start"] <= now <= m["end"]
                for m in maintenance
            )
            if in_maintenance:
                enabled = False

        change = {
            "path": schedule_path,
            "cron": cron,
            "timezone": timezone,
            "enabled": enabled,
            "dry_run": dry_run
        }

        if not dry_run:
            # Update schedule via Windmill API
            try:
                client.update_schedule(
                    workspace=workspace,
                    path=schedule_path,
                    schedule={
                        "schedule": cron,
                        "timezone": timezone,
                        "enabled": enabled
                    }
                )
                change["status"] = "updated"
            except Exception as e:
                change["status"] = "error"
                change["error"] = str(e)
        else:
            change["status"] = "would_update"

        changes.append(change)

    return {
        "total_schedules": len(schedule_configs),
        "changes": changes,
        "dry_run": dry_run
    }


def get_maintenance_windows():
    """Fetch maintenance windows from configuration."""
    try:
        config = wmill.get_variable("u/admin/maintenance_windows")
        return config.get("windows", [])
    except:
        return []

7. Approval Flows

# scripts/approvals/expense_approval.py
"""
Expense approval workflow with multi-level approvals.
"""

import wmill
from typing import Optional
from enum import Enum


class ApprovalLevel(Enum):
    AUTO = "auto"
    MANAGER = "manager"
    DIRECTOR = "director"
    EXECUTIVE = "executive"


def main(
    expense_id: str,
    amount: float,
    category: str,
    description: str,
    requestor_email: str,
    receipts: Optional[list] = None,
):
    """
    Process expense approval request.

    Args:
        expense_id: Unique expense ID
        amount: Expense amount in USD
        category: Expense category
        description: Expense description
        requestor_email: Email of person requesting expense
        receipts: List of receipt file URLs

    Returns:
        Approval request status and next steps
    """
    # Determine approval level based on amount and category
    approval_level = determine_approval_level(amount, category)

    # Get approvers for this level
    approvers = get_approvers(approval_level, requestor_email)

    if approval_level == ApprovalLevel.AUTO:
        # Auto-approve small expenses
        return {
            "expense_id": expense_id,
            "status": "approved",
            "approval_level": approval_level.value,
            "auto_approved": True,
            "message": f"Expense auto-approved (amount: ${amount:.2f})"
        }

    # Create approval request in database
    db = wmill.get_resource("u/admin/expenses_db")
    approval_id = create_approval_request(
        db,
        expense_id=expense_id,
        amount=amount,
        category=category,
        description=description,
        requestor=requestor_email,
        approvers=approvers,
        level=approval_level.value
    )

    # Send notification to approvers
    slack = wmill.get_resource("u/admin/slack_webhook")
    send_approval_notification(
        slack,
        expense_id=expense_id,
        amount=amount,
        category=category,
        description=description,
        requestor=requestor_email,
        approvers=approvers,
        approval_link=f"https://windmill.example.com/approvals/{approval_id}"
    )

    # Return approval URL for Windmill's built-in approval step
    return {
        "expense_id": expense_id,
        "approval_id": approval_id,
        "status": "pending_approval",
        "approval_level": approval_level.value,
        "approvers": approvers,
        "resume_url": wmill.get_resume_url(),  # For approval continuation
        "message": f"Expense pending {approval_level.value} approval"
    }


def determine_approval_level(amount: float, category: str) -> ApprovalLevel:
    """Determine required approval level based on business rules."""
    # Category-specific rules
    high_scrutiny_categories = ["travel", "equipment", "consulting"]

    if amount <= 100:
        return ApprovalLevel.AUTO
    elif amount <= 1000:
        return ApprovalLevel.MANAGER
    elif amount <= 5000 or category in high_scrutiny_categories:
        return ApprovalLevel.DIRECTOR
    else:
        return ApprovalLevel.EXECUTIVE


def get_approvers(level: ApprovalLevel, requestor: str) -> list:
    """Get list of approvers for given level."""
    approvers_config = wmill.get_variable("u/admin/approvers_config")

    if level == ApprovalLevel.MANAGER:
        # Get requestor's manager
        return approvers_config.get("managers", {}).get(requestor, [])
    elif level == ApprovalLevel.DIRECTOR:
        return approvers_config.get("directors", [])
    elif level == ApprovalLevel.EXECUTIVE:
        return approvers_config.get("executives", [])
    return []


def create_approval_request(db, **kwargs) -> str:
    """Create approval request in database."""
    import psycopg2
    import uuid

    approval_id = str(uuid.uuid4())

    conn = psycopg2.connect(**db)
    try:
        with conn.cursor() as cur:
            cur.execute("""
                INSERT INTO expense_approvals
                (id, expense_id, amount, category, description,
                 requestor, approvers, level, status, created_at)
                VALUES (%s, %s, %s, %s, %s, %s, %s, %s, 'pending', NOW())
            """, (
                approval_id,
                kwargs["expense_id"],
                kwargs["amount"],
                kwargs["category"],
                kwargs["description"],
                kwargs["requestor"],
                kwargs["approvers"],
                kwargs["level"]
            ))
            conn.commit()
    finally:
        conn.close()

    return approval_id


def send_approval_notification(slack, **kwargs):
    """Send Slack notification for approval request."""
    import requests

    blocks = [
        {
            "type": "header",
            "text": {
                "type": "plain_text",
                "text": "Expense Approval Required"
            }
        },
        {
            "type": "section",
            "fields": [
                {"type": "mrkdwn", "text": f"*Amount:* ${kwargs['amount']:.2f}"},
                {"type": "mrkdwn", "text": f"*Category:* {kwargs['category']}"},
                {"type": "mrkdwn", "text": f"*Requestor:* {kwargs['requestor']}"},
                {"type": "mrkdwn", "text": f"*Description:* {kwargs['description']}"}
            ]
        },
        {
            "type": "actions",
            "elements": [
                {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "Review & Approve"},
                    "url": kwargs["approval_link"],
                    "style": "primary"
                }
            ]
        }
    ]

    requests.post(slack["url"], json={"blocks": blocks})

8. Resource and Secrets Management

# scripts/admin/manage_resources.py
"""
Manage Windmill resources and secrets programmatically.
"""

import wmill
from typing import Optional


def main(
    action: str,  # "list", "get", "create", "update", "delete"
    resource_path: Optional[str] = None,
    resource_type: Optional[str] = None,
    resource_value: Optional[dict] = None,
    description: Optional[str] = None,
):
    """
    Manage Windmill resources (secrets, connections, configs).

    Args:
        action: Operation to perform
        resource_path: Path to resource (e.g., "u/admin/my_api")
        resource_type: Type of resource (for create)
        resource_value: Resource value (for create/update)
        description: Resource description

    Returns:
        Operation result
    """
    workspace = wmill.get_workspace()

    if action == "list":
        # List all resources in workspace
        resources = wmill.list_resources()
        return {
            "total": len(resources),
            "resources": [
                {
                    "path": r["path"],
                    "resource_type": r.get("resource_type"),
                    "description": r.get("description", "")
                }
                for r in resources
            ]
        }

    elif action == "get":
        if not resource_path:
            raise ValueError("resource_path required for get action")

        try:
            value = wmill.get_resource(resource_path)
            return {
                "path": resource_path,
                "value": value,
                "found": True
            }
        except Exception as e:
            return {
                "path": resource_path,
                "found": False,
                "error": str(e)
            }

    elif action == "create":
        if not all([resource_path, resource_type, resource_value]):
            raise ValueError("resource_path, resource_type, and resource_value required for create")

        # Validate resource type exists
        valid_types = ["postgresql", "mysql", "mongodb", "s3", "smtp", "slack", "http"]
        if resource_type not in valid_types and not resource_type.startswith("c/"):
            print(f"Warning: Unknown resource type '{resource_type}'")

        # Create resource via API
        result = wmill.create_resource(
            path=resource_path,
            resource_type=resource_type,
            value=resource_value,
            description=description or ""
        )

        return {
            "action": "created",
            "path": resource_path,
            "resource_type": resource_type,
            "success": True
        }

    elif action == "update":
        if not all([resource_path, resource_value]):
            raise ValueError("resource_path and resource_value required for update")

        result = wmill.update_resource(
            path=resource_path,
            value=resource_value
        )

        return {
            "action": "updated",
            "path": resource_path,
            "success": True
        }

    elif action == "delete":
        if not resource_path:
            raise ValueError("resource_path required for delete")

        result = wmill.delete_resource(path=resource_path)

        return {
            "action": "deleted",
            "path": resource_path,
            "success": True
        }

    else:
        raise ValueError(f"Unknown action: {action}")

Integration Examples

Integration with Database and Slack

# scripts/monitoring/database_health_check.py
"""
Monitor database health and alert on issues.
"""

import wmill
from datetime import datetime
import psycopg2


def main(
    check_connections: bool = True,
    check_slow_queries: bool = True,
    slow_query_threshold_ms: int = 5000,
    alert_channel: str = "#database-alerts",
):
    """
    Run database health checks and alert on issues.

    Args:
        check_connections: Check connection pool status
        check_slow_queries: Check for slow running queries
        slow_query_threshold_ms: Threshold for slow query alerts
        alert_channel: Slack channel for alerts

    Returns:
        Health check results
    """
    db = wmill.get_resource("u/admin/production_db")
    slack = wmill.get_resource("u/admin/slack_webhook")

    results = {
        "timestamp": datetime.now().isoformat(),
        "checks": {},
        "alerts": []
    }

    conn = psycopg2.connect(**db)

    try:
        with conn.cursor() as cur:
            # Check active connections
            if check_connections:
                cur.execute("""
                    SELECT
                        count(*) as total,
                        count(*) FILTER (WHERE state = 'active') as active,
                        count(*) FILTER (WHERE state = 'idle') as idle,
                        count(*) FILTER (WHERE state = 'idle in transaction') as idle_in_txn
                    FROM pg_stat_activity
                    WHERE datname = current_database()
                """)
                conn_stats = cur.fetchone()

                results["checks"]["connections"] = {
                    "total": conn_stats[0],
                    "active": conn_stats[1],
                    "idle": conn_stats[2],
                    "idle_in_transaction": conn_stats[3]
                }

                # Alert if too many connections
                if conn_stats[0] > 80:
                    results["alerts"].append({
                        "type": "high_connections",
                        "message": f"High connection count: {conn_stats[0]}/100",
                        "severity": "warning"
                    })

            # Check slow queries
            if check_slow_queries:
                cur.execute("""
                    SELECT
                        pid,
                        now() - pg_stat_activity.query_start AS duration,
                        query,
                        state
                    FROM pg_stat_activity
                    WHERE (now() - pg_stat_activity.query_start) > interval '%s milliseconds'
                    AND state != 'idle'
                    AND query NOT LIKE '%%pg_stat_activity%%'
                """, (slow_query_threshold_ms,))

                slow_queries = cur.fetchall()

                results["checks"]["slow_queries"] = {
                    "count": len(slow_queries),
                    "threshold_ms": slow_query_threshold_ms,
                    "queries": [
                        {
                            "pid": q[0],
                            "duration": str(q[1]),
                            "query": q[2][:200],
                            "state": q[3]
                        }
                        for q in slow_queries[:5]
                    ]
                }

                if slow_queries:
                    results["alerts"].append({
                        "type": "slow_queries",
                        "message": f"Found {len(slow_queries)} slow queries",
                        "severity": "warning"
                    })

    finally:
        conn.close()

    # Send Slack alerts
    if results["alerts"]:
        send_slack_alert(slack, alert_channel, results)

    return results


def send_slack_alert(slack, channel, results):
    """Send health check alerts to Slack."""
    import requests

    alert_texts = [
        f"*{a['severity'].upper()}*: {a['message']}"
        for a in results["alerts"]
    ]

    requests.post(slack["url"], json={
        "channel": channel,
        "blocks": [
            {
                "type": "header",
                "text": {
                    "type": "plain_text",
                    "text": "Database Health Alert"
                }
            },
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": "\n".join(alert_texts)
                }
            },
            {
                "type": "context",
                "elements": [
                    {
                        "type": "mrkdwn",
                        "text": f"Timestamp: {results['timestamp']}"
                    }
                ]
            }
        ]
    })

Best Practices

1. Script Organization

scripts/
├── data/
│   ├── fetch_api_data.py
│   ├── transform_records.ts
│   └── aggregate_metrics.py
├── integrations/
│   ├── sync_crm.py
│   ├── webhook_handler.ts
│   └── slack_notifications.py
├── devops/
│   ├── deploy_service.sh
│   ├── health_checks.py
│   └── cleanup_resources.py
└── admin/
    ├── manage_resources.py
    └── user_management.py

2. Error Handling

import wmill

def main(input_data: dict):
    try:
        result = process_data(input_data)
        return {"success": True, "data": result}
    except ValueError as e:
        # Business logic error - don't retry
        wmill.set_state({"error": str(e), "retryable": False})
        raise
    except ConnectionError as e:
        # Transient error - allow retry
        wmill.set_state({"error": str(e), "retryable": True})
        raise

3. Resource Management

# Always use resources for credentials
api_key = wmill.get_resource("u/admin/api_key")  # Good
# api_key = "sk-1234567890"  # Never hardcode

# Use variables for configuration
config = wmill.get_variable("u/admin/app_config")

4. Testing Scripts

# Test script locally
wmill script run f/data/fetch_api_data \
  --data '{"api_endpoint": "https://api.example.com", "limit": 10}'

# Run with specific resource
wmill script run f/data/fetch_api_data \
  --resource u/admin/test_api_credentials

Troubleshooting

Common Issues

Issue: Script timeout

# Increase timeout in script metadata
# At top of script file:
# extra_perms:
#   timeout: 600  # 10 minutes

Issue: Import errors in Python

# Add dependencies to script header
# requirements:
# pandas==2.0.0
# requests==2.31.0

import pandas as pd
import requests

Issue: Resource not found

# List available resources
wmill resource list

# Check resource path
wmill resource get u/admin/my_resource

Debugging Tips

# Add logging
import wmill

def main(data: dict):
    print(f"Input: {data}")  # Shows in logs
    wmill.set_state({"debug": "checkpoint_1"})

    result = process(data)

    print(f"Result: {result}")
    return result

# Check script state
# wmill script get-state f/my_script

# View recent runs
wmill job list --script f/data/my_script --limit 10

# Get job logs
wmill job get <job_id>

Version History

Version	Date	Changes
1.0.0	2026-01-17	Initial release with comprehensive workflow patterns

Resources

This skill provides production-ready patterns for Windmill workflow automation, tested across enterprise scenarios with Python, TypeScript, Go, and Bash scripts.

windmill

About