### Run Hatchet Quickstart Source: https://docs.hatchet.run/v1/quickstart Initializes the quickstart project environment. ```sh hatchet quickstart ``` -------------------------------- ### Ruby Dockerfile Example Source: https://docs.hatchet.run/v1/docker Dockerfile for Ruby applications, including system dependency installation for native gems and bundle configuration. ```dockerfile FROM ruby:3.3-slim ENV HATCHET_ENV=production # Install system dependencies for native gems RUN apt-get update && \ apt-get install -y build-essential && \ apt-get clean && \ rm -rf /var/lib/apt/lists/* WORKDIR /app COPY Gemfile Gemfile.lock ./ RUN bundle config set --local without 'development test' && \ bundle install COPY . /app CMD ["bundle", "exec", "ruby", "worker.rb"] ``` -------------------------------- ### Common Environment Variable Examples Source: https://docs.hatchet.run/v1/compute/environment-variables Examples of standard environment variable definitions for application configuration. ```text # Examples of commonly used variables NODE_ENV=production LOG_LEVEL=info API_TIMEOUT=30000 DATABASE_URL=postgresql://user:pass@host:5432/db ``` -------------------------------- ### Install Hatchet SDKs Source: https://docs.hatchet.run/v1/from-temporal-to-hatchet Commands to install the Hatchet SDK for various programming languages. ```bash pip install hatchet-sdk ``` ```bash npm install @hatchet-dev/typescript-sdk ``` ```bash go get github.com/hatchet-dev/hatchet/sdks/go ``` ```bash gem install hatchet-sdk ``` -------------------------------- ### Install Python OpenTelemetry extra Source: https://docs.hatchet.run/v1/opentelemetry Install the necessary dependencies for OpenTelemetry support in the Python SDK. ```bash pip install hatchet-sdk[otel] ``` -------------------------------- ### Optimize Docker library installation Source: https://docs.hatchet.run/v1/compute/gpu Avoid meta-packages to keep image sizes small and install only necessary CUDA libraries. ```dockerfile # DO NOT use meta-packages # ❌ RUN apt-get install cuda-runtime-* # ✅ Install only required libraries RUN apt-get install -y \ cuda-nvcc-12-2 \ libcublas-12-2 \ libcudnn8 ``` -------------------------------- ### Install Hatchet CLI Skills Source: https://docs.hatchet.run/v1/using-coding-agents Installs reference documentation for AI agents to understand Hatchet CLI operations. ```bash hatchet skills install ``` ```bash hatchet skills install --dir ./my-project ``` -------------------------------- ### Example Dockerfile Directory Structure Source: https://docs.hatchet.run/v1/compute/git-ops Demonstrates a project structure with multiple Dockerfiles for different worker configurations. ```text . ├── Dockerfile.worker1 # For worker pool 1 ├── Dockerfile.worker2 # For worker pool 2 └── services/ └── special/ └── Dockerfile # For specialized worker ``` -------------------------------- ### Install Hatchet CLI via Homebrew Source: https://docs.hatchet.run/v1/quickstart Installation method for MacOS users using Homebrew. ```sh brew install hatchet-dev/hatchet/hatchet --cask ``` -------------------------------- ### Declare and start a worker Source: https://docs.hatchet.run/v1/workers Initialize a worker with a unique name and a list of workflows, then start the worker process. ```python def main() -> None: worker = hatchet.worker("dag-worker", workflows=[dag_workflow]) worker.start() ``` ```typescript import { hatchet } from '../hatchet-client'; import { simple } from './workflow'; import { parent, child } from './workflow-with-child'; import { simpleWithZod } from './zod'; async function main() { const worker = await hatchet.worker('simple-worker', { // 👀 Declare the workflows that the worker can execute workflows: [simple, simpleWithZod, parent, child], // 👀 Declare the number of concurrent task runs the worker can accept slots: 100, }); await worker.start(); } if (require.main === module) { main(); } ``` ```go worker, err := client.NewWorker("simple-worker", hatchet.WithWorkflows(task)) if err != nil { log.Fatalf("failed to create worker: %v", err) } interruptCtx, cancel := cmdutils.NewInterruptContext() defer cancel() err = worker.StartBlocking(interruptCtx) if err != nil { log.Fatalf("failed to start worker: %v", err) } ``` ```ruby def main worker = HATCHET.worker("dag-worker", workflows: [DAG_WORKFLOW]) worker.start end ``` -------------------------------- ### Install Hatchet CLI via Native Script Source: https://docs.hatchet.run/v1/quickstart Recommended installation method for MacOS, Linux, and WSL environments. ```sh curl -fsSL https://install.hatchet.run/install.sh | bash ``` -------------------------------- ### Install TypeScript OpenTelemetry packages Source: https://docs.hatchet.run/v1/opentelemetry Install the required OpenTelemetry core and exporter packages for the TypeScript SDK. ```bash npm install @opentelemetry/api @opentelemetry/instrumentation @opentelemetry/sdk-trace-base @opentelemetry/exporter-trace-otlp-grpc ``` -------------------------------- ### Install Hatchet SDK Source: https://docs.hatchet.run/v1/from-celery-to-hatchet Install the Hatchet SDK using pip to begin the migration process alongside existing Celery dependencies. ```bash pip install hatchet-sdk ``` -------------------------------- ### Example Repository Structure Source: https://docs.hatchet.run/v1/compute/git-ops A recommended directory layout for managing multiple worker configurations within a single repository. ```text . ├── Dockerfile.worker1 # Primary worker configuration ├── Dockerfile.worker2 # Secondary worker configuration ├── .dockerignore ├── src/ │ ├── worker1/ # Worker 1 specific code │ │ └── worker.py │ └── worker2/ # Worker 2 specific code │ └── worker.py └── hatchet.yaml # Optional Hatchet configuration ``` -------------------------------- ### Go Dockerfile Example Source: https://docs.hatchet.run/v1/docker Multi-stage build for Go applications, compiling the binary in the first stage and running it in a minimal production image. ```dockerfile # Stage 1: Build FROM golang:1.26-alpine3.21 AS builder WORKDIR /app COPY . . RUN go mod download RUN go build -o hatchet-worker . # Stage 2: Production FROM golang:1.26-alpine3.21 WORKDIR /app COPY --from=builder hatchet-worker . CMD ["/app/hatchet-worker"] ``` -------------------------------- ### Worker connection logs Source: https://docs.hatchet.run/v1/workers Example output confirming the worker has successfully connected to the engine. ```text [INFO] 🪓 -- starting hatchet... [DEBUG] 🪓 -- 'test-worker' waiting for ['simpletask:step1'] [DEBUG] 🪓 -- acquired action listener: efc4aaf2-... [DEBUG] 🪓 -- sending heartbeat ``` -------------------------------- ### Initialize V1 Reflection SDK Worker Source: https://docs.hatchet.run/v1/migrating/migration-guide-go Demonstrates setting up a client, defining input/output types, and starting a blocking worker using the current reflection-based SDK. ```go package main import ( "context" "log" hatchet "github.com/hatchet-dev/hatchet/sdks/go" ) func main() { client, err := hatchet.NewClient() if err != nil { log.Fatal(err) } // Define input/output types type Input struct { Message string `json:"message"` } type Output struct { Result string `json:"result"` } // Create a simple task task := client.NewStandaloneTask("simple-task", func(ctx hatchet.Context, input Input) (Output, error) { return Output{Result: "Processed: " + input.Message}, nil }) // Start worker worker, err := client.NewWorker("worker", hatchet.WithWorkflows(task)) if err != nil { log.Fatal(err) } if err := worker.StartBlocking(context.Background()); err != nil { log.Fatal(err) } } ``` -------------------------------- ### JavaScript Dockerfile Examples Source: https://docs.hatchet.run/v1/docker Multi-stage Docker builds for Node.js applications using npm, pnpm, or yarn package managers. ```dockerfile # Stage 1: Build FROM node:18 AS builder WORKDIR /app COPY package*.json ./ RUN npm ci COPY . . RUN npm run build # Stage 2: Production FROM node:22-alpine WORKDIR /app COPY package*.json ./ RUN npm ci --omit=dev COPY --from=builder /app/dist ./dist ENV NODE_ENV=production CMD ["node", "dist/worker.js"] ``` ```dockerfile # Stage 1: Build FROM node:18 AS builder WORKDIR /app # Install pnpm RUN npm install -g pnpm COPY pnpm-lock.yaml package.json ./ RUN pnpm install --frozen-lockfile COPY . . RUN pnpm build # Stage 2: Production FROM node:22-alpine WORKDIR /app RUN npm install -g pnpm COPY pnpm-lock.yaml package.json ./ RUN pnpm install --frozen-lockfile --prod COPY --from=builder /app/dist ./dist ENV NODE_ENV=production CMD ["node", "dist/worker.js"] ``` ```dockerfile # Stage 1: Build FROM node:18 AS builder WORKDIR /app COPY package.json yarn.lock ./ RUN yarn install --frozen-lockfile COPY . . RUN yarn build # Stage 2: Production FROM node:22-alpine WORKDIR /app COPY package.json yarn.lock ./ RUN yarn install --frozen-lockfile --production COPY --from=builder /app/dist ./dist ENV NODE_ENV=production CMD ["node", "dist/worker.js"] ``` -------------------------------- ### Run worker with TypeScript Source: https://docs.hatchet.run/v1/workers Define a start script in package.json and execute it via npm. ```json "scripts": { "start:worker": "ts-node src/worker.ts" } ``` ```bash npm run start:worker ``` -------------------------------- ### Install GPU library dependencies Source: https://docs.hatchet.run/v1/compute/gpu Install common CUDA libraries and specific NVIDIA drivers required for GPU operations. ```dockerfile # Install commonly needed libraries RUN apt-get install -y \ libcublas-12-2 \ libcudnn8 \ nvidia-driver-525 ``` -------------------------------- ### Python Dockerfile Examples Source: https://docs.hatchet.run/v1/docker Docker configurations for Python applications using either Poetry or pip for dependency management. ```dockerfile FROM python:3.13-slim ENV PYTHONUNBUFFERED=1 \ POETRY_VERSION=1.4.2 \ HATCHET_ENV=production # Install system dependencies and Poetry RUN apt-get update && \ apt-get install -y curl && \ curl -sSL https://install.python-poetry.org | python3 - && \ ln -s /root/.local/bin/poetry /usr/local/bin/poetry && \ apt-get clean && \ rm -rf /var/lib/apt/lists/* WORKDIR /app COPY pyproject.toml poetry.lock* /app/ RUN poetry config virtualenvs.create false && \ poetry install --no-interaction --no-ansi COPY . /app CMD ["poetry", "run", "python", "worker.py"] ``` ```dockerfile FROM python:3.13-slim ENV PYTHONUNBUFFERED=1 \ HATCHET_ENV=production WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . /app CMD ["python", "worker.py"] ``` -------------------------------- ### Implement a multi-step onboarding flow with sleeps Source: https://docs.hatchet.run/v1/from-temporal-to-hatchet Examples of workflows that include waiting periods, requiring execution timeouts to cover the total duration of the task. ```python @hatchet.durable_task( name="OnboardingFlow", input_validator=OnboardingInput, # The timeout has to cover the whole wall-clock span of the run, sleeps included. execution_timeout=timedelta(days=7), ) async def onboarding_flow(input: OnboardingInput, ctx: DurableContext) -> None: await send_welcome_email.aio_run(input) await ctx.aio_sleep_for(timedelta(days=3)) await send_followup_email.aio_run(input) ``` ```typescript export const onboardingFlow = hatchet.durableTask({ name: 'OnboardingFlow', // The timeout has to cover the whole wall-clock span of the run, sleeps included. executionTimeout: '168h', fn: async (input: SignupInput, ctx): Promise => { await sendWelcomeEmail.run(input); await ctx.sleepFor('72h'); return sendFollowupEmail.run(input); }, }); ``` ```go onboardingFlow := client.NewStandaloneDurableTask("OnboardingFlow", func(ctx hatchet.DurableContext, input OnboardingInput) (OnboardingOutput, error) { if _, err := sendWelcomeEmail.Run(ctx, input); err != nil { return OnboardingOutput{}, fmt.Errorf("sending welcome email: %w", err) } if _, err := ctx.SleepFor(72 * time.Hour); err != nil { return OnboardingOutput{}, fmt.Errorf("sleeping between emails: %w", err) } if _, err := sendFollowupEmail.Run(ctx, input); err != nil { return OnboardingOutput{}, fmt.Errorf("sending follow-up email: %w", err) } return OnboardingOutput{EmailsSent: 2}, nil }, // The execution timeout has to cover the sleep, not just the work. hatchet.WithExecutionTimeout(168*time.Hour), ) ``` ```ruby # The execution timeout has to cover the whole wall-clock span of the run, # sleeps included, so a three-day sleep needs a timeout longer than three days. ONBOARDING_FLOW = HATCHET.durable_task( name: "OnboardingFlow", execution_timeout: 604_800 ) do |input, ctx| welcome = SEND_WELCOME_EMAIL.run(input) ctx.sleep_for(duration: 259_200) followup = SEND_FOLLOWUP_EMAIL.run(input) { "user_id" => input["user_id"], "welcome_message_id" => welcome["message_id"], "followup_message_id" => followup["message_id"] } end ``` -------------------------------- ### Define a Temporal Activity Source: https://docs.hatchet.run/v1/from-temporal-to-hatchet Examples of defining activities in Temporal across different SDKs. ```python from temporalio import activity @activity.defn async def charge_order(order_id: str) -> bool: return await payments.charge(order_id) ``` ```typescript // activities.ts — activities are plain exported functions. export async function chargeOrder(orderId: string): Promise { return payments.charge(orderId); } ``` ```go import "context" func ChargeOrder(ctx context.Context, orderID string) (bool, error) { return payments.Charge(ctx, orderID) } ``` ```ruby require "temporalio/activity" class ChargeOrder < Temporalio::Activity::Definition def execute(order_id) Payments.charge(order_id) end end ``` -------------------------------- ### Configure Cancel Newest Strategy Source: https://docs.hatchet.run/v1/concurrency Examples of implementing the CANCEL_NEWEST concurrency strategy across different supported languages. ```python class WorkflowInput(BaseModel): group: str concurrency_cancel_newest_workflow = hatchet.workflow( name="ConcurrencyCancelNewest", concurrency=ConcurrencyExpression( expression="input.group", max_runs=1, limit_strategy=ConcurrencyLimitStrategy.CANCEL_NEWEST, ), input_validator=WorkflowInput, ) ``` ```typescript export const concurrencyCancelNewestWorkflow = hatchet.workflow({ name: 'concurrencycancelnewest', concurrency: { expression: 'input.group', maxRuns: 1, limitStrategy: ConcurrencyLimitStrategy.CANCEL_NEWEST, }, }); ``` ```go var maxRuns int32 = 1 strategy := types.CancelNewest return client.NewStandaloneTask("cancel-newest", func(ctx worker.HatchetContext, input ConcurrencyInput) (*TransformedOutput, error) { // Random sleep between 200ms and 1000ms time.Sleep(time.Duration(200+rand.Intn(800)) * time.Millisecond) return &TransformedOutput{ TransformedMessage: input.Message, }, nil }, hatchet.WithWorkflowConcurrency(types.Concurrency{ Expression: "input.GroupKey", MaxRuns: &maxRuns, LimitStrategy: &strategy, }), ) ``` ```ruby CONCURRENCY_CANCEL_NEWEST_WORKFLOW = HATCHET.workflow( name: "ConcurrencyCancelNewest", concurrency: Hatchet::ConcurrencyExpression.new( expression: "input.group", max_runs: 1, limit_strategy: :cancel_newest ) ) ``` -------------------------------- ### Prometheus configuration for metrics Source: https://docs.hatchet.run/v1/worker-healthchecks Example configuration for scraping Hatchet worker metrics. ```yaml scrape_configs: - job_name: "hatchet" scrape_interval: 5s static_configs: - targets: ["localhost:8001"] ``` -------------------------------- ### Start a Celery worker Source: https://docs.hatchet.run/v1/from-celery-to-hatchet Standard CLI command for launching a Celery worker with specified concurrency. ```bash celery -A tasks worker --loglevel=info --concurrency=4 ``` -------------------------------- ### Implement a Go Standalone Task Source: https://docs.hatchet.run/v1/inter-service-triggering Example of defining and running a standalone task in Go, which can be triggered by other services sharing the same name and schema. ```go package main import ( "context" "fmt" "log" hatchet "github.com/hatchet-dev/hatchet/sdks/go" ) type StubInput struct { Message string `json:"message"` } type StubOutput struct { Ok bool `json:"ok"` } func StubWorkflow(client *hatchet.Client) *hatchet.StandaloneTask { return client.NewStandaloneTask("stub-workflow", func(ctx hatchet.Context, input StubInput) (StubOutput, error) { return StubOutput{ Ok: true, }, nil }) } func main() { client, err := hatchet.NewClient() if err != nil { log.Fatalf("failed to create hatchet client: %v", err) } task := StubWorkflow(client) // we are simply running the task here, but it can be implemented in another service / worker // and in another language with the same name and input-output types result, err := task.Run(context.Background(), StubInput{Message: "Hello, World!"}) if err != nil { log.Fatalf("failed to run task: %v", err) } fmt.Println(result) } ``` -------------------------------- ### Fetch Metrics via cURL Source: https://docs.hatchet.run/v1/prometheus-metrics Example command to retrieve metrics using a Bearer token. ```bash curl -H "Authorization: Bearer your-api-token-here" \ https://cloud.onhatchet.run/api/v1/tenants/707d0855-80ab-4e1f-a156-f1c4546cbf52/prometheus-metrics ``` -------------------------------- ### Install Hatchet Docs MCP and Agent Skills Source: https://docs.hatchet.run/v1/quickstart Commands to integrate Hatchet documentation and CLI skills into AI coding assistants. ```sh hatchet docs install ``` ```sh hatchet skills install ``` -------------------------------- ### Configure Task Retries Source: https://docs.hatchet.run/v1/retry-policies Examples of setting the retries property across different supported languages. ```python @simple_workflow.task(retries=3) def always_fail(input: EmptyModel, ctx: Context) -> dict[str, str]: raise Exception("simple task failed") ``` ```typescript export const retries = hatchet.task({ name: 'retries', retries: 3, fn: async (_, ctx) => { throw new Error('intentional failure'); }, }); ``` ```go retries := client.NewStandaloneTask("retries-task", func(ctx hatchet.Context, input RetriesInput) (*RetriesResult, error) { return nil, errors.New("intentional failure") }, hatchet.WithRetries(3)) ``` ```ruby SIMPLE_RETRY_WORKFLOW.task(:always_fail, retries: 3) do |input, ctx| raise "simple task failed" end ``` -------------------------------- ### Configure task retries and timeouts Source: https://docs.hatchet.run/v1/from-temporal-to-hatchet Examples of applying retry policies, backoff settings, and execution timeouts to tasks in various languages. ```python @hatchet.task( name="charge-order-with-retries", input_validator=OrderInput, retries=10, backoff_factor=2.0, backoff_max_seconds=10, execution_timeout=timedelta(seconds=30), schedule_timeout=timedelta(minutes=10), ) async def charge_order_with_retries(input: OrderInput, ctx: Context) -> ChargeOutput: # Raising `NonRetryableException` here would stop Hatchet from retrying at all. if ctx.retry_count < 2: raise RuntimeError(f"payment provider unavailable for {input.order_id}") return ChargeOutput(charged=True, charge_id=await submit_charge(input.order_id)) ``` ```typescript export const chargeOrderWithRetries = hatchet.task({ name: 'charge-order-with-retries', retries: 10, backoff: { // Factor to increase the wait time between retries. factor: 2, // Maximum number of seconds to wait between retries. maxSeconds: 10, }, executionTimeout: '30s', scheduleTimeout: '10m', fn: async (input: OrderInput): Promise => { const amountCents = await payments.charge(input.orderId); return { charged: amountCents > 0, amountCents }; }, }); ``` ```go chargeOrderWithRetries := client.NewStandaloneTask("charge-order-with-retries", func(ctx hatchet.Context, input OrderInput) (ChargeOutput, error) { amount, err := capturePayment(input.OrderID) if err != nil { return ChargeOutput{}, fmt.Errorf("charging order %q: %w", input.OrderID, err) } return ChargeOutput{Charged: true, AmountCents: amount}, nil }, // retries counts retries, not attempts: maximum_attempts=11 becomes 10. hatchet.WithRetries(10), // factor, maxBackoffSeconds hatchet.WithRetryBackoff(2, 10), hatchet.WithExecutionTimeout(30*time.Second), hatchet.WithScheduleTimeout(10*time.Minute), ) ``` ```ruby CHARGE_ORDER_WITH_RETRIES = HATCHET.task( name: "charge-order-with-retries", # Hatchet counts retries, not attempts: this is 11 attempts in total. retries: 10, backoff_factor: 2.0, backoff_max_seconds: 10, execution_timeout: 30, schedule_timeout: 600 ) do |input, ctx| charge = Payments.charge(input["order_id"], input["amount_cents"]) { "transaction_id" => charge["transaction_id"], "attempt" => ctx.retry_count + 1 } end ``` -------------------------------- ### Install Hatchet Docs for Claude Code Source: https://docs.hatchet.run/v1/using-coding-agents Configures Claude Code to access Hatchet documentation via the MCP server. ```bash hatchet docs install claude-code ``` -------------------------------- ### Execute child workflows in Temporal Source: https://docs.hatchet.run/v1/from-temporal-to-hatchet Examples of spawning child workflows concurrently in Temporal across multiple languages. ```python results = await asyncio.gather(*[ workflow.execute_child_workflow(ProcessItem.run, item) for item in items ]) ``` ```typescript import { executeChild } from "@temporalio/workflow"; const results = await Promise.all( items.map((item) => executeChild(processItem, { args: [item] })), ); ``` ```go // Collect the child futures, then await them. var futures []workflow.Future for _, item := range items { futures = append(futures, workflow.ExecuteChildWorkflow(ctx, ProcessItem, item)) } for _, f := range futures { var result ItemOutput if err := f.Get(ctx, &result); err != nil { return err } } ``` ```ruby results = items.map do |item| Temporalio::Workflow.execute_child_workflow( ProcessItem, item, schedule_to_close_timeout: 300 ) end ``` -------------------------------- ### Configure Cancel In Progress Concurrency Source: https://docs.hatchet.run/v1/concurrency Examples of setting the CANCEL_IN_PROGRESS limit strategy across different supported languages. ```python class WorkflowInput(BaseModel): group: str concurrency_cancel_in_progress_workflow = hatchet.workflow( name="ConcurrencyCancelInProgress", concurrency=ConcurrencyExpression( expression="input.group", max_runs=1, limit_strategy=ConcurrencyLimitStrategy.CANCEL_IN_PROGRESS, ), input_validator=WorkflowInput, ) ``` ```typescript export const concurrencyCancelInProgressWorkflow = hatchet.workflow({ name: 'concurrencycancelinprogress', concurrency: { expression: 'input.group', maxRuns: 1, limitStrategy: ConcurrencyLimitStrategy.CANCEL_IN_PROGRESS, }, }); ``` ```go var maxRuns int32 = 1 strategy := types.CancelInProgress return client.NewStandaloneTask("cancel-in-progress", func(ctx worker.HatchetContext, input ConcurrencyInput) (*TransformedOutput, error) { // Random sleep between 200ms and 1000ms time.Sleep(time.Duration(200+rand.Intn(800)) * time.Millisecond) return &TransformedOutput{ TransformedMessage: input.Message, }, nil }, hatchet.WithWorkflowConcurrency(types.Concurrency{ Expression: "input.GroupKey", MaxRuns: &maxRuns, LimitStrategy: &strategy, }), ) ``` ```ruby CONCURRENCY_CANCEL_IN_PROGRESS_WORKFLOW = HATCHET.workflow( name: "ConcurrencyCancelInProgress", concurrency: Hatchet::ConcurrencyExpression.new( expression: "input.group", max_runs: 1, limit_strategy: :cancel_in_progress ) ) ``` -------------------------------- ### Declare Static Rate Limits Source: https://docs.hatchet.run/v1/rate-limits Use the Admin client to define a rate limit key, its capacity, and the duration before starting workers. ```python RATE_LIMIT_KEY = "test-limit" hatchet.rate_limits.put(RATE_LIMIT_KEY, 2, RateLimitDuration.SECOND) ``` ```typescript hatchet.ratelimits.upsert({ key: 'api-service-rate-limit', limit: 10, duration: RateLimitDuration.SECOND, }); ``` ```go err = client.RateLimits().Upsert(features.CreateRatelimitOpts{ Key: RATE_LIMIT_KEY, Limit: 10, Duration: types.Second, }) if err != nil { log.Fatalf("failed to create rate limit: %v", err) } ``` ```ruby def main HATCHET.rate_limits.put(RATE_LIMIT_KEY, 2, :second) worker = HATCHET.worker( "rate-limit-worker", slots: 10, workflows: [RATE_LIMIT_WORKFLOW] ) worker.start end ``` -------------------------------- ### Fetch Task Stats via cURL Source: https://docs.hatchet.run/v1/autoscaling-workers Example command to retrieve task statistics using an API token. ```bash curl -H "Authorization: Bearer your-api-token-here" \ https://cloud.onhatchet.run/api/v1/tenants/707d0855-80ab-4e1f-a156-f1c4546cbf52/task-stats ``` -------------------------------- ### Execute child tasks in Hatchet Source: https://docs.hatchet.run/v1/from-temporal-to-hatchet Examples of implementing fan-out patterns using Hatchet durable tasks. ```python class ItemInput(BaseModel): item_id: str class ItemOutput(BaseModel): item_id: str result: str class FanOutInput(BaseModel): item_ids: list[str] class FanOutOutput(BaseModel): results: list[ItemOutput] @hatchet.task(name="process-item", input_validator=ItemInput) async def process_item(input: ItemInput, ctx: Context) -> ItemOutput: return ItemOutput(item_id=input.item_id, result=await handle_item(input.item_id)) @hatchet.durable_task(name="ProcessItems", input_validator=FanOutInput) async def process_items(input: FanOutInput, ctx: DurableContext) -> FanOutOutput: results = await asyncio.gather( *[ process_item.aio_run(ItemInput(item_id=item_id)) for item_id in input.item_ids ] ) return FanOutOutput(results=results) ``` ```typescript export const packOrder = hatchet.durableTask({ name: 'PackOrder', executionTimeout: '30m', fn: async (input: { items: string[] }): Promise<{ packed: number }> => { const results = await Promise.all(input.items.map((item) => processItem.run({ item }))); return { packed: results.filter((result) => result.packed).length }; }, }); ``` ```go var wg sync.WaitGroup outputs := make([]ItemOutput, len(items)) errs := make([]error, len(items)) wg.Add(len(items)) for i, item := range items { go func(i int, item Item) { defer wg.Done() result, err := processItem.Run(ctx, ItemInput{Item: item}) if err != nil { errs[i] = fmt.Errorf("running item %d: %w", i, err) return } if err := result.Into(&outputs[i]); err != nil { errs[i] = fmt.Errorf("decoding result for item %d: %w", i, err) } }(i, item) } wg.Wait() if err := errors.Join(errs...); err != nil { return ShipmentOutput{}, err } ``` ```ruby PROCESS_ITEM = HATCHET.task(name: "process-item") do |input, _ctx| { "item" => input["item"], "processed" => true } end FAN_OUT_ORDER_ITEMS = HATCHET.durable_task(name: "FanOutOrderItems", execution_timeout: 300) do |input, _ctx| items = input["items"] || [] # Ruby's trigger methods are synchronous, so fan out with the bulk API rather # than one call per child. results = PROCESS_ITEM.run_many( items.map { |item| PROCESS_ITEM.create_bulk_run_item(input: { "item" => item }) } ) { "results" => results } end ``` -------------------------------- ### Define a batch task Source: https://docs.hatchet.run/v1/batch-tasks Examples of defining batch tasks across different supported languages with specific size and interval configurations. ```python @hatchet.batch_task( batch_max_size=3, batch_max_interval=timedelta(milliseconds=200), input_validator=SimpleInput, ) async def batch_simple( tasks: dict[BatchMemberId, SimpleInput], context: Context ) -> dict[BatchMemberId, SimpleOutput]: return { id: SimpleOutput(transformed_message=inp.message.upper()) for id, inp in tasks.items() } ``` ```typescript export const batchSimple = hatchet.batchTask({ name: 'batch-simple', batch: { maxSize: 3, maxInterval: 200 }, fn: async (tasks: Record) => { const out: Record = {}; Object.entries(tasks).forEach(([id, input]) => { out[id] = { transformed_message: input.message.toUpperCase() }; }); return out; }, }); ``` ```go batchSimple := client.NewStandaloneBatchTask("batch-simple", func(ctx hatchet.Context, tasks map[string]SimpleInput) (map[string]SimpleOutput, error) { out := make(map[string]SimpleOutput, len(tasks)) for id, inp := range tasks { out[id] = SimpleOutput{TransformedMessage: strings.ToUpper(inp.Message)} } return out, nil }, hatchet.BatchConfig{ MaxSize: 3, MaxInterval: durationPtr(200 * time.Millisecond), }, ) ``` ```ruby BATCH_SIMPLE = HATCHET.batch_task( name: 'ruby-e2e-batch-simple', batch: Hatchet::BatchTaskConfig.new(max_size: 3, max_interval_ms: 200) ) do |inputs, _ctx| inputs.transform_values { |input| { 'transformed_message' => input['message'].upcase } } end ``` -------------------------------- ### Run Celery worker with specific queue Source: https://docs.hatchet.run/v1/from-celery-to-hatchet Start a Celery worker process bound to a specific queue. ```bash celery -A tasks worker -Q image-processing ``` -------------------------------- ### Define Temporal Workflow for Order Processing Source: https://docs.hatchet.run/v1/from-temporal-to-hatchet Examples of Temporal workflow definitions that execute three sequential activities. ```python @workflow.defn class ProcessOrderWorkflow: @workflow.run async def run(self, order_id: str) -> dict: valid = await workflow.execute_activity( validate_order, order_id, start_to_close_timeout=timedelta(seconds=30) ) charged = await workflow.execute_activity( charge_order, order_id, start_to_close_timeout=timedelta(seconds=30) ) return await workflow.execute_activity( fulfill_order, order_id, start_to_close_timeout=timedelta(seconds=30) ) ``` ```typescript import { proxyActivities } from "@temporalio/workflow"; // Only import the activity types import type * as activities from "./activities"; const { validateOrder, chargeOrder, fulfillOrder } = proxyActivities({ startToCloseTimeout: "30 seconds", }); export async function processOrder(orderId: string): Promise { await validateOrder(orderId); await chargeOrder(orderId); return fulfillOrder(orderId); } ``` ```go func ProcessOrderWorkflow(ctx workflow.Context, orderID string) (bool, error) { ctx = workflow.WithActivityOptions(ctx, workflow.ActivityOptions{ StartToCloseTimeout: 30 * time.Second, }) if err := workflow.ExecuteActivity(ctx, ValidateOrder, orderID).Get(ctx, nil); err != nil { return false, err } if err := workflow.ExecuteActivity(ctx, ChargeOrder, orderID).Get(ctx, nil); err != nil { return false, err } var fulfilled bool err := workflow.ExecuteActivity(ctx, FulfillOrder, orderID).Get(ctx, &fulfilled) return fulfilled, err } ``` ```ruby require "temporalio/workflow" class ProcessOrderWorkflow < Temporalio::Workflow::Definition def execute(order_id) Temporalio::Workflow.execute_activity( ValidateOrder, order_id, start_to_close_timeout: 30 ) Temporalio::Workflow.execute_activity( ChargeOrder, order_id, start_to_close_timeout: 30 ) Temporalio::Workflow.execute_activity( FulfillOrder, order_id, start_to_close_timeout: 30 ) end end ``` -------------------------------- ### Define and start a Hatchet worker Source: https://docs.hatchet.run/v1/from-celery-to-hatchet Programmatic worker definition in Hatchet, explicitly registering workflows and setting execution slots. ```python def start_worker() -> None: worker = hatchet.worker("image-worker", slots=4, workflows=[process_image]) worker.start() ``` -------------------------------- ### Initialize shared application client Source: https://docs.hatchet.run/v1/from-celery-to-hatchet Comparison of initializing a Celery app versus a Hatchet client. ```python from celery import Celery app = Celery("tasks", broker="redis://localhost:6379", backend="redis://localhost:6379") ``` ```python from hatchet_sdk import Hatchet hatchet = Hatchet() ``` -------------------------------- ### Define Hatchet Durable Task for Order Processing Source: https://docs.hatchet.run/v1/from-temporal-to-hatchet Examples of Hatchet durable tasks that replicate the sequential execution logic of the original Temporal workflows. ```python @hatchet.durable_task(name="ProcessOrder", input_validator=OrderInput) async def process_order(input: OrderInput, ctx: DurableContext) -> FulfillOutput: await validate_order.aio_run(input) await charge_order.aio_run(input) return await fulfill_order.aio_run(input) ``` ```typescript export const processOrder = hatchet.durableTask({ name: 'ProcessOrder', executionTimeout: '10m', fn: async (input: OrderInput): Promise => { await validateOrder.run(input); await chargeOrder.run(input); return fulfillOrder.run(input); }, }); ``` ```go processOrder := client.NewStandaloneDurableTask("ProcessOrder", func(ctx hatchet.DurableContext, input OrderInput) (FulfillOutput, error) { if _, err := validateOrder.Run(ctx, input); err != nil { return FulfillOutput{}, fmt.Errorf("validating order: %w", err) } if _, err := chargeOrder.Run(ctx, input); err != nil { return FulfillOutput{}, fmt.Errorf("charging order: %w", err) } result, err := fulfillOrder.Run(ctx, input) if err != nil { return FulfillOutput{}, fmt.Errorf("fulfilling order: %w", err) } var fulfilled FulfillOutput if err := result.Into(&fulfilled); err != nil { return FulfillOutput{}, fmt.Errorf("decoding fulfillment: %w", err) } return fulfilled, nil }, ) ``` -------------------------------- ### Migrate from V1 Generics SDK to V1 Reflection SDK Source: https://docs.hatchet.run/v1/migrating/migration-guide-go Transition from the deprecated V1 Generics SDK factory-based task creation to the current Reflection SDK. ```go package main import ( "log" "strings" "github.com/hatchet-dev/hatchet/pkg/client/create" v1 "github.com/hatchet-dev/hatchet/pkg/v1" "github.com/hatchet-dev/hatchet/pkg/v1/factory" "github.com/hatchet-dev/hatchet/pkg/v1/worker" "github.com/hatchet-dev/hatchet/pkg/v1/workflow" v0Worker "github.com/hatchet-dev/hatchet/pkg/worker" ) func V1Old() { hatchet, err := v1.NewHatchetClient() if err != nil { log.Fatal(err) } simple := factory.NewTask( create.StandaloneTask{Name: "simple-task", OnEvents: []string{"user:create"}}, func(ctx v0Worker.HatchetContext, input SimpleInput) (*SimpleResult, error) { return &SimpleResult{TransformedMessage: strings.ToLower(input.Message)}, nil }, hatchet, ) _, err = hatchet.Worker(worker.WorkerOpts{ Name: "worker", Workflows: []workflow.WorkflowBase{simple}, }) if err != nil { log.Fatal(err) } } ``` ```go package main import ( "log" "strings" hatchet "github.com/hatchet-dev/hatchet/sdks/go" ) type SimpleInput struct { Message string `json:"message"` } type SimpleResult struct { TransformedMessage string `json:"result"` } func V1() { client, err := hatchet.NewClient() if err != nil { log.Fatal(err) } workflow := client.NewStandaloneTask("simple-workflow", func(ctx hatchet.Context, input SimpleInput) (SimpleResult, error) { log.Println("executed step1") return SimpleResult{TransformedMessage: strings.ToLower(input.Message)}, nil }, hatchet.WithWorkflowEvents("user:create")) _, err = client.NewWorker( "worker", hatchet.WithWorkflows(workflow), ) if err != nil { log.Fatal(err) } } ``` -------------------------------- ### Run Hatchet Worker Source: https://docs.hatchet.run/v1/quickstart Starts the worker process locally. ```sh hatchet worker dev ``` -------------------------------- ### Migrate from V0 SDK to V1 Reflection SDK Source: https://docs.hatchet.run/v1/migrating/migration-guide-go Compare the legacy V0 worker registration pattern with the current V1 Reflection SDK approach. ```go package main import ( "log" "github.com/hatchet-dev/hatchet/pkg/client" v0Worker "github.com/hatchet-dev/hatchet/pkg/worker" ) func V0() { c, err := client.New() if err != nil { log.Fatal(err) } worker, err := v0Worker.NewWorker( v0Worker.WithClient(c), v0Worker.WithName("worker"), ) if err != nil { log.Fatal(err) } err = worker.RegisterWorkflow( &v0Worker.WorkflowJob{ On: v0Worker.Event("user:create"), Name: "simple-workflow", Steps: []*v0Worker.WorkflowStep{ { Name: "step1", Function: func(ctx v0Worker.HatchetContext) error { log.Println("executed step1") return nil }, }, }, }, ) if err != nil { log.Fatal(err) } } ``` ```go package main import ( "log" "strings" hatchet "github.com/hatchet-dev/hatchet/sdks/go" ) type SimpleInput struct { Message string `json:"message"` } type SimpleResult struct { TransformedMessage string `json:"result"` } func V1() { client, err := hatchet.NewClient() if err != nil { log.Fatal(err) } workflow := client.NewStandaloneTask("simple-workflow", func(ctx hatchet.Context, input SimpleInput) (SimpleResult, error) { log.Println("executed step1") return SimpleResult{TransformedMessage: strings.ToLower(input.Message)}, nil }, hatchet.WithWorkflowEvents("user:create")) _, err = client.NewWorker( "worker", hatchet.WithWorkflows(workflow), ) if err != nil { log.Fatal(err) } } ``` -------------------------------- ### GET /metrics Source: https://docs.hatchet.run/v1/worker-healthchecks Returns metrics for the worker, intended for use by monitoring systems like Prometheus. ```APIDOC ## GET /metrics ### Description Returns metrics for the worker, intended for use by monitoring systems like Prometheus. ### Method GET ### Endpoint /metrics ### Response #### Success Response (200) - **metrics** (text/plain) - Prometheus-formatted metrics data ``` -------------------------------- ### Define a GPU-enabled Dockerfile Source: https://docs.hatchet.run/v1/compute/gpu A base Dockerfile configuration for installing CUDA libraries and setting up an application entrypoint. ```dockerfile # Base image FROM ubuntu:22.04 # Install CUDA and required libraries RUN apt-get update && apt-get install -y --no-install-recommends \ cuda-nvcc-12-2 \ libcublas-12-2 \ libcudnn8 \ && rm -rf /var/lib/apt/lists/* # Your application setup WORKDIR /app COPY . . # Application entrypoint CMD ["python", "worker.py"] ``` -------------------------------- ### GET /api/v1/tenants/{tenantId}/prometheus-metrics Source: https://docs.hatchet.run/v1/prometheus-metrics Retrieves Prometheus metrics for a specific tenant in Prometheus Text Format. ```APIDOC ## GET /api/v1/tenants/{tenantId}/prometheus-metrics ### Description Retrieves metrics for individual tenants in Prometheus Text Format, which can be scraped by services like Grafana or DataDog. ### Method GET ### Endpoint /api/v1/tenants/{tenantId}/prometheus-metrics ### Parameters #### Path Parameters - **tenantId** (string) - Required - The unique identifier of the tenant. ### Authentication - **Authorization** (header) - Required - Bearer token authentication using a valid API token. ### Response #### Success Response (200) - **Content-Type** (text/plain) - Returns metrics in standard Prometheus Text Format, including HELP comments, TYPE declarations, and metric samples. ``` -------------------------------- ### Instantiate workflows and tasks with the factory pattern Source: https://docs.hatchet.run/v1/migrating/v1-sdk-improvements Use the factory pattern to define and execute workflows and tasks with full type safety. ```go type SimpleInput struct { Message string } type SimpleResult struct { TransformedMessage string } simple := factory.NewTask( create.StandaloneTask{ Name: "simple-task", }, func(ctx worker.HatchetContext, input SimpleInput) (*SimpleResult, error) { return &SimpleResult{ TransformedMessage: strings.ToLower(input.Message), }, nil }, hatchet, // a Hatchet client instance ) // somewhere else in your code result, err := simple.Run(ctx, SimpleInput{ Message: "Hello, World!", }) // result is fully typed! ``` -------------------------------- ### Define Workflow in Ruby Source: https://docs.hatchet.run/v1/advanced-assignment/worker-affinity Initialize a workflow using the Ruby SDK. ```ruby AFFINITY_WORKER_WORKFLOW = HATCHET.workflow(name: "AffinityWorkflow") ``` -------------------------------- ### Configure CUDA version compatibility Source: https://docs.hatchet.run/v1/compute/gpu Ensure the installed CUDA toolkit version matches the host driver requirements. ```dockerfile # Ensure CUDA toolkit version matches driver RUN apt-get install -y cuda-nvcc-12-2 ``` -------------------------------- ### Import Go OpenTelemetry package Source: https://docs.hatchet.run/v1/opentelemetry Import the Hatchet OpenTelemetry package for Go applications. ```go import hatchetotel "github.com/hatchet-dev/hatchet/sdks/go/opentelemetry" ``` -------------------------------- ### Initialize a Worker with Middleware in TypeScript Source: https://docs.hatchet.run/v1/middleware Demonstrates creating a worker using a middleware-enabled client and registering workflows. ```typescript import { hatchetWithMiddleware } from './client'; import { taskWithMiddleware } from './workflow'; async function main() { const worker = await hatchetWithMiddleware.worker('task-with-middleware', { workflows: [taskWithMiddleware], }); await worker.start(); } if (require.main === module) { main(); } ``` -------------------------------- ### Implement multi-stage Docker builds Source: https://docs.hatchet.run/v1/compute/gpu Use multi-stage builds to separate build-time dependencies from the final runtime image. ```dockerfile # Build stage FROM ubuntu:22.04 AS builder RUN apt-get update && apt-get install -y cuda-nvcc-12-2 # Runtime stage FROM ubuntu:22.04 COPY --from=builder /usr/local/cuda-12.2 /usr/local/cuda-12.2 ``` -------------------------------- ### Manual Slot Release Implementation Source: https://docs.hatchet.run/v1/advanced-assignment/manual-slot-release Examples of using the release_slot method to free worker capacity after resource-heavy operations. ```python slot_release_workflow = hatchet.workflow(name="SlotReleaseWorkflow") @slot_release_workflow.task() def step1(input: EmptyModel, ctx: Context) -> dict[str, str]: print("RESOURCE INTENSIVE PROCESS") time.sleep(10) # 👀 Release the slot after the resource-intensive process, so that other steps can run ctx.release_slot() print("NON RESOURCE INTENSIVE PROCESS") return {"status": "success"} ``` ```go _ = workflow.NewTask("step1", func(ctx hatchet.Context, _ any) (*StepOutput, error) { fmt.Println("RESOURCE INTENSIVE PROCESS") time.Sleep(10 * time.Second) // Release the slot after the resource-intensive process, // so that other steps can run on this worker. if releaseErr := ctx.ReleaseSlot(); releaseErr != nil { return nil, fmt.Errorf("failed to release slot: %w", releaseErr) } fmt.Println("NON RESOURCE INTENSIVE PROCESS") return &StepOutput{Status: "success"}, nil }) ``` ```ruby SLOT_RELEASE_WORKFLOW = HATCHET.workflow(name: "SlotReleaseWorkflow") SLOT_RELEASE_WORKFLOW.task(:step1) do |input, ctx| puts "RESOURCE INTENSIVE PROCESS" sleep 10 # Release the slot after the resource-intensive process, so that other steps can run ctx.release_slot puts "NON RESOURCE INTENSIVE PROCESS" { "status" => "success" } end ```