菜单
Wires Forze durable function contracts with DurableFunctionEventSpec, DurableFunctionSpec, InngestDepsModule, event emit, function registration, DurableFunctionStepDepKey, and FastAPI serve. Use for event-driven durable handlers with Inngest in applications using forze[inngest].
用 Codex 或 Claude 帮你安装 复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。
基于 SOC 职业分类
Uses Forze queue, pub/sub, and stream contracts with QueueSpec, PubSubSpec, StreamSpec, SQSDepsModule, RabbitMQDepsModule, Redis adapters, and Mock adapters. Use when producing, consuming, or testing async messages/events.
Implements Forze document and search access the kit-first way: DocumentFacade and SearchFacade over build_document_registry / build_search_registry, kit DTOs (DocumentIdDTO, ListRequestDTO, DocumentUpdateDTO), the query DSL, pagination, cache-aware DocumentSpec / SearchSpec, and the Postgres / Mongo / Firestore / Meilisearch backends behind them. Use when building data access features.
Optimize code only after a real, measured performance problem - distinguish macro (design-level) from micro (fine-tuned) performance, reach for data-structure and algorithm wins before micro-tweaks, profile to find real hotspots, and don't trade readability or adaptability for unmeasured speed. Use when optimizing code, reviewing performance suggestions, choosing "faster" constructs, or when the user mentions performance, optimization, speed, or efficiency.
Name variables, functions, classes, and modules well by avoiding known anti-patterns - single letters, abbreviations, types baked into names, missing units, "Base"/"Abstract" class names, and "Utils"/"Helper" grab-bags. Use when naming or renaming code, reviewing names in a diff, or when the user mentions naming, identifiers, variable names, readability, or that a name feels off.
Keep code flat and readable by limiting indentation depth, using guard clauses with early returns, and extracting nested blocks into well-named functions. Use when writing or refactoring code with deep nesting, pyramid-shaped if/else, arrow code, or when the user mentions nesting, indentation, guard clauses, early returns, extracting functions, cyclomatic complexity, or making code more readable.
Make code explain itself instead of relying on comments - name sub-expressions, extract complex conditions into well-named functions, lean on the type system, and reserve comments for the "why" the code cannot express. Use when writing or reviewing comments, refactoring hard-to-read logic, deciding whether a comment is needed, or when the user mentions comments, self-documenting code, or code readability.
| name | forze-inngest-durable-functions |
| description | Wires Forze durable function contracts with DurableFunctionEventSpec, DurableFunctionSpec, InngestDepsModule, event emit, function registration, DurableFunctionStepDepKey, and FastAPI serve. Use for event-driven durable handlers with Inngest in applications using forze[inngest]. |
Use when your application emits events from HTTP handlers and runs memoized steps in Inngest workers. Install forze[inngest] (and forze[inngest,fastapi] for serve). Contracts live in forze.application.contracts.durable.function; wire them with forze_inngest.
from forze.application.contracts.durable.function import DurableFunctionEventSpec
from forze.base.serialization import PydanticModelCodec
invoice_paid = DurableFunctionEventSpec(
name="app/invoice.paid",
codec=PydanticModelCodec(model_type=InvoicePaidPayload),
)
from forze.application.contracts.durable.function import (
DurableFunctionCronTrigger,
DurableFunctionEventTrigger,
DurableFunctionInvokeSpec,
DurableFunctionSpec,
)
on_invoice_paid = DurableFunctionSpec(
name="on-invoice-paid",
run=DurableFunctionInvokeSpec(args_type=InvoicePaidPayload, return_type=Result),
triggers=(DurableFunctionEventTrigger(event="app/invoice.paid"),),
)
from forze.application.execution import DepsRegistry, LifecyclePlan
from forze_inngest import InngestClient, InngestDepsModule, inngest_lifecycle_step
client = InngestClient(app_id="my-app")
module = InngestDepsModule(
client=client,
events={invoice_paid.name: {}},
)
deps = DepsRegistry.from_modules(module)
lifecycle = LifecyclePlan.from_steps(inngest_lifecycle_step())
There is no ctx.durable_function_event(...). Use resolve_configurable with
route=spec.name:
port = ctx.deps.resolve_configurable(
ctx,
DurableFunctionEventCommandDepKey,
invoice_paid,
route=invoice_paid.name,
)
await port.send(InvoicePaidPayload(invoice_id="inv-1"))
Canonical (frozen registry operation): set operation on the spec and bind the
same frozen_registry as HTTP routes.
from forze_inngest import InngestFunctionBinding, register_functions
from forze_inngest.fastapi import serve
scan_spec = DurableFunctionSpec(
name="scan-inbox",
operation="jobs.scan_inbox",
run=DurableFunctionInvokeSpec(args_type=CronTickArgs, return_type=None),
triggers=(DurableFunctionCronTrigger(expression="0 */3 * * *"),),
)
binding = InngestFunctionBinding.for_registry_operation(scan_spec, frozen_registry)
register_functions(client, [binding], ctx_factory=make_execution_context)
serve(app, client, [binding], ctx_factory=make_execution_context)
Escape hatch: handler_factory when operation is unset (custom handler, not a
registry op).
binding = InngestFunctionBinding(
spec=on_invoice_paid,
handler_factory=lambda ctx: OnInvoicePaidHandler(deps=ctx.deps),
)
Inside the function handler only:
step = ctx.deps.provide(DurableFunctionStepDepKey)
await step.run("notify", lambda: notifier.send(...))
DurableWorkflow* types or forze_temporal in Inngest-based apps for the same work — pick Temporal or Inngest per use case.DurableFunctionStepDepKey from ordinary HTTP handlers.start() command port — Inngest runs are event- or cron-driven.Docs are versioned. These links use
latest(the newest release). If your app pins an olderforzeminor, replacelatestin the URL with that version (e.g..../forze/0.3/...) or use the version selector on the site.