# 风车工作流程自动化技能 Master Windmill 实现开发人员优先的工作流程自动化，将脚本转换为具有自动生成 UI 的生产工作流程。该技能涵盖 Python/TypeScript/Go/Bash 中的脚本编写、流程编排、审批流程、计划和企业部署模式。 ## 何时使用此技能 ### 在以下情况下使用： * 开发人员更喜欢编写代码而不是可视化工具 * 需要自动生成脚本参数的 UI * 用最少的前端工作构建内部工具 * Python、TypeScript、Go 或 Bash 是主要语言 * 将工作流程自动化与内部工具相结合 * 需要自动化的代码审查和版本控制 * 需要带有审计跟踪的批准流程 * 自托管数据主权 ### 不要在以下情况下使用： * 非开发人员需要构建工作流程（使用n8n、Activepieces） * 需要 400 多个预构建集成（使用 n8n） * 具有依赖关系的复杂 DAG 编排（使用 Airflow） * CI/CD 管道与 git 紧密耦合（使用 GitHub Actions） * 首选简单的视觉自动化（使用 Activepieces） ## 先决条件 ### 安装选项 **选项 1：Docker Compose（推荐）** # docker-compose.yml 版本：'3.8' 服务：风车：图片：ghcr.io/windmill-labs/windmill:main 重新启动：始终端口： - “8000:8000” 环境： - DATABASE_URL=postgres://windmill:${POSTGRES_PASSWORD}@postgres:5432/windmill?sslmode=禁用 - 模式=独立 - BASE_URL=http://localhost:8000 - RUST_LOG=信息 - NUM_WORKERS=4 - DISABLE_SERVER=假 - DISABLE_WORKERS=假取决于： postgres：条件：服务健康卷： -worker_dependency_cache:/tmp/windmill/cache postgres：图片：postgres：15 重新启动：始终环境： - POSTGRES_USER=风车 - POSTGRES_PASSWORD=${POSTGRES_PASSWORD} - POSTGRES_DB=风车卷： - postgres_data:/var/lib/postgresql/data 健康检查：测试：[“CMD-SHELL”，“pg_isready -U风车”] 间隔：10秒超时：5秒重试次数：5 LSP：图片：ghcr.io/windmill-labs/windmill-lsp:latest 重新启动：始终端口： - “3001:3001” 卷： - lsp_cache:/root/.cache 卷： postgres_数据：工人依赖缓存： lsp_缓存： # 生成密码导出 POSTGRES_PASSWORD=$(openssl rand -hex 32) # 启动服务 docker 组成-d # 通过 http://localhost:8000 访问 UI # 默认凭据：admin@windmill.dev/changeme **选项 2：带有 Helm 的 Kubernetes** # 添加 Windmill Helm 存储库 helm 仓库添加风车 https://windmill-labs.github.io/windmill-helm-charts 舵回购更新 # 创建命名空间 kubectl 创建命名空间风车 # 创建秘密 kubectl 创建秘密通用风车秘密 \ --命名空间风车 \ --from-literal=postgres-password=$(openssl rand -hex 32) # 安装风车 helm 安装风车风车/风车 \ --命名空间风车 \ --set postgresql.auth.existingSecret=windmill-secrets \ --set Windmill.baseUrl=https://windmill.example.com \ --设置windmill.workers.replicas=3 # 获取负载均衡器IP kubectl get svc -n 风车风车应用程序 **选项 3：本地开发** # 安装 Windmill CLI npm install -g 风车-cli # 或者用点 pip 安装 wmill # 登录实例 wmill 工作区添加 my-workspace https://windmill.example.com wmill 工作空间切换 my-workspace # 初始化项目 wmill初始化 # 将本地脚本同步到 Windmill wmill 同步推送 ### 开发设置 # 安装特定于语言的依赖项 # Python 开发 pip install wmill pandas numpy requests # TypeScript/Deno 开发 # Windmill 对 TypeScript 使用 Deno 运行时版本号 # 围棋开发去安装 github.com/windmill-labs/windmill-go-client@latest # Bash 脚本开箱即用 ## 核心能力 ### 1\. Python脚本 # 脚本/data_processing/fetch_and_transform.py ”“” 从 API 获取数据并进行转换以进行分析。自动生成带有所有参数输入字段的 UI。 ”“” 导入wmill 从日期时间导入日期时间，时间增量导入请求将 pandas 导入为 pd 定义主要（ api_端点：str，日期范围天数：int = 7， include_metadata: 布尔 = True, output_format: str = "json", # 下拉列表：json、csv、parquet 过滤器：字典=无，）： ”“” 从外部 API 获取和转换数据。参数： api_endpoint：从中获取数据的 API 端点 URL date_range_days：要获取数据的天数（默认值：7） include_metadata：响应中是否包含元数据 output_format：输出格式 - json、csv 或 parquet 过滤器：应用于数据的可选过滤器返回：将数据转换为指定格式 ”“” # 从 Windmill 资源获取 API 凭据 api_credentials = wmill.get_resource(“u/admin/api_credentials”) # 计算日期范围结束日期 = datetime.now() 开始日期=结束日期-时间增量（天=日期范围天） # 获取数据标题= { "授权": f"承载{api_credentials['api_key']}", “内容类型”：“应用程序/json” } 参数 = { “开始日期”：开始日期.isoformat(), “结束日期”：结束日期.isoformat(), } 如果过滤器： params.update(过滤器) 响应 = requests.get( f"{api_endpoint}/数据", 标题=标题，参数=参数，超时=30 ）响应.raise_for_status() 数据 = 响应.json() # 使用 pandas 进行变换 df = pd.DataFrame(data["记录"]) # 应用转换如果 df.columns 中有“时间戳”： df["时间戳"] = pd.to_datetime(df["时间戳"]) df["日期"] = df["时间戳"].dt.日期 df["小时"] = df["时间戳"].dt.小时如果 df.columns 中有“值”： df["value_normalized"] = (df["value"] - df["value"].min()) / ( df["值"].max() - df["值"].min() ） # 生成汇总统计数据摘要= { “总记录”：len（df）， “日期范围”：{ “开始”：str(start_date.date()), “结束”：str(end_date.date()) }, "统计": df.describe().to_dict() 如果不是 df.empty else {} } # 格式化输出如果输出格式==“json”：结果 = df.to_dict(orient="记录") elif 输出格式==“csv”：结果 = df.to_csv(index=False) 其他： # 对于 parquet，以 dict 返回（Windmill 处理序列化）结果 = df.to_dict(orient="记录") 如果包含元数据：返回{ “数据”：结果， “元数据”：摘要， “格式”：输出格式， “generate_at”：datetime.now().isoformat() } 返回结果 # 脚本/集成/sync_crm_to_database.py ”“” 通过重复数据删除将 CRM 联系人同步到内部数据库。 ”“” 导入wmill 从输入 import 可选导入psycopg2 从 psycopg2.extras 导入execute_values 定义主要（ crm_list_id：str，批量大小：int = 100， dry_run：布尔=假， update_existing：布尔= True，）： ”“” 将 CRM 联系人同步到 PostgreSQL 数据库。参数： crm_list_id：要同步的 CRM 列表 ID batch_size：每批的记录数 dry_run：如果为 True，则实际上不写入数据库 update_existing：如果为 True，则更新现有记录返回：同步统计数据 ”“” # 获取资源 crm_api = wmill.get_resource(“u/admin/crm_api”) db_conn = wmill.get_resource("u/admin/postgres_warehouse") # 从 CRM 中获取联系人导入请求联系人 = [] 页数 = 1 而真实：响应 = requests.get( f"{crm_api['base_url']}/lists/{crm_list_id}/contacts", headers={"授权": f"承载 {crm_api['api_key']}"}, params={“page”：页面，“per_page”：batch_size} ）响应.raise_for_status() 数据 = 响应.json() 联系人.扩展（数据[“联系人”]）如果不是 data.get("has_more"): 打破页 += 1 print(f"从 CRM 中获取了 {len(contacts)} 联系人") 如果干运行：返回{ “模式”：“干运行”， “contacts_fetched”：len（联系人）， “样本”：联系人[:5] } # 连接数据库 conn = psycopg2.connect（主机=db_conn["主机"], 端口=db_conn[“端口”], 数据库=db_conn["数据库"], 用户=db_conn["用户"], 密码=db_conn["密码"] ）统计= {“插入”：0，“更新”：0，“跳过”：0，“错误”：[]} 尝试：将 conn.cursor() 作为 cur：对于联系人中的联系人：尝试： # 检查是否存在当前.执行( “从联系人中选择 ID，其中电子邮件 = %s”，（联系方式[“电子邮件”]，））现有 = cur.fetchone() 如果存在：如果更新现有： cur.执行(""" 更新联系人 SET 名称 = %s，公司 = %s，电话 = %s， update_at = NOW()，crm_id = %s 其中电子邮件 = %s “””，（联系人[“姓名”], contact.get("公司"), contact.get("电话"), 联系人[“id”], 联系方式[“电子邮件”] ））统计数据[“已更新”] += 1 其他：统计数据[“跳过”] += 1 其他： cur.执行(""" 插入联系人（电子邮件、姓名、公司、电话、crm_id、创建时间）值（%s、%s、%s、%s、%s、NOW()） “””，（联系方式[“电子邮件”]，联系人[“姓名”], contact.get("公司"), contact.get("电话"), 联系方式[“id”] ））统计数据[“插入”] += 1 除了异常 e：统计数据[“错误”].append({ “联系人”：联系人[“电子邮件”]， “错误”：str(e) }) conn.commit() 最后： conn.close() 返回{ “模式”：“直播”， “contacts_fetched”：len（联系人）， **统计数据 } ### 2\. TypeScript/Deno 脚本 // 脚本/api/webhook_handler.ts /** * 通过验证和路由处理传入的 webhook。 * 使用具有 TypeScript 支持的 Deno 运行时。 */ 从“npm:windmill-client@1”导入*作为wmill； // 定义自动生成的 UI 的输入类型类型 WebhookPayload = { 事件类型：字符串；数据：记录<字符串，未知>；时间戳：字符串；签名？：字符串； }; 类型处理程序配置 = { validate_signature：布尔值；允许的事件：字符串[]； forward_to_slack：布尔值； }; 导出异步函数 main( 有效负载：WebhookPayload，配置：HandlerConfig = { 验证签名：正确， allowed_events: ["订单.创建", "订单.更新", "付款.完成"], forward_to_slack：正确， } ): 承诺<{ 已处理：布尔值；事件类型：字符串； actions_taken：字符串[]； }> { const 操作： string[] = []； // 从资源中获取 webhook 秘密 const webhookSecret =等待 wmill.getResource("u/admin/webhook_secret"); // 如果需要验证签名 if (config.validate_signature && payload.signature) { const crypto =等待导入（“节点：加密”）; const 预期签名 = 加密 .createHmac("sha256", webhookSecret.secret) .update(JSON.stringify(payload.data)) .digest(“十六进制”); if (payload.signature !== ExpectedSignature) { throw new Error("无效的 webhook 签名"); } actions.push("signature_validated"); } // 检查事件是否允许 if (!config.allowed_events.includes(payload.event_type)) { 返回{ 已处理：假， event_type：payload.event_type， actions_taken：[“event_filtered”]， }; } // 根据事件类型进行路由开关（payload.event_type）{ 案例“订单.创建”：等待handleOrderCreated(payload.data); actions.push("order_processed"); 休息; 案例“order.updated”：等待handleOrderUpdated(payload.data); actions.push("order_updated"); 休息; 案例“付款.完成”：等待handlePaymentCompleted(payload.data); actions.push(" payment_recorded"); 休息; } // 如果配置则转发到 Slack 如果（config.forward_to_slack）{ const slackWebhook = 等待 wmill.getResource("u/admin/slack_webhook"); 等待获取（slackWebhook.url，{ 方法：“POST”， headers: { "Content-Type": "application/json" }, 正文：JSON.stringify({ text: `Webhook 收到：${payload.event_type}`, 块：[ { 类型：“部分”，文本：{ 类型：“mrkdwn”，文本：`*事件：* ${payload.event_type}\n*时间戳：* ${payload.timestamp}`, }, }, ], }), }); actions.push("slack_notified"); } 返回{ 处理：真实， event_type：payload.event_type， actions_taken：采取的行动， }; } 异步函数handleOrderCreated（数据：记录<字符串，未知>）{ console.log("处理新订单：", data); // 实现 } 异步函数handleOrderUpdated（数据：记录<字符串，未知>）{ console.log("处理订单更新：", data); // 实现 } 异步函数handlePaymentCompleted（数据：记录<字符串，未知>）{ console.log("正在处理支付：", data); // 实现 } // 脚本/data/aggregate_metrics.ts /** * 将多个来源的指标聚合到统一的仪表板数据中。 */ 从“npm:windmill-client@1”导入*作为wmill；类型 MetricsSource = "数据库" | “API”| “缓存”; type AggregationPeriod = "每小时" | “每日” | “每周”| “每月”；接口度量配置{ 来源：MetricsSource[]； period：聚合期； include_comparisons：布尔值；自定义尺寸？：字符串[]； } 接口聚合指标{ 句点：字符串；总收入：数量；订单总数：数量；平均订单值：数字； unique_customers：数量； top_products: 数组<{ 名称: 字符串;收入：数量；数量：数量 }>; by_dimension：记录<字符串，记录<字符串，数字>>；比较？：{ previous_period: 记录<字符串, 数字>; change_percent：记录<字符串，数字>； }; } 导出异步函数 main( 开始日期：字符串，结束日期：字符串，配置：MetricConfig = { 来源：[“数据库”]，期间：“每日”， include_comparisons：正确， } ): Promise<聚合指标> { // 获取数据库连接 const dbConfig = 等待 wmill.getResource("u/admin/analytics_db"); // 数据库客户端动态导入 const { 客户端 } = 等待导入("npm:pg@8"); const client = new Client(dbConfig); 等待客户端.connect(); 尝试{ // 获取基本指标 const 指标查询 = ` 选择 DATE_TRUNC('${config.period}',created_at)作为周期， COUNT(*) 作为总订单数， SUM(total_amount) 作为total_revenue， COUNT(DISTINCT customer_id) 作为 unique_customers 来自订单其中created_at >= $1 并且created_at < $2 GROUP BY DATE_TRUNC('${config.period}',created_at) 按期间排序 `; constmetricsResult = 等待 client.query(metricsQuery, [ 开始日期，结束日期， ]); // 跨时期聚合 const 总收入 =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 ）； // 获取热门产品 const topProductsQuery = ` 选择 p.姓名， SUM(oi.quantity) 作为数量， SUM(oi.quantity * oi.unit_price) 作为收入来自 order_items oi 连接产品 p ON oi.product_id = p.id JOIN 订单 o ON oi.order_id = o.id 其中 o.created_at >= $1 并且 o.created_at < $2 按 p.id、p.name 分组按收入 DESC 排序限制 10 `; const topProductsResult = 等待 client.query(topProductsQuery, [ 开始日期，结束日期， ]); const 结果：AggregateMetrics = { 期间：`${start_date} 到 ${end_date}`，总收入：总收入，订单总数：订单总数， avg_order_value：总订单数 > 0 ？总收入/总订单数：0， unique_customers：独特的客户， top_products: topProductsResult.rows.map((r) => ({ 名称：r.name，收入：parseFloat(r.revenue), 数量：parseInt(r.quantity), }））， by_dimension：{}， }; // 添加维度细分如果（config.custom_dimensions）{ for (config.custom_dimensions 的 const 维度) { 常量维度查询 = ` 选择 ${维度}， SUM(total_amount) 作为收入来自订单其中created_at >= $1 并且created_at < $2 按${维度}分组 `; const dimResult = 等待 client.query(dimensionQuery, [ 开始日期，结束日期， ]); result.by_dimension[维度] = Object.fromEntries( dimResult.rows.map((r) => [r[维度], parseFloat(r.revenue)]) ）； } } // 添加周期比较如果（config.include_comparisons）{ const periodDays = Math.ceil( (新日期(结束日期).getTime() - 新日期(开始日期).getTime()) / (1000*60*60*24) ）； const prevStart = 新日期( 新日期(start_date).getTime() - periodDays * 24 * 60 * 60 * 1000 ） .toISOString() .split("T")[0]; const prevEnd = 开始日期； const prevQuery = ` 选择 COUNT(*) 作为总订单数， COALESCE(SUM(total_amount), 0) 作为total_revenue， COUNT(DISTINCT customer_id) 作为 unique_customers 来自订单其中created_at >= $1 并且created_at < $2 `; const prevResult = 等待 client.query(prevQuery, [prevStart, prevEnd]); const prev = prevResult.rows[0]; const prevRevenue = parseFloat(prev.total_revenue || 0); const prevOrders = parseInt(prev.total_orders || 0); 结果.比较 = { 上一个时期：{ 收入：prev收入，订单：上一个订单， }, 变化百分比：{ 收入：prev收入 > 0 ？ ((总收入 - 上一个收入) / 上一个收入) * 100 : 0, 订单：prevOrders > 0 ? ((总订单数 - 上一个订单数) / 上一个订单数) * 100 : 0, }, }; } 返回结果； }最后{ 等待 client.end(); } } ### 3\. Go 脚本 // 脚本/性能/batch_processor.go // 使用 Go 进行高性能批处理。包装内导入（ “上下文” “编码/json” “FMMT” “同步” “时间” wmill“github.com/windmill-labs/windmill-go-client” ）类型 BatchConfig 结构 { BatchSize int `json:"batch_size"` 并发 int `json:"并发"` RetryAttempts int `json:"retry_attempts"` TimeoutSecs int `json:"timeout_secs"` } 类型 ProcessResult 结构体 { TotalItems int `json:"total_items"` 成功 int `json:"成功"` 失败 int `json:"失败"` 处理时间 float64 `json:"processing_time_seconds"` 错误 []ProcessError `json:"errors,omitempty"` } 类型 ProcessError 结构体 { ItemID 字符串 `json:"item_id"` 错误字符串 `json:"error"` } 函数主（项目[]地图[字符串]界面{}，配置批量配置， ) (*处理结果，错误) { 开始时间 := 时间.Now() // 设置默认值如果 config.BatchSize == 0 { 配置.BatchSize = 100 } 如果config.Concurrency == 0 { 配置.并发 = 4 } 如果 config.RetryAttempts == 0 { 配置.RetryAttempts = 3 } 如果 config.TimeoutSecs == 0 { 配置.TimeoutSecs = 30 } // 获取API凭证 ctx := context.Background() 资源，err := wmill.GetResource("u/admin/processing_api") 如果错误！= nil { return nil, fmt.Errorf("获取资源失败：%w", err) } apiKey := 资源["api_key"].(字符串) 结果 := &处理结果{ 总项目数: len(项目), } var musync.Mutex var wgsync.WaitGroup // 创建工作池 itemChan := make(chan map[string]interface{}, config.BatchSize) errorChan := make(chan ProcessError, len(items)) // 启动工人对于我：= 0； i < config.Concurrency;我++ { 工作组.添加(1) 去函数（）{ 推迟 wg.Done() for item := range itemChan { 错误：= processItem（ctx，项目，apiKey，配置） mu.Lock() 如果错误！= nil { 结果.失败++ errorChan <- 进程错误{ ItemID: fmt.Sprintf("%v", item["id"]), 错误：err.Error(), } } 否则{ 结果.成功++ } mu.Unlock() } }() } // 发送项目给worker 对于 _, item := 范围项 { itemChan <- 项目 } 关闭（itemChan） // 等待所有worker完成 wg.Wait() 关闭（错误Chan） // 收集错误 for err := 范围 errorChan { 结果.错误 = 追加(结果.错误, 错误) } result.ProcessingTime = time.Since(startTime).Seconds() 返回结果，无 } func 进程项（ ctx 上下文.Context, 项目地图[string]接口{}， apiKey 字符串，配置批量配置，）错误{ var LastErr 错误对于尝试：= 1；尝试 <= config.RetryAttempts;尝试++ { ctx, 取消 := context.WithTimeout(ctx, time.Duration(config.TimeoutSecs)*time.Second) 推迟取消() // 处理项（简化 - 实际实现将进行 API 调用）选择{ 案例 <-ctx.Done(): 最后的错误 = ctx.Err() 默认： // 模拟处理时间.睡眠(10 * 时间.毫秒) // 验证项目 if _, ok := item["id"]; ！好吧{ return fmt.Errorf("缺少必填字段：id") } 返回零 } // 重试前的指数退避如果尝试 < config.RetryAttempts { time.Sleep(time.Duration(1<<尝试) * time.Second) } } return fmt.Errorf("%d 次尝试后失败：%w", config.RetryAttempts, lastErr) } ### 4\. bash脚本 #!/bin/bash # 脚本/devops/deploy_service.sh # 部署具有健康检查和回滚能力的服务。 # Windmill 自动提供这些作为环境变量 # SERVICE_NAME、版本、环境、DRY_RUN 设置-euo管道故障 # 从 Windmill 资源中获取秘密 DEPLOY_KEY=$(curl -s -H "授权：持有者 $WM_TOKEN" \ “$BASE_INTERNAL_URL/api/w/$WM_WORKSPACE/resources/get/u/admin/deploy_key”| jq -r '.value.key') AWS_REGION=$(curl -s -H "授权：持有者 $WM_TOKEN" \ “$BASE_INTERNAL_URL/api/w/$WM_WORKSPACE/resources/get/u/admin/aws_config”| jq -r '.value.region') 日志（）{ echo "[$(日期 '+%Y-%m-%d %H:%M:%S')] $1" } 部署服务（）{ 本地服务=1美元本地版本=$2 本地环境=$3 记录“将$service版本$version部署到$env” # 更新ECS服务如果[[“$DRY_RUN”==“true”]];然后 log“[DRY RUN] 会将 ECS 服务 $service 更新到 $version” 返回0 菲 aws ecs 更新服务 \ --cluster "${env}-cluster" \ --服务“$服务”\ --强制新部署 \ --区域“$AWS_REGION” 记录“已启动部署” } 等待健康（）{ 本地服务=1美元本地环境=$2 本地 max_wait=300 本地间隔=10 本地已过去=0 log“等待$service变得健康......” while [[ $elapsed -lt $max_wait ]];做本地 running_count=$(aws ecs 描述服务 \ --cluster "${env}-cluster" \ --服务“$服务”\ --区域“$AWS_REGION”\ --query 'services[0].runningCount' \ --输出文本）本地desired_count = $（aws ecs描述服务\ --cluster "${env}-cluster" \ --服务“$服务”\ --区域“$AWS_REGION”\ --query '服务[0].desiredCount' \ --输出文本） if [[ "$running_count" == "$desired_count" ]];然后记录“服务健康：$running_count/$desired_count 任务正在运行” 返回0 菲 log“正在等待...（$running_count/$desired_count 任务正在运行）” 睡眠$间隔经过=$((经过+间隔)) 完成记录“错误：服务在 ${max_wait} 秒内未变得正常” 返回1 } 回滚（）{ 本地服务=1美元本地环境=$2 记录“在 $env 中回滚 $service” # 获取之前的任务定义本地 prev_task_def=$(aws ecs 描述服务 \ --cluster "${env}-cluster" \ --服务“$服务”\ --区域“$AWS_REGION”\ --query '服务[0].部署[1].taskDefinition' \ --输出文本）如果 [[ -z "$prev_task_def" || "$prev_task_def" == "无"]];然后记录“错误：找不到用于回滚的先前任务定义” 返回1 菲 aws ecs 更新服务 \ --cluster "${env}-cluster" \ --服务“$服务”\ --任务定义“$prev_task_def”\ --区域“$AWS_REGION” 记录“已启动回滚到 $prev_task_def” } # 主要执行主要（）{ 日志“===部署开始===” 记录“服务：$SERVICE_NAME” 记录“版本：$VERSION” 记录“环境：$ENVIRONMENT” 记录“试运行：$DRY_RUN” # 部署如果！部署服务“$SERVICE_NAME”“$VERSION”“$ENVIRONMENT”；然后记录“错误：部署失败” 1号出口菲 # 等待健康（在试运行中跳过） if [[ "$DRY_RUN" != "true" ]];然后如果！ wait_for_healthy "$SERVICE_NAME" "$ENVIRONMENT";然后记录“部署运行状况检查失败，正在启动回滚” 回滚“$SERVICE_NAME”“$ENVIRONMENT” 1号出口菲菲日志“===部署成功完成===” # Windmill 将结果输出为 JSON 猫< 5000 模块： - ID：请求批准值：类型：审批超时: 86400 # 24 小时批准者： - admin@example.com - manager@example.com - id：process_approved 值：类型：脚本路径：f/付款/process_ payment 输入变换：订单 ID：类型： JavaScript expr: results.validate_order.order_id 金额：类型： JavaScript expr: results.validate_order.total - 摘要：正常顺序 expr: results.validate_order.total <= 5000 模块： - id：process_normal 值：类型：脚本路径：f/付款/process_ payment 输入变换：订单 ID：类型： JavaScript expr: results.validate_order.order_id 金额：类型： JavaScript expr: results.validate_order.total - id: 创建货件值：类型：脚本路径：f/fulfillment/create_shipment 输入变换：订单 ID：类型： JavaScript expr: results.validate_order.order_id 送货地址：类型： JavaScript expr: results.validate_order.shipping_address - id: 发送确认值：类型：脚本路径：f/notifications/send_order_confirmation 输入变换：电子邮件：类型： JavaScript expr: results.validate_order.customer_email 订单详情：类型： JavaScript 表达式: | { order_id: results.validate_order.order_id, 总计：结果.validate_order.total，跟踪编号：结果.创建发货.跟踪编号 } 架构： $架构：https://json-schema.org/draft/2020-12/schema 类型：对象属性：订单：类型：对象属性：客户电子邮件：类型：字符串格式：电子邮件项目：类型：数组项目：类型：对象送货地址：类型：对象需要： - 客户_电子邮件 - 物品 - 送货地址需要： - 订单 ### 6\.日程管理 # 脚本/调度/dynamic_scheduler.py ”“” 根据业务规则动态管理计划。 ”“” 导入wmill 从日期时间导入日期时间，时间增量从输入导入列表，可选定义主要（ Schedule_configs：列表[字典]， dry_run：布尔=真，）： ”“” 根据配置更新 Windmill 计划。参数： Schedule_configs：计划配置列表 dry_run：如果为 True，则仅报告会发生变化的内容返回：时间表变更摘要 ”“” 客户端 = wmill.Client() 工作空间 = wmill.get_workspace() 变化=[] 对于schedule_configs中的配置：计划路径=配置[“路径”] 启用 = config.get("启用", True) cron = config.get("cron") 时区 = config.get("时区", "UTC") # 检查营业时间限制 if config.get("business_hours_only", False): # 修改 cron 使其仅在工作时间运行（9-17）如果计划任务： cron_parts = cron.split() 如果 len(cron_parts) >= 5: cron_parts[1] = "9-17" # 小时 cron_parts[4] = "1-5" # 仅工作日 cron =“”.join(cron_parts) # 检查维护窗口约束 if config.get("skip_maintenance_windows", False): 维护 = get_maintenance_windows() 现在 = 日期时间.now() in_maintenance = 任何( m["开始"] <= 现在 <= m["结束"] 维护中的 m ）如果在维护中：启用=假改变={ “路径”：schedule_path， “克朗”：克朗， “时区”：时区， “启用”：启用， “干运行”：干运行 } 如果不是dry_run： # 通过 Windmill API 更新日程尝试：客户端.update_schedule( 工作空间=工作空间，路径=计划路径，时间表={ “计划”：cron， “时区”：时区， “启用”：启用 } ）更改[“状态”] =“已更新” 除了异常 e：更改[“状态”] =“错误” 更改[“错误”] = str(e) 其他：更改[“状态”] =“would_update” 更改.追加（更改）返回{ “total_schedules”：len（schedule_configs）， “变化”：改变， “干运行”：干运行 } def get_maintenance_windows(): """从配置中获取维护窗口。""" 尝试： config = wmill.get_variable("u/admin/maintenance_windows") 返回 config.get("windows", []) 除外：返回[] ### 7\.审批流程 # 脚本/批准/expense_approval.py ”“” 具有多级审批的费用审批工作流程。 ”“” 导入wmill 从输入 import 可选从枚举导入枚举类批准级别（枚举）：自动=“自动” 经理=“经理” 导演=“导演” 执行=“执行” 定义主要（费用_id：str，金额：浮动，类别： STR, 描述：str，请求者电子邮件：str，收据：可选[列表] = 无，）： ”“” 处理费用批准请求。参数： cost_id：唯一的费用ID amount：费用金额（美元）类别：费用类别描述：费用描述 requestor_email: 请求费用者的电子邮件收据：收据文件 URL 列表返回：批准请求状态和后续步骤 ”“” # 根据金额和类别确定审批级别批准级别=确定批准级别（金额，类别） # 获取该级别的批准者批准者 = get_approvers(approval_level, requestor_email) 如果approval_level == ApprovalLevel.AUTO： # 自动批准小额开支返回{ “费用_id”：费用_id， “状态”：“已批准”， “批准级别”：批准级别.值， “auto_approved”：正确， "message": f"费用自动批准（金额：${amount:.2f}）" } # 在数据库中创建批准请求 db = wmill.get_resource("u/admin/expenses_db") 批准 ID = 创建批准请求（分贝，费用_id=费用_id, 金额=金额，类别=类别，描述=描述，请求者=请求者_电子邮件，批准者=批准者，级别=approval_level.value ） # 向审批者发送通知 slack = wmill.get_resource(“u/admin/slack_webhook”) 发送批准通知（松弛，费用_id=费用_id, 金额=金额，类别=类别，描述=描述，请求者=请求者_电子邮件，批准者=批准者，批准_link = f“https://windmill.example.com/approvals/{approval_id}” ） # 返回 Windmill 内置审批步骤的审批 URL 返回{ “费用_id”：费用_id， “approval_id”：approval_id， “状态”：“待批准”， “批准级别”：批准级别.值， “批准者”：批准者， "resume_url": wmill.get_resume_url(), # 用于批准继续 "message": f"费用待 {approval_level.value} 批准" } def确定_批准级别（金额：float，类别：str）->批准级别： """根据业务规则确定所需的审批级别。""" # 类别特定规则 high_scrutiny_categories = [“旅行”,“装备”,“咨询”] 如果金额 <= 100：返回 ApprovalLevel.AUTO elif 金额 <= 1000: 返回 ApprovalLevel.MANAGER elif 金额 <= 5000 或 high_scrutiny_categories 中的类别：返回 ApprovalLevel.DIRECTOR 其他：返回 ApprovalLevel.EXECUTIVE def get_approvers(级别: ApprovalLevel, 请求者: str) -> 列表: """获取给定级别的审批者列表。""" approvers_config = wmill.get_variable(“u/admin/approvers_config”) 如果级别 == ApprovalLevel.MANAGER： # 获取请求者的经理 returnapprovers_config.get("managers", {}).get(requestor, []) elif 级别 == ApprovalLevel.DIRECTOR: 返回approvers_config.get（“董事”，[]） elif 级别 == ApprovalLevel.EXECUTIVE：返回approvers_config.get（“执行人员”，[]）返回[] def create_approval_request(db, **kwargs) -> str: """在数据库中创建批准请求。""" 导入psycopg2 导入uuid 批准id = str(uuid.uuid4()) conn = psycopg2.connect(**db) 尝试：将 conn.cursor() 作为 cur： cur.执行(""" 插入费用_批准（id、费用_id、金额、类别、描述、请求者、批准者、级别、状态、创建时间）值（%s、%s、%s、%s、%s、%s、%s、%s、'待处理'、NOW()） “””，（批准_id， kwargs["expense_id"], kwargs[“金额”]， kwargs["类别"], kwargs[“描述”]， kwargs["请求者"], kwargs["批准者"], kwargs[“级别”] ）） conn.commit() 最后： conn.close() 返回approval_id def send_approval_notification(slack, **kwargs): """发送 Slack 通知以请求批准。""" 导入请求块= [ { “类型”：“标题”， “文本”：{ “类型”：“纯文本”， "text": "需要费用批准" } }, { “类型”：“部分”， “字段”：[ {"type": "mrkdwn", "text": f"*金额:* ${kwargs['amount']:.2f}"}, {"type": "mrkdwn", "text": f"*类别:* {kwargs['category']}"}, {"type": "mrkdwn", "text": f"*请求者:* {kwargs['requestor']}"}, {"type": "mrkdwn", "text": f"*描述:* {kwargs['description']}"} ] }, { “类型”：“动作”， “元素”：[ { “类型”：“按钮”， "text": {"type": "plain_text", "text": "审核并批准"}, “url”：kwargs[“approval_link”]， “风格”：“主要” } ] } ] requests.post(slack["url"], json={"blocks": 块}) ### 8\.资源和秘密管理 # 脚本/admin/manage_resources.py ”“” 以编程方式管理 Windmill 资源和机密。 ”“” 导入wmill 从输入 import 可选定义主要（动作：str，#“列表”，“获取”，“创建”，“更新”，“删除” 资源路径：可选[str] =无，资源类型：可选[str] =无，资源值：可选[字典] = 无，描述：可选[str] =无，）： ”“” 管理 Windmill 资源（秘密、连接、配置）。参数： action：要执行的操作 resources_path：资源路径（例如“u/admin/my_api”） resource_type：资源类型（用于创建） resource_value：资源值（用于创建/更新）描述：资源描述返回：运算结果 ”“” 工作空间 = wmill.get_workspace() 如果操作==“列表”： # 列出工作空间中的所有资源资源 = wmill.list_resources() 返回{ “总计”：len（资源）， “资源”：[ { “路径”：r[“路径”]， "资源类型": r.get("资源类型"), “描述”：r.get(“描述”，“”) } 对于资源中的 r ] } elif 操作==“获取”：如果不是资源路径：引发 ValueError("获取操作所需的资源路径") 尝试：值 = wmill.get_resource(resource_path) 返回{ “路径”：资源路径， “价值”：价值， “发现”：真实 } 除了异常 e：返回{ “路径”：资源路径， “发现”：错误， “错误”：str(e) } elif 操作==“创建”：如果不是全部（[资源路径，资源类型，资源值]）： raise ValueError("创建所需的资源路径、资源类型和资源值") # 验证资源类型是否存在 valid_types = [“postgresql”，“mysql”，“mongodb”，“s3”，“smtp”，“slack”，“http”] 如果resource_type不在valid_types中且不在resource_type.startswith("c/")中： print(f"警告：未知资源类型'{resource_type}'") # 通过API创建资源结果 = wmill.create_resource( 路径=资源路径，资源类型=资源类型，值=资源值，描述=描述或“” ）返回{ “行动”：“创建”， “路径”：资源路径， “资源类型”：资源类型， “成功”：正确 } elif 操作==“更新”：如果不是全部（[资源路径，资源值]）： raise ValueError(“更新所需的资源路径和资源值”) 结果 = wmill.update_resource( 路径=资源路径，值=资源值）返回{ “行动”：“已更新”， “路径”：资源路径， “成功”：正确 } elif 操作==“删除”：如果不是资源路径： raise ValueError("删除需要resource_path") 结果 = wmill.delete_resource(path=resource_path) 返回{ "action": "已删除", “路径”：资源路径， “成功”：正确 } 其他：引发 ValueError(f"未知操作：{action}") ## 集成示例 ### 与数据库和 Slack 集成 # 脚本/监控/database_health_check.py ”“” 监控数据库运行状况并就问题发出警报。 ”“” 导入wmill 从日期时间导入日期时间导入psycopg2 定义主要（检查连接：布尔=真， check_slow_queries：布尔= True， Slow_query_threshold_ms：int = 5000， Alert_channel: str = "#database-alerts", ）： ”“” 运行数据库健康检查并就问题发出警报。参数： check_connections：检查连接池状态 check_slow_queries：检查运行缓慢的查询 Slow_query_threshold_ms：慢查询警报的阈值 Alert_channel：警报的 Slack 通道返回：健康检查结果 ”“” db = wmill.get_resource("u/admin/product_db") slack = wmill.get_resource(“u/admin/slack_webhook”) 结果={ “时间戳”：datetime.now().isoformat(), “检查”：{}， “警报”：[] } conn = psycopg2.connect(**db) 尝试：将 conn.cursor() 作为 cur： # 检查活动连接如果检查连接： cur.执行(""" 选择计数(*)作为总数， count(*) FILTER (WHERE state = 'active') 为活动状态， count(*) FILTER (WHERE state = 'idle') 为空闲， count(*) FILTER (WHERE state = 'idle in transaction') asidle_in_txn 来自 pg_stat_activity WHERE 数据名 = current_database() “””） conn_stats = cur.fetchone() 结果[“检查”][“连接”] = { “总计”：conn_stats[0]， “活动”：conn_stats[1]， “空闲”：conn_stats[2]， “idle_in_transaction”：conn_stats[3] } # 如果连接过多则发出警报如果 conn_stats[0] > 80：结果[“警报”].append({ “类型”：“高连接”， "message": f"连接数高: {conn_stats[0]}/100", “严重性”：“警告” }) # 检查慢查询如果检查慢查询： cur.执行(""" 选择进程号， now() - pg_stat_activity.query_start AS 持续时间，查询，状态来自 pg_stat_activity WHERE (now() - pg_stat_activity.query_start) > 间隔 '%s 毫秒' AND 状态 != '空闲' AND 查询 NOT LIKE '%%pg_stat_activity%%' """, (slow_query_threshold_ms,)) Slow_queries = cur.fetchall() 结果[“检查”][“slow_queries”] = { “计数”：len（slow_queries）， “threshold_ms”：slow_query_threshold_ms， “查询”：[ { “pid”：q[0]， “持续时间”：str(q[1]), “查询”：q[2][:200]， “状态”：q[3] } 对于 Slow_queries[:5] 中的 q ] } 如果慢查询：结果[“警报”].append({ “类型”：“慢速查询”， "message": f"发现 {len(slow_queries)} 慢查询", “严重性”：“警告” }) 最后： conn.close() # 发送 Slack 警报如果结果[“警报”]： send_slack_alert（松弛，alert_channel，结果）返回结果 def send_slack_alert(slack, 通道, 结果): """向 Slack 发送健康检查警报。""" 导入请求警报文本 = [ f"*{a['严重性'].upper()}*: {a['消息']}" 对于结果[“警报”] ] requests.post(slack["url"], json={ “频道”：频道， “块”：[ { “类型”：“标题”， “文本”：{ “类型”：“纯文本”， "text": "数据库健康警报" } }, { “类型”：“部分”， “文本”：{ “类型”：“mrkdwn”， "文本": "\n".join(alert_texts) } }, { “类型”：“上下文”， “元素”：[ { “类型”：“mrkdwn”， "text": f"时间戳: {results['timestamp']}" } ] } ] }) ## 最佳实践 ### 1\.脚本组织脚本/ ├── 数据/ │ ├── fetch_api_data.py │ ├── 转换记录.ts │ └──aggregate_metrics.py ├── 整合/ │ ├──sync_crm.py │ ├── webhook_handler.ts │ └── slack_notifications.py ├── devops/ │ ├── 部署_service.sh │ ├── health_checks.py │ └── cleanup_resources.py └── 管理员/ ├── 管理资源.py └── 用户管理.py ### 2\.错误处理导入wmill def main(input_data: dict): 尝试：结果=过程数据（输入数据）返回{“成功”：True，“数据”：结果} 除了 ValueError 为 e： # 业务逻辑错误 - 不要重试 wmill.set_state({"error": str(e), "retryable": False}) 提高除了 ConnectionError 为 e： # 暂时错误 - 允许重试 wmill.set_state({"error": str(e), "retryable": True}) 提高 ### 3\.资源管理 # 始终使用资源作为凭证 api_key = wmill.get_resource("u/admin/api_key") # 好 # api_key = "sk-1234567890" # 切勿硬编码 # 使用变量进行配置 config = wmill.get_variable("u/admin/app_config") ### 4\.测试脚本 # 本地测试脚本 wmill 脚本运行 f/data/fetch_api_data \ --data '{"api_endpoint": "https://api.example.com", "limit": 10}' # 使用特定资源运行 wmill 脚本运行 f/data/fetch_api_data \ --resource u/admin/test_api_credentials ## 故障排除 ### 常见问题 **问题：脚本超时** # 增加脚本元数据的超时时间 # 在脚本文件顶部： # 额外权限： # 超时: 600 # 10 分钟 **问题：Python 中的导入错误** # 将依赖项添加到脚本头 # 要求： # 熊猫==2.0.0 # 请求==2.31.0 将 pandas 导入为 pd 导入请求 **问题：找不到资源** # 列出可用资源 wmill 资源列表 # 检查资源路径 wmill 资源获取 u/admin/my_resource ### 调试技巧 # 添加日志记录导入wmill def main(数据: 字典): print(f"Input: {data}") # 在日志中显示 wmill.set_state({"debug": "checkpoint_1"}) 结果=过程（数据） print(f"结果：{结果}") 返回结果 # 检查脚本状态 # wmill 脚本获取状态 f/my_script # 查看最近的运行情况 wmill 作业列表 --script f/data/my_script --limit 10 # 获取作业日志 wmill 作业获取 ## 版本历史版本 |日期 |变化 ---|---|--- 1.0.0 | 2026-01-17 |具有全面工作流程模式的初始版本 ## 资源 * [Windmill 文档]() * [脚本中心]() * [GitHub 存储库]() * [Discord 社区]() * [API 参考]() ** ** _此技能为 Windmill 工作流程自动化提供了生产就绪模式，并使用 Python、TypeScript、Go 和 Bash 脚本在企业场景中进行了测试。_

风车

包含技能