mirror/hatchet
1
0
Fork 0
mirror of https://github.com/hatchet-dev/hatchet.git synced 2026-04-21 17:28:59 -05:00
Code Issues Packages Projects Releases Wiki Activity

[Docs] Concurrency, Additional Metadata (#1415)

Browse Source 
* fix: child spawning python example

* fix: more things

* feat: additional meta

* feat: concurrency

* fix: proof

* fix: lint

* fix: python migration

* fix: test

* fix: lint

* fix: link on migration guide

---------

Co-authored-by: Alexander Belanger <alexander@hatchet.run>
This commit is contained in:
Matt Kaye
2025-03-26 21:34:02 -04:00
committed by GitHub
parent 9c389ee5e3
commit 0c4e52a97c
14 changed files with 426 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frontend/docs/pages/home/_meta.json
+6 -10
View File
@@ -50,6 +50,9 @@
"child-spawning": {
"title": "Child Spawning"
},
"additional-metadata": {
"title": "Additional Metadata"
},
"--error-handling": {
"title": "Error Handling",
@@ -83,19 +86,12 @@
"docker": "Running with Docker",
"worker-healthchecks": "Worker Health Checks",
"additional-metadata": {
"title": "Additional Metadata",
"display": "hidden"
},
"--flow-control": {
"title": "Flow Control",
"type": "separator",
"display": "hidden"
"type": "separator"
},
"concurrency-keys": {
"title": "Concurrency Keys",
"display": "hidden"
"concurrency": {
"title": "Concurrency"
},
"rate-limits": {
"title": "Rate Limits",
frontend/docs/pages/home/additional-metadata.mdx
+121 -1
View File
@@ -1 +1,121 @@
TODO V1 DOCS
import { Callout, Card, Cards, Steps, Tabs } from "nextra/components";
import UniversalTabs from "../../components/UniversalTabs";
# Additional Metadata
Hatchet allows you to attach arbitrary key-value string pairs to events and workflow runs, which can be used for filtering, searching, or any other lookup purposes. This additional metadata is not part of the event payload or workflow input data but provides supplementary information for better organization and discoverability.
## Attaching Metadata
You can attach additional metadata when pushing events or triggering workflow runs using the Hatchet client libraries:
<Tabs items={['Event Push', 'Workflow Run Trigger']}>
<Tabs.Tab>
<UniversalTabs items={['Python', 'Typescript', 'Go']}>
<Tabs.Tab>
```python
hatchet.event.push(
"user:create",
{'userId': '1234'},
options=PushEventOptions(
additional_metadata={
"source": "api" # Arbitrary key-value pair
}
)
)
```
</Tabs.Tab>
<Tabs.Tab>
```typescript
hatchet.event.push(
'user:create',
{
test: 'test',
},
{
additionalMetadata: {
source: 'api', // Arbitrary key-value pair
},
}
);
```
</Tabs.Tab>
<Tabs.Tab>
```go
err := c.Event().Push(
context.Background(),
"user:create",
testEvent,
client.WithEventMetadata(map[string]string{
"source": "api", // Arbitrary key-value pair
}),
)
```
</Tabs.Tab>
</UniversalTabs>
</Tabs.Tab>
<Tabs.Tab>
<UniversalTabs items={['Python', 'Typescript', 'Go']}>
<Tabs.Tab>
```python
simple_workflow.run(
SimpleInput(user_id=1234),
options=TriggerWorkflowOptions(
additional_metadata={
"hello": "moon" # Arbitrary key-value pair
}
)
)
```
</Tabs.Tab>
<Tabs.Tab>
```typescript
const workflowRunId = await hatchet.admin.run_workflow(
'user-workflow',
{
userId: '1234',
},
{
additionalMetadata: {
source: 'api', // Arbitrary key-value pair
},
}
);
```
</Tabs.Tab>
<Tabs.Tab>
```go
workflowRunId, err := c.Admin().RunWorkflow(
"user-workflow",
&userCreateEvent{
UserID: "1234",
},
client.WithRunMetadata(map[string]interface{}{
"source": "api", // Arbitrary key-value pair
}),
)
```
</Tabs.Tab>
</UniversalTabs>
</Tabs.Tab>
</Tabs>
> **Note:** This feature is in development. Cron, Schedule, and Child Workflow triggers do not currently support additional metadata.
## Filtering in the Dashboard
Once you've attached additional metadata to events or workflow runs, this data will be available in the Event and Workflow Run list views in the Hatchet dashboard. You can use the filter input field to search for events or workflow runs based on the additional metadata key-value pairs you've attached.
For example, you can filter events by the `source` metadata keys to quickly find events originating from a specific source or environment.
![Blocks](/addl-meta.gif)
## Use Cases
Some common use cases for additional metadata include:
- Tagging events or workflow runs with environment information (e.g., `production`, `staging`, `development`)
- Specifying the source or origin of events (e.g., `api`, `webhook`, `manual`)
- Categorizing events or workflow runs based on business-specific criteria (e.g., `priority`, `region`, `product`)
By leveraging additional metadata, you can enhance the organization, searchability, and discoverability of your events and workflow runs within Hatchet.
frontend/docs/pages/home/child-spawning.mdx
+13 -41
View File
@@ -2,6 +2,12 @@ import { Callout, Card, Cards, Steps, Tabs } from "nextra/components";
import UniversalTabs from "@/components/UniversalTabs";
import { GithubSnippet, getSnippets } from "@/components/code";
export const FanoutPy = {
path: "examples/fanout/worker.py",
};
export const getStaticProps = ({}) => getSnippets([FanoutPy]);
# Procedural Child Workflow Spawning
Hatchet supports the dynamic creation of child workflows during a parent workflow's execution. This powerful feature enables:
@@ -18,49 +24,15 @@ To implement child workflow spawning, you first need to create both parent and c
<UniversalTabs items={["Python", "Typescript", "Go"]}>
<Tabs.Tab title="Python">
```python
from hatchet_sdk import Context, Hatchet
from pydantic import BaseModel
First, we'll declare a couple of workflows for the parent and child:
hatchet = Hatchet(debug=True)
<GithubSnippet src={FanoutPy} target="FanoutParent" />
# Child workflow definition
class ChildInput(BaseModel):
n: int
We also created a task on the parent workflow that spawns the child workflows. Now, we'll add a couple of tasks to the child workflow:
class ValueOutput(BaseModel):
value: int
<GithubSnippet src={FanoutPy} target="FanoutChild" />
child = hatchet.workflow(name="child")
@child.task()
def value_task(input: ChildInput, ctx: Context) -> ValueOutput:
return ValueOutput(value=input.n)
# Parent workflow definition
class ParentInput(BaseModel):
n: int
class SumOutput(BaseModel):
result: int
parent = hatchet.workflow(name="parent")
@parent.task()
def sum_task(input: ParentInput, ctx: Context) -> SumOutput:
sum_value = 0
# Run multiple child workflows concurrently
child_results = await child.aio_run_many([
child.create_run_workflow_config(input=ChildInput(n=i))
for i in range(input.n)
])
# Sum the results
sum_value = sum(child.get("result") for child in child_results.values())
return SumOutput(result=sum_value)
```
And that's it! The fanout parent will run and spawn the child, and then will collect the results from its tasks.
</Tabs.Tab>
<Tabs.Tab title="Typescript">
@@ -309,7 +281,7 @@ import asyncio
async def run_child_workflows(n: int) -> list[dict[str, Any]]:
return await child.aio_run_many([
create_run_workflow_config(
child.create_run_workflow_config(
options=TriggerWorkflowOptions(
input=ChildInput(n=i),
)
@@ -380,7 +352,7 @@ When working with child workflows, it's important to properly handle errors. Her
```python
try:
child_result = child.run_as_child(ctx, ChildInput(n=i))
child_result = child.run(ChildInput(a="foobar"))
except Exception as e:
# Handle error from child workflow
print(f"Child workflow failed: {e}")
frontend/docs/pages/home/concurrency/_meta.json
+5
View File
@@ -0,0 +1,5 @@
{
"overview": "Overview",
"cancel-in-progress": "Cancel In Progress",
"round-robin": "Round Robin"
}
frontend/docs/pages/home/concurrency/cancel-in-progress.mdx
+111
View File
@@ -0,0 +1,111 @@
import { Tabs, Callout } from "nextra/components";
import UniversalTabs from "../../../components/UniversalTabs";
import { GithubSnippet, getSnippets } from "@/components/code";
export const ConcurrencyPy = {
path: "examples/concurrency_limit/worker.py",
};
export const getStaticProps = ({}) => getSnippets([ConcurrencyPy]);
# The `CANCEL_IN_PROGRESS` Concurrency Limit Strategy in Hatchet
Hatchet's `CANCEL_IN_PROGRESS` concurrency limit strategy is a powerful tool for managing resource contention in your workflows. This strategy allows you to cancel currently running workflow instances to free up slots for new instances when the concurrency limit is reached.
## How it works
When a workflow run finishes and the concurrency limit is reached for the workflow version and group key, the `CANCEL_IN_PROGRESS` strategy will:
1. Fetch all currently running workflow runs for the given workflow version and concurrency key.
2. Fetch the queued workflow runs for the same workflow version and group key, up to the maximum number of concurrent runs allowed.
3. Cancel the oldest running workflow runs to make room for the queued runs, ensuring that there is space for all the queued runs to start executing.
4. Start executing the queued workflow runs in the freed-up slots.
This strategy ensures that queued workflow runs can start executing as soon as possible, even if the concurrency limit is reached, by canceling older running instances.
## When to use `CANCEL_IN_PROGRESS`
The `CANCEL_IN_PROGRESS` strategy is particularly useful in scenarios where:
- You have long-running workflow instances that may become stale or irrelevant if newer instances are triggered.
- You want to prioritize processing the most recent data or events, even if it means canceling older workflow instances.
- You have resource-intensive workflows where it's more efficient to cancel an in-progress instance and start a new one than to wait for the old instance to complete.
- Your user UI allows for multiple inputs, but only the most recent is relevant (i.e. chat messages, form submissions, etc.).
However, it's important to note that canceling a workflow instance may leave your system in an inconsistent state if the canceled instance was in the middle of updating a resource. Make sure to design your workflows to handle cancellation gracefully and ensure data consistency.
{/* TODO V1 DOCS - Add back link to cancellation docs on `handle cancellation gracefully` when we write those */}
## How to use `CANCEL_IN_PROGRESS`
To use the `CANCEL_IN_PROGRESS` concurrency limit strategy, define a `concurrency` configuration in your workflow definition:
<UniversalTabs items={['Python', 'Typescript']}>
{/* <UniversalTabs items={['Python', 'Typescript', 'Go']}> TODO V1 DOCS -- Add this back */}
<Tabs.Tab>
<GithubSnippet src={ConcurrencyPy} target="Workflow" />
</Tabs.Tab>
<Tabs.Tab>
```typescript
hatchet.workflow<SimpleInput, SimpleOutput>({
name: "simple-concurrency",
concurrency: {
maxRuns: 100,
limitStrategy: ConcurrencyLimitStrategy.GROUP_ROUND_ROBIN,
expression: "input.GroupKey",
},
});
```
</Tabs.Tab>
{/* <Tabs.Tab>
TODO V1 DOCS -- Go example
```go
import (
"github.com/hatchet-dev/hatchet/pkg/client/types"
"github.com/hatchet-dev/hatchet/pkg/worker"
)
type MyUser struct {
UserId string `json:"user_id"`
}
err = w.RegisterWorkflow(
&worker.WorkflowJob{
Name: "concurrency-limit-per-user",
On: worker.Events("concurrency-test-event"),
Description: "This limits concurrency to 1 run at a time per user.",
Concurrency: worker.Concurrency(getConcurrencyKey).MaxRuns(10).LimitStrategy(types.CancelInProgress),
Steps: []\*worker.WorkflowStep{
// your steps here...
},
},
)
```
</Tabs.Tab> */}
CEL expressions are evaluated on the Hatchet engine (not in your worker) and can be used to extract values from the workflow input or [additional metadata](../additional-metadata.mdx).
In this example:
- `maxRuns` sets the maximum number of concurrent instances allowed for this workflow.
- `limitStrategy` is set to `CANCEL_IN_PROGRESS`, indicating that when the concurrency limit is reached, currently running instances should be canceled to make room for new ones.
- `expression` (recommended) is a CEL expression that is used to determine the concurrency key. In this example, the expression is `input.group_key`, which means that instances are grouped by `group_key`.
</UniversalTabs>
<Callout type="info">
Note that only one of `key` or `expression` should be provided, not both.
</Callout>
With this configuration, Hatchet will automatically manage your workflow's concurrency, canceling in-progress instances as needed to ensure that new instances can start executing immediately.
Remember to design your workflows to handle cancellation gracefully, and consider using checkpointing or idempotency techniques to ensure data consistency in the face of canceled instances.
```
frontend/docs/pages/home/concurrency/overview.mdx
+62
View File
@@ -0,0 +1,62 @@
import { Tabs } from "nextra/components";
import UniversalTabs from "../../../components/UniversalTabs";
## Concurrency Control in Hatchet Workflows
Hatchet provides powerful concurrency control features to help you manage the execution of your workflows. This is particularly useful when you have workflows that may be triggered frequently or have long-running steps, and you want to limit the number of concurrent executions to prevent overloading your system, ensure fairness, or avoid race conditions.
### Why use concurrency control?
There are several reasons why you might want to use concurrency control in your Hatchet workflows:
1. **Fairness**: When you have multiple clients or users triggering workflows, concurrency control can help ensure fair access to resources. By limiting the number of concurrent runs per client or user, you can prevent a single client from monopolizing the system and ensure that all clients get a fair share of the available resources.
2. **Resource management**: If your workflow steps are resource-intensive (e.g., they make external API calls or perform heavy computations), running too many instances concurrently can overload your system. By limiting concurrency, you can ensure your system remains stable and responsive.
3. **Avoiding race conditions**: If your workflow steps modify shared resources, running multiple instances concurrently can lead to race conditions and inconsistent data. Concurrency control helps you avoid these issues by ensuring only a limited number of instances run at a time.
4. **Compliance with external service limits**: If your workflow steps interact with external services that have rate limits, concurrency control can help you stay within those limits and avoid being throttled or blocked.
5. **Spike Protection**: When you have workflows that are triggered by external events, such as webhooks or user actions, you may experience spikes in traffic that can overwhelm your system. Concurrency control can help you manage these spikes by limiting the number of concurrent runs and queuing new runs until resources become available.
### Available Strategies:
- [`CANCEL_IN_PROGRESS`](./cancel-in-progress): Cancel the currently running workflow instances for the same concurrency key to free up slots for the new instance.
- [`GROUP_ROUND_ROBIN`](./round-robin): Distribute workflow instances across available slots in a round-robin fashion based on the `key` function.
- `CANCEL_NEWEST`: Cancel the newest workflow runs when the concurrency limit is reached.
> We're always open to adding more strategies to fit your needs. Join our [discord](https://discord.gg/ZMeUafwH89) to let us know.
### Setting concurrency on workers
In addition to setting concurrency limits at the workflow level, you can also control concurrency at the worker level by passing the `slots` (formerly `max_runs`) option when creating a new `Worker` instance:
<UniversalTabs items={['Python', 'Typescript', 'Go']}>
<Tabs.Tab>
```python
worker = hatchet.worker("my-worker", slots=5)
```
</Tabs.Tab>
<Tabs.Tab>
```typescript
const worker = hatchet.worker("my-worker", {
slots: 5,
});
```
</Tabs.Tab>
<Tabs.Tab>
```go
w, err := worker.NewWorker(
worker.WithClient(c),
worker.WithMaxRuns(5),
worker.WithName("my-worker")
)
```
</Tabs.Tab>
</UniversalTabs>
This limits the worker to a maximum of 5 concurrent step runs across all workflows. Once the limit is reached, the worker will not accept new actions until a running step completes.
Worker-level concurrency limits are independent of workflow-level limits. The Hatchet engine automatically distributes actions to available workers, and queues actions if all workers are at their concurrency limit.
By combining workflow-level and worker-level concurrency controls, you can fine-tune your Hatchet system for optimal performance and resource utilization.
frontend/docs/pages/home/concurrency/round-robin.mdx
+91
View File
@@ -0,0 +1,91 @@
import { Tabs } from "nextra/components";
import UniversalTabs from "../../../components/UniversalTabs";
import { GithubSnippet, getSnippets } from "@/components/code";
export const ConcurrencyPy = {
path: "examples/concurrency_limit_rr/worker.py",
};
export const ConcurrencyTs = {
path: "src/v1/examples/concurrency-rr/workflow.ts",
};
export const getStaticProps = ({}) =>
getSnippets([ConcurrencyPy, ConcurrencyTs]);
# The `GROUP_ROUND_ROBIN` Concurrency Limit Strategy in Hatchet
Hatchet's `GROUP_ROUND_ROBIN` concurrency limit strategy is an advanced way to manage resource contention in your workflows while ensuring fair distribution of resources across different groups of tenants, users, or other concurrency key. This strategy allows you to process workflow instances in a round-robin fashion within each group, as defined by a key function.
## How it works
When a new workflow instance is triggered, the `GROUP_ROUND_ROBIN` strategy will:
1. Determine the group that the instance belongs to based on the `key` function defined in the workflow's concurrency configuration.
2. Check if there are any available slots for the instance's group based on the `maxRuns` limit of available workers.
3. If a slot is available, the new workflow instance starts executing immediately.
4. If no slots are available, the new workflow instance is added to a queue for its group.
5. When a running workflow instance completes and a slot becomes available for a group, the next queued instance for that group (in round-robin order) is dequeued and starts executing.
This strategy ensures that workflow instances are processed fairly across different groups, preventing any one group from monopolizing the available resources. It also helps to reduce latency for instances within each group, as they are processed in a round-robin fashion rather than strictly in the order they were triggered.
## When to use `GROUP_ROUND_ROBIN`
The `GROUP_ROUND_ROBIN` strategy is particularly useful in scenarios where:
- You have multiple clients or users triggering workflow instances, and you want to ensure fair resource allocation among them.
- You want to process instances within each group in a round-robin fashion to minimize latency and ensure that no single instance within a group is starved for resources.
- You have long-running workflow instances and want to avoid one group's instances monopolizing the available slots.
Keep in mind that the `GROUP_ROUND_ROBIN` strategy may not be suitable for all use cases, especially those that require strict ordering or prioritization of the most recent events.
## How to use `GROUP_ROUND_ROBIN`
To use the `GROUP_ROUND_ROBIN` concurrency limit strategy, define a `concurrency` configuration in your workflow definition:
CEL expressions are evaluated on the Hatchet engine (not in your worker) and can be used to extract values from the workflow input or [additional metadata](/features/additional-metadata).
<UniversalTabs items={['Python', 'Typescript', 'Go']}>
<Tabs.Tab>
<GithubSnippet src={ConcurrencyPy} target="Workflow" />
</Tabs.Tab>
<Tabs.Tab>
<GithubSnippet src={ConcurrencyTs} target="Workflow" />
</Tabs.Tab>
<Tabs.Tab>
{/* TODO V1 DOCS -- Go example here */}
```go
import (
"github.com/hatchet-dev/hatchet/pkg/client/types"
"github.com/hatchet-dev/hatchet/pkg/worker"
)
type MyUser struct {
UserId string `json:"user_id"`
}
err = w.RegisterWorkflow(
&worker.WorkflowJob{
Name: "concurrency-limit-per-user",
On: worker.Events("concurrency-test-event"),
Description: "This limits concurrency to 10 run at a time per user.",
Concurrency: worker.Concurrency(getConcurrencyKey).MaxRuns(10).LimitStrategy(types.GroupRoundRobin),
Steps: []*worker.WorkflowStep{
// your steps here...
},
},
)
```
</Tabs.Tab>
</UniversalTabs>
> NOTE: Only one of `key` or `expression` should be provided, not both.
With this configuration, Hatchet will automatically manage your workflow's concurrency, processing instances within each group in a round-robin fashion and ensuring fair distribution of resources across groups.
frontend/docs/pages/home/migration-guide-engine.mdx
+1 -1
View File
@@ -63,7 +63,7 @@ For instructions on upgrading to the latest SDKs, please refer to the following
While we'd prefer to avoid any breaking changes, v1 is architecturally very different from v0, which means that the following APIs will be modified/replaced:
- While we haven't published an official REST API doc, we have often recommended usage of the REST API in our SDKs to implement replays, retrieving task status, and dead-letter queueing. The current API for listing, cancelling and replaying workflow runs will not work against a v1 engine. We will be providing an upgrade path using new endpoints which are more conducive to bulk replays and cancellations.
- We will only be supporting [CEL-based concurrency keys](https://docs.hatchet.run/home/features/concurrency/round-robin#how-to-use-group_round_robin), and we will not be supporting custom concurrency methods defined on the client. If you require custom logic to compute the concurrency key that can't be captured in a CEL expression, we recommend computing the key ahead of time and passing it as part of the input to the workflow. **Workflows registered against a v1 engine with a custom concurrency method (instead of an expression) will not use a concurrency queue.**
- We will only be supporting [CEL-based concurrency keys](https://docs.hatchet.run/home/concurrency/round-robin#how-to-use-group_round_robin), and we will not be supporting custom concurrency methods defined on the client. If you require custom logic to compute the concurrency key that can't be captured in a CEL expression, we recommend computing the key ahead of time and passing it as part of the input to the workflow. **Workflows registered against a v1 engine with a custom concurrency method (instead of an expression) will not use a concurrency queue.**
- Concurrency queues previously did not respect the `ScheduleTimeout` value set on the workflow level, so concurrency queues had no timeouts. In v1, concurrency queues will respect the schedule timeout value as well.
- User-defined events sent via `event.push` will no longer be displayed in the Hatchet UI and will not be available over the REST API. Events will still trigger workflows, but in nearly all instances of Hatchet that we surveyed, events were not used, accessed after writes, or replayed. We are considering adding an option to mark a user-defined event as `Durable`, but that will not be available in the initial v1 release. **Please let us know if you are dependent on the events view or APIs.**
frontend/docs/pages/home/migration-guide-python.mdx
+3 -1
View File
@@ -25,7 +25,7 @@ The API has changed significantly in the V1 SDK. Even in this simple example, th
1. Tasks can now be declared with `hatchet.task`, meaning you no longer _have_ to create a workflow explicitly to define a task. This should feel similar to how e.g. Celery handles task definition. Note that we recommend declaring a workflow in many cases, but the simplest possible way to get set up is to use `hatchet.task`.
2. Tasks have a new signature. They now take two arguments: `input` and `context`. The `input` is either of type `input_validator` (a Pydantic model you provide to the workflow), or is an `EmptyModel`, which is a helper Pydantic model Hatchet provides and uses as a default. The `context` is once again the Hatchet `Context` object.
3. Workflows can now be registered on a worker by using the `workflows` keyword argument to the `worker` method, although the old `register_workflows` method is still available.
4. `max_runs` on the worker has been renamed to `slots`.
4. The Hatchet client's REST API wrappers have been significantly reworked. For instance, if you'd like to use the REST API to list out workflow runs, you can use `await hatchet.workfows.aio_list()`
#### Pydantic
@@ -56,11 +56,13 @@ Typing improvements:
4. All external-facing types that are used for triggering workflows, scheduling workflows, etc. are now Pydantic objects, as opposed to being `TypedDict`s.
5. The return type of each `Task` is restricted to a `JSONSerializableMapping` or a Pydantic model, to better align with what the Hatchet Engine expects.
6. The `ClientConfig` now uses Pydantic Settings, and we've removed the static methods on the Client for `from_environment` and `from_config` in favor of passing configuration in correctly. See the [configuration example](./client.mdx) for more details.
7. The REST API wrappers, which previously were under `hatchet.rest`, have been completely overhauled.
Naming changes:
1. We no longer have nested `aio` clients for async methods. Instead, async methods throughout the entire SDK are preferred by `aio_`, similar to [Langchain's use of the `a` prefix](https://python.langchain.com/docs/concepts/streaming/#stream-and-astream) to indicate async. For example, to run a workflow, you may now either use `workflow.run()` or `workflow.aio_run()`.
2. All functions on Hatchet clients are now _verbs_. For instance `hatchet.admin.workflow_run_get` is now `hatchet.admin.get_workflow_run`.
3. `max_runs` on the worker has been renamed to `slots`.
Removals:
sdks/python/examples/concurrency_limit/test_concurrency_limit.py
+3 -1
View File
@@ -14,7 +14,9 @@ async def test_run(hatchet: Hatchet) -> None:
# Start all runs
for i in range(1, num_runs + 1):
run = concurrency_limit_workflow.run_no_wait(WorkflowInput(run=i, group=str(i)))
run = concurrency_limit_workflow.run_no_wait(
WorkflowInput(run=i, group_key=str(i))
)
runs.append(run)
# Wait for all results
sdks/python/examples/concurrency_limit/trigger.py
+1 -1
View File
@@ -1,3 +1,3 @@
from examples.concurrency_limit.worker import WorkflowInput, concurrency_limit_workflow
concurrency_limit_workflow.run(WorkflowInput(group="test", run=1))
concurrency_limit_workflow.run(WorkflowInput(group_key="test", run=1))
sdks/python/examples/concurrency_limit/worker.py
+5 -9
View File
@@ -13,25 +13,24 @@ from hatchet_sdk import (
hatchet = Hatchet(debug=True)
# ❓ Concurrency
# This workflow shows example usage of Hatchet's concurrency features
# ❓ Workflow
class WorkflowInput(BaseModel):
run: int
group: str
group_key: str
concurrency_limit_workflow = hatchet.workflow(
name="ConcurrencyDemoWorkflow",
concurrency=ConcurrencyExpression(
expression="input.group",
expression="input.group_key",
max_runs=5,
limit_strategy=ConcurrencyLimitStrategy.CANCEL_IN_PROGRESS,
),
input_validator=WorkflowInput,
)
# ‼️
@concurrency_limit_workflow.task()
def step1(input: WorkflowInput, ctx: Context) -> dict[str, Any]:
@@ -40,9 +39,6 @@ def step1(input: WorkflowInput, ctx: Context) -> dict[str, Any]:
return {"run": input.run}
# ‼️
def main() -> None:
worker = hatchet.worker(
"concurrency-demo-worker", slots=10, workflows=[concurrency_limit_workflow]
sdks/python/examples/concurrency_limit_rr/worker.py
+2
View File
@@ -12,6 +12,7 @@ from hatchet_sdk import (
hatchet = Hatchet(debug=True)
# ❓ Workflow
class WorkflowInput(BaseModel):
group: str
@@ -25,6 +26,7 @@ concurrency_limit_rr_workflow = hatchet.workflow(
),
input_validator=WorkflowInput,
)
# ‼️
@concurrency_limit_rr_workflow.task()
sdks/typescript/src/v1/examples/concurrency-rr/workflow.ts
+2
View File
@@ -17,6 +17,7 @@ const sleep = (ms: number) =>
setTimeout(resolve, ms);
});
// ❓ Workflow
export const simpleConcurrency = hatchet.workflow<SimpleInput, SimpleOutput>({
name: 'simple-concurrency',
concurrency: {
@@ -25,6 +26,7 @@ export const simpleConcurrency = hatchet.workflow<SimpleInput, SimpleOutput>({
expression: 'input.GroupKey',
},
});
// !!
simpleConcurrency.task({
name: 'to-lower',