@ canonical/ 任务

用于可相容,可测试,可干燥的CLI操作的一元效应框架.

任务是对计算过程的纯描述——在解释之前不会执行. 这使得您可以测试文件系统操作, shell 命令, 和用户提示而不触摸真实的 I/O.

安装

bun add @canonical/task

入境点

货包装了两个入口,按它们能达到的程度分成:

• @canonical/task ——无节点基础:核心类型,任务僧尼,效果构建器,原始人,梳理人,干跑/测试翻译(dryRun, (中文(简体) ), dryRunWith, (中文(简体) ), collectEffects,...),撤销收藏(collectUndos),以及 TaskExecutionError. 所构建的节点可以从本条目的导入图(由关闭行走测试所支撑)得到,因此它为浏览器和边缘目标打包干净. 所有关于 说明, (中文(简体) ), 编译,以及 预览 工作住在这里.
• @canonical/task/node - 节点触控口译员: runTask 编号 : run 编号 : executeEffect (执行文件系统、执行和即时效果,导入) node:fs/promises, (中文(简体) ), node:path,以及 node:child_process和 runUndo。只是实际上 运行 针对主机的任务需要此条目.

因为模块解析发生在树摇之前, 拆分事项是: 如果口译员住在基项中, 导入一个纯函数就会拉动 node: 导入到模块图中并失败任何浏览器-平台的构建——无论捆绑实际使用什么.

快速启动

import { gen, $, writeFile, readFile, info } from "@canonical/task";
import { runTask } from "@canonical/task/node";

const setup = gen(function* () {
  const name = yield* $(readFile("name.txt"));
  yield* $(writeFile("greeting.txt", `Hello, ${name}!`));
  yield* $(info("Done"));
});

await runTask(setup);

核心概念

任务修饰

A 类 Task<A> 有三样东西:

• 纯 - 完成的计算持有一个数值
• 效果 - 附带效应说明
• 失败 ——计算失败,结构错误

任务由 flatMap 或者,在工学上,使用生成语法:

import { flatMap, readFile, writeFile } from "@canonical/task";

// flatMap style
const greet = flatMap(
  readFile("name.txt"),
  (name) => writeFile("greeting.txt", `Hello, ${name}!`),
);

生成语法

这个 gen 编号 : $ 对子允许您不嵌入而写入顺序效果代码 flatMap 锁链。 使用 yield* 与 $(task) 打开任务并获得其值 :

import { gen, $, readFile, writeFile, info, exists } from "@canonical/task";

const migrate = gen(function* () {
  const raw = yield* $(readFile("config.json"));
  const config = JSON.parse(raw);

  config.version = 2;

  yield* $(writeFile("config.json", JSON.stringify(config, null, 2)));
  yield* $(info(`Migrated to v${config.version}`));

  return config;
});

在引擎盖下, gen 编曲 flatMap 调用——任务在解释前仍然是纯数据结构. 使用您喜欢的样式; gen 建议任何超过两到三个步骤.

效果

效果是纯数据——贴有标签的工会描述应该发生什么:

初级

任务返回包装符,用于每种效果。 这些是你最常用的构件:

import {
  readFile, writeFile, appendFile, copyFile, copyDirectory,
  deleteFile, deleteDirectory, mkdir, exists, symlink, glob,
  exec, execSimple,
  prompt, promptText, promptConfirm, promptSelect, promptMultiselect,
  log, debug, info, warn, error,
  getContext, setContext, withContext,
  noop, succeed,
} from "@canonical/task";

文件系统

const content = yield* $(readFile("src/index.ts"));
yield* $(writeFile("dist/index.js", compiled));
yield* $(appendFile("log.txt", `Built at ${Date.now()}\n`));
yield* $(copyFile("template.json", "output/package.json"));
yield* $(copyDirectory("templates/", "output/"));
yield* $(mkdir("output/lib"));
yield* $(symlink("../shared/utils", "src/utils"));

const found = yield* $(exists("tsconfig.json"));
const files = yield* $(glob("src/**/*.ts", "."));

进程执行

const result = yield* $(exec("git", ["status", "--short"]));
// result: { stdout: string, stderr: string, exitCode: number }

const simple = yield* $(execSimple("ls -la"));

用户提示

const name = yield* $(promptText("name", "Project name?", "my-app"));
const ok = yield* $(promptConfirm("confirm", "Continue?", true));
const lang = yield* $(promptSelect("lang", "Language?", [
  { label: "TypeScript", value: "ts" },
  { label: "JavaScript", value: "js" },
]));
const features = yield* $(promptMultiselect("features", "Features?", [
  { label: "Linting", value: "lint" },
  { label: "Testing", value: "test" },
]));

日志

yield* $(debug("Verbose detail"));   // shown with --verbose
yield* $(info("Progress update"));
yield* $(warn("Non-fatal issue"));
yield* $(error("Something broke"));

A. 背景情况

上下文是一个密钥-价值商店,在任务执行期间存在. 使用它来在步骤之间传递数据, 而不需要线程值 :

import { gen, $, setContext, getContext, withContext } from "@canonical/task";

const setup = gen(function* () {
  yield* $(setContext("projectName", "my-app"));

  // later, in a different part of the task tree
  const name = yield* $(getContext<string>("projectName"));
  // name: "my-app"
});

// withContext sets a key for the duration of a child task
const scoped = withContext("env", "production", deployTask);

上下文辅以 Map<string, unknown> 在制作翻译。 默认情况下干运行的解析器不会持续上下文 – 使用 dryRunWith 以提供模拟上下文值.

错误模型

结构化任务失败 TaskError 值 :

interface TaskError {
  code: string;              // programmatic error code
  message: string;           // human-readable description
  cause?: unknown;           // original error that caused this failure
  context?: Record<string, unknown>;  // additional structured data
  stack?: string;            // stack trace if available
  suppressed?: TaskError[];  // for parallel: all errors, not just the first
}

创建错误

import { fail, failWith } from "@canonical/task";

const notFound = failWith("FILE_NOT_FOUND", "Config file missing");

const detailed = fail({
  code: "VALIDATION_FAILED",
  message: "Schema mismatch",
  context: { path: "config.json", expected: "v2" },
});

框架定义了基础错误代码( :FILE_NOT_FOUND, (中文(简体) ), EXEC_FAILED, (中文(简体) ), PROMPT_CANCELLED, (中文(简体) ), TASK_INTERRUPTED, (中文(简体) ), INTERNAL;消费者可以使用任意字符串代码.

处理错误

import { recover, mapError, orElse, optional, attempt, fold } from "@canonical/task";

// recover: catch a failure and produce a new task
const safe = recover(riskyTask, (err) =>
  err.code === "FILE_NOT_FOUND" ? writeFile("config.json", "{}") : fail(err),
);

// mapError: transform the error without changing recovery
const retagged = mapError(innerTask, (err) => ({
  ...err,
  code: "DEPLOY_FAILED",
  context: { ...err.context, phase: "build" },
}));

// orElse: try A, fall back to B
const config = orElse(readFile("config.local.json"), readFile("config.json"));

// optional: swallow failure, return undefined
const maybePkg = optional(readFile("package.json"));

// attempt: capture success or failure as a value
const result = yield* $(attempt(riskyTask));
if (result.ok) { /* result.value */ } else { /* result.error */ }

// fold: handle both branches and unify the type
const status = fold(deployTask, () => "deployed", (err) => `failed: ${err.code}`);

并行错误

何时 parallel 任务失败, 第一个错误变为主错误 TaskError 和任何剩余错误作为附件附上 suppressed数字 :

// If tasks[0] and tasks[2] both fail:
// error.code      → first failure's code
// error.suppressed → [task[2]'s error]

组合器

将任务编成更大的工作流程:

顺序

import { sequence, sequence_, traverse, traverse_ } from "@canonical/task";

// Run in order, collect results
const contents = sequence([readFile("a.txt"), readFile("b.txt")]);

// Run in order, discard results
sequence_([writeFile("a.txt", "A"), writeFile("b.txt", "B")]);

// Map + sequence over an array
const compiled = traverse(sourceFiles, (f) => readFile(f));

// Map + sequence, discard results
traverse_(files, (f) => deleteFile(f));

并行执行

import { parallel, parallelN, race } from "@canonical/task";

// Run concurrently, collect all results
const all = parallel([fetchA, fetchB, fetchC]);

// Run with concurrency limit (batches of 3)
const throttled = parallelN(3, manyTasks);

// Return first to complete
const fastest = race([mirrorA, mirrorB]);

条件

import { when, unless, ifElse, whenM, ifElseM } from "@canonical/task";

// Run only if condition is true
when(isDirty, cleanUp);

// Run only if condition is false
unless(isCI, promptForConfirmation);

// Choose between two tasks
ifElse(hasConfig, loadConfig, useDefaults);

// Condition is itself a task
whenM(exists("package.json"), installDeps);

// Both condition and branches are tasks
ifElseM(exists(".env"), loadEnv, createEnv);

调度

import { switchMap } from "@canonical/task";

// Detect-then-dispatch: run a detection task, branch on its result
const result = switchMap(
  detectHarness,                // Task<"jest" | "vitest" | null>
  {
    jest: configureJest,        // Task<Config>
    vitest: configureVitest,    // Task<Config>
  },
  useDefaultConfig,             // fallback Task<Config>
);

处理错误

import { retry, orElse, optional, attempt, bracket, ensure } from "@canonical/task";

// Retry up to 3 times
const resilient = retry(flakyTask, 3);

// Try primary, fall back to secondary
const config = orElse(readFile("local.json"), readFile("default.json"));

// Swallow errors, return undefined
const maybe = optional(readFile("optional.json"));

// Capture result or error as a value
const result = attempt(riskyTask);

// Resource management (acquire → use → release, even on failure)
const safe = bracket(acquireConn, useConn, releaseConn);

// Ensure cleanup runs regardless of outcome
const withCleanup = ensure(mainTask, cleanup);

公用事业

import { tap, tapError, fold, zip, zip3 } from "@canonical/task";

// Side-effect without changing the value
const logged = tap(readFile("a.txt"), (content) => info(`Read ${content.length} bytes`));

// Side-effect on failure
const observed = tapError(riskyTask, (err) => error(`Failed: ${err.code}`));

// Handle both success and failure → unified type
const status = fold(deploy, () => "ok", (err) => `fail: ${err.code}`);

// Combine tasks into tuples
const [a, b] = yield* $(zip(readFile("a.txt"), readFile("b.txt")));
const [x, y, z] = yield* $(zip3(taskX, taskY, taskZ));

流利构建器

这个 task() 编号 : of() API 提供了一个可连锁的选项 :

import { task, of, mkdir, writeFile, info } from "@canonical/task";

const result = task(mkdir("output"))
  .andThen(writeFile("output/a.txt", "A"))
  .andThen(writeFile("output/b.txt", "B"))
  .andThen(info("Done!"))
  .unwrap();

const doubled = of(21)
  .map((n) => n * 2)
  .flatMap((n) => writeFile("answer.txt", String(n)))
  .unwrap();

打电话 .unwrap() 以提取基础 Task<A> 当你需要传递给梳理员或口译员时.

口译员

任务是惰性数据,直到一个解释者走过结构并决定如何处理每个效果. 包裹运送了三名口译员:

制作口译runTask(中文(简体) )

执行对真 I/O 的效果——文件系统,进程,提示 :

import { runTask } from "@canonical/task/node";

const value = await runTask(myTask);

runTask 接受一个 RunTaskOptions 对象 :

await runTask(myTask, {
  // Shared key-value context for ReadContext/WriteContext effects
  context: new Map([["env", "production"]]),

  // Custom prompt handler (required if the task uses Prompt effects)
  promptHandler: async (effect) => {
    // effect.question: PromptQuestion
    return "user input";
  },

  // Log routing (default: console.log with level prefix)
  onLog: (level, message) => logger[level](message),

  // Effect lifecycle hooks
  onEffectStart: (effect) => { /* before each effect */ },
  onEffectComplete: (effect, durationMs) => { /* after each effect */ },

  // AbortSignal for interruption
  signal: controller.signal,
});

当任务失败时, runTask 丢出一个 TaskExecutionError 包起来 TaskError数字 :

import { TaskExecutionError } from "@canonical/task";
import { runTask } from "@canonical/task/node";

try {
  await runTask(myTask);
} catch (err) {
  if (err instanceof TaskExecutionError) {
    console.log(err.code);          // error code string
    console.log(err.taskError);     // full TaskError object
  }
}

干线传译员(dryRun(中文(简体) )

收集效果而不执行它们。 每个效果都有一个模拟返回值,这样任务就可以继续:

import { dryRun } from "@canonical/task";

const { value, effects } = dryRun(myTask);
// value: the task's return value (using mocked effect results)
// effects: Effect[] — every effect the task would have performed

默认模拟 : ReadFile → "[mock content of <path>]", (中文(简体) ), Exists → true, (中文(简体) ), Exec → { stdout: "", stderr: "", exitCode: 0 }, (中文(简体) ), Prompt 缺省或首选,写效果 undefined。 。 。 .

干运行的口译员跟踪虚拟文件系统——由 WriteFile 或者说 MakeDir 效果可见于后续 Exists 在同一跑道内检查.

自定义模拟 dryRunWith

覆盖每个效果类型的模拟行为 :

import { dryRunWith } from "@canonical/task";

const mocks = new Map([
  ["ReadFile", (effect) => {
    if (effect.path === "package.json") return '{"name": "my-app"}';
    return "default content";
  }],
  ["Exists", () => false],
]);

const { value, effects } = dryRunWith(myTask, mocks);

效应分析公用事业

import {
  collectEffects, countEffects, filterEffects,
  getFileWrites, getAffectedFiles,
} from "@canonical/task";

const effects = collectEffects(myTask);

countEffects(effects);
// { WriteFile: 3, ReadFile: 1, Log: 2 }

filterEffects(effects, "WriteFile");
// [{ _tag: "WriteFile", path: "...", content: "..." }, ...]

getFileWrites(effects);
// [{ path: "a.txt", content: "A" }, ...]

getAffectedFiles(effects);
// ["a.txt", "b.txt", "output/"] — sorted, deduplicated

撤消口译( E)runUndo(中文(简体) )

以被嘲弄的前向效果行走任务树( 如 dryRun),收集 undo 任务,然后按倒转顺序执行。 这样可以 --undo 在任何没有存储状态的 CLI 命令上—— 相同的任务定义加上相同的答案产生决定性的撤消.

import { collectUndos } from "@canonical/task";
import { runUndo } from "@canonical/task/node";

// Forward run:
await runTask(generator.generate(answers));

// Undo (later, same answers):
await runUndo(generator.generate(answers));

如何撤销元数据

可写效果带有可选 undo?: Task<void> 字段。 初级设备提供合理的默认:

自定义取消

覆盖默认撤消或提供不存在默认的默认:

// Custom undo for append
appendFile(indexPath, exportLine, true, {
  undo: removeLineFromFile(indexPath, exportLine),
});

// Disable default undo
writeFile(tempPath, content, { undo: null });

不执行而收集撤消

const undos = collectUndos(myTask);
// undos: Task<void>[] — undo tasks in forward execution order
// (runUndo reverses them automatically)

可编译性

撤消所有现有的组合符自动作曲 :

• when(false, writeFile(...)) - 跳过的效果不会产生倒数
• sequence_([...]) ——撤销顺序收集,倒转执行
• parallel([...]) - 从所有儿童那里收集的废品
• gen(function* () { ... }) - 工作相同

测试断言

import { assertEffects, assertFileWrites, expectTask } from "@canonical/task";

// Assert exact effect sequence
assertEffects(myTask, [
  { _tag: "MakeDir", path: "output" },
  { _tag: "WriteFile", path: "output/index.ts" },
]);

// Assert which files would be written
assertFileWrites(myTask, ["output/index.ts", "output/package.json"]);

// Fluent matcher
const result = expectTask(myTask);
result.toHaveValue("done");
result.toHaveEffectCount(3);
result.toWriteFile("output/index.ts");
result.toNotWriteFile("output/secret.key");

API 总结

许可证

LGPL-3.0 (英语)