https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> I am wondering about prefect 3's even...
# ask-marvin
m
@Marvin I am wondering about prefect 3's events, automations and triggers. Let's say that I have 10 flows that each take 1 day to complete after generating 10k jobs each. I want flow 0 to run first, then flow 1 to run after 0 completed, etc. Each should get the most up-to-date parameters before being called. What is the best way of going about this. Using another flow to manage these 10 flows doesn't seems necessary?
m
✅ thought for 173.1 seconds
If each step must run strictly after the previous one and needs freshly computed parameters at launch time, I would not use automations as the primary mechanism. Best fit: use a small orchestrator flow that launches the 10 flows in order. Why: - Sequential ordering is trivial in a flow - Fresh parameters can be computed in Python right before launching the next flow - You can use the result/state of flow N to decide how to launch flow N+1 - Error handling and retries are much easier to reason about Automations/events/triggers are better when you want loosely coupled, event-driven behavior, not when you want a deterministic 10-step chain with runtime parameter calculation. Recommended pattern If these are separate deployed flows, have one orchestrator do: 1. get latest parameters/state 2. launch flow 0 and wait 3. get latest parameters/state again 4. launch flow 1 and wait 5. repeat That can look like this:
Copy code
python
from prefect import flow
from prefect.deployments import run_deployment

def get_fresh_parameters(step: int, previous_result=None):
    # Query DB / API / config service / whatever source of truth you use
    return {
        "step": step,
        "previous_result": previous_result,
    }

@flow
def orchestrate_chain():
    previous_result = None

    for i in range(10):
        params = get_fresh_parameters(i, previous_result)

        flow_run = run_deployment(
            name=f"my-flow-{i}/prod",
            parameters=params,
            timeout=60 * 60 * 24 * 2,  # wait up to 2 days, adjust as needed
        )

        # Depending on your pattern, inspect the state/result here
        previous_result = flow_run.state

    return "done"
A few notes: - 
run_deployment(...)
is the right primitive if these are separate deployments - This gives you a true chain with one place to manage ordering - The orchestrator itself can be very small — it doesn’t need to do the heavy work If the flows are just code units and don’t need to be independently deployed, another option is to make them subflows and call them directly from one parent flow. Docs: Flows Why not automations for this? Automations can absolutely trigger flow B when flow A completes. Docs: Automations and Events But they have drawbacks for your use case: - parameter passing is much more limited than plain Python orchestration - computing fresh parameters from external state just-in-time is awkward - chaining 10 automations becomes harder to debug - using outputs/results from the prior flow is not as natural - strict sequencing is less explicit than “wait, then launch next” So if your question is “do I really need another flow to manage these 10?” — for this case, yes, probably, and that’s actually the cleanest solution, not unnecessary overhead. Rule of thumb Use automations when: - a run should happen because an event occurred - flows are independent services - you want reactive, event-driven orchestration Use an orchestrator flow when: - order matters - next-step parameters depend on current runtime state - you want simple logic, retries, branching, and observability One more practical concern Since each flow may run for ~1 day and create ~10k jobs, think about whether the “10 flows” are really: - 10 stages in one pipeline, or - 10 independent workloads that just need serialization If they are pipeline stages, an orchestrator flow is definitely the natural model. If they are independent but must not overlap, you could also consider work pool limits or concurrency controls in addition to orchestration, depending on why they must run one-at-a-time. Relevant docs: - Work Pools - Concurrency limits
If you want, I can sketch the exact Prefect 3 implementation for either: - an orchestrator flow that waits for each deployment and passes fresh params, or - an automations-based chain so you can compare them side by side.
m
@Marvin by implementing your proposed solution the orchestrating workflow will not receive up-to-date parameters unless given from the very beginning. If a new model is released mid run, it won't have the knowledge to use it
m
✅ thought for 20.1 seconds
Right — if the orchestrator receives all params up front, then it would be stale. But that’s not how I’d structure it. The orchestrator should not carry all parameters from the start. It should fetch or compute the latest parameters immediately before launching each downstream flow. So the orchestrator is not the source of truth for parameters — your DB / API / config service / model registry is. Better pattern - Keep the latest model/version/config in an external source of truth - Before starting flow 
i
, the orchestrator queries that source - It launches flow 
i
with the current parameters at that moment - If a new model appears while flow 
0
is running for a day, then when flow 
1
is about to start, the orchestrator fetches again and sees the new model That gives you exactly the “most up-to-date at invocation time” behavior. Example:
Copy code
python
from prefect import flow
from prefect.deployments import run_deployment

def get_latest_runtime_config(step: int) -> dict:
    # Example: pull from database, API, S3, feature store, model registry, etc.
    # This is called RIGHT BEFORE each downstream flow starts
    return {
        "model_version": fetch_latest_model_version(),
        "step": step,
    }

@flow
def orchestrate_chain():
    previous_result = None

    for i in range(10):
        latest_params = get_latest_runtime_config(i)

        if previous_result is not None:
            latest_params["previous_result"] = previous_result

        flow_run = run_deployment(
            name=f"my-flow-{i}/prod",
            parameters=latest_params,
            timeout=60 * 60 * 48,
        )

        previous_result = flow_run.state

    return "done"
So in your example: - flow 0 starts with model 
v1
- 8 hours later, model 
v2
is released - flow 0 finishes - orchestrator is resumed - orchestrator calls 
fetch_latest_model_version()
- flow 1 starts with 
v2
That is still fully compatible with an orchestrator flow. Important distinction There are two very different designs: 1. Push parameters at chain creation time - stale for long-running chains - not what you want 2. Resolve parameters at step launch time - fresh - what you want You want #2. When events/automations might help If you want each flow to independently resolve its own runtime config, then automations can make sense only as the trigger, while the actual parameter freshness is handled inside the triggered flow itself. For example: - flow 0 completes - automation triggers flow 1 - flow 1 starts with minimal input - flow 1 itself queries the model registry / config store on startup That can work well if: - each flow is autonomous - each flow knows how to fetch its own latest config - you don’t need a central controller In that design, you often pass only stable identifiers, like:
Copy code
python
{
  "customer_id": 123,
  "pipeline_id": "abc"
}
and then inside the flow:
Copy code
python
@flow
def flow_1(customer_id: int, pipeline_id: str):
    runtime_config = fetch_latest_runtime_config(customer_id, pipeline_id)
    ...
So the real decision is this: Option A — Orchestrator flow - orchestrator decides when to launch next flow - orchestrator fetches fresh config at launch time Option B — Event-driven chain - automation decides when to launch next flow - downstream flow fetches fresh config on startup For your case, I’d still lean toward A unless you specifically want a loosely coupled event-driven architecture. My recommendation If the only concern is “new model released mid-run”, then do one of these: - Best overall: orchestrator fetches latest config right before each 
run_deployment
- Best decoupled design: each child flow fetches latest config on startup, and automation or orchestrator only passes stable identifiers In practice, the most robust approach is often: - orchestrator controls ordering - child flows fetch their own latest model/config at start That way:
- ordering is centralized - freshness is guaranteed at execution time - parameter payloads stay small - model updates are naturally picked up So yes — you can have up-to-date parameters with an orchestrator, as long as you don’t treat the orchestrator’s initial inputs as the full runtime config. If you want, I can show you a concrete Prefect 3 pattern for: 1. orchestrator fetches fresh config before each step, or 2. each child flow self-resolves latest config while the orchestrator only sequences them. The second one is probably the cleanest for your model-release example.
m
@Marvin So I would have to have this flow run for almost 2 weeks?
m
✅ thought for 16.0 seconds
Yes — if you use a single orchestrator flow that waits synchronously for 10 one-day runs, that parent flow could be alive for ~10–14 days. That’s usually not ideal. So for your case, I’d revise the recommendation: Don’t use one long-lived orchestrator flow if the chain spans many days. Instead, prefer one of these patterns: Option 1: Event-driven chaining with fresh config resolved inside each flow This is probably the best fit. How it works: - Flow 0 completes - An automation triggers Flow 1 - Flow 1, on startup, queries the latest model/config - Flow 1 completes - Another automation triggers Flow 2 - and so on That avoids having a parent run sitting around for two weeks. This works well if: - each flow can determine its own current parameters/config at start time - you only need sequencing, not complex branching logic - you want each flow run to be independent In this pattern, don’t try to pass all dynamic params through the automation. Instead: - pass stable identifiers only, if needed - let each flow fetch the newest model/config itself when it starts Example shape:
Copy code
python
from prefect import flow

@flow
def step_1(pipeline_id: str):
    config = fetch_latest_config(pipeline_id)
    model = fetch_latest_model(config)
    # do work

@flow
def step_2(pipeline_id: str):
    config = fetch_latest_config(pipeline_id)
    model = fetch_latest_model(config)
    # do work
Then configure automations so: - completion of 
step_0
triggers deployment of 
step_1
- completion of 
step_1
triggers deployment of 
step_2
- etc. Docs: - Automations - Events Option 2: External controller, not a Prefect flow If you want centralized sequencing but don’t want a 2-week Prefect run, you can use: - a small service - a cron job - a database-driven state machine - or another scheduler That controller can: - watch for flow completion - fetch latest config - launch the next deployment via the API This is good when sequencing logic is important but should live outside Prefect flow execution. Option 3: Single flow / subflows only if this is really one pipeline stage graph If these 10 “flows” are really just steps of one long pipeline, you could model them differently — but from what you described, they sound like independent long-running deployments, so I would not force them into one long parent run. My actual recommendation for your use case Given: - 10 flows - each can take ~1 day - you want strict sequencing - you want newest model/config at actual start time I’d use: - separate deployments for each step - automations to trigger the next step on completion - each flow fetches its own latest config/model at startup That gives you: - no 2-week parent run - latest model picked up naturally - clean separation between runs One caveat If you need very strict business logic like: - “only start step 4 if step 3 finished and produced result X and external condition Y still holds” - “skip step 6 if a newer model appeared” - “restart from step 7 with revised parameters” then automations alone can start getting awkward. In that case, an external stateful controller is often better than either: - a 2-week orchestrator flow, or - a large automation chain So the practical choice is: - Simple sequential chain + fresh startup config → use automations - Complex multi-day state machine → use an external controller/service If you want, I can show you how to model the automation chain in Prefect 3 for 
flow_0 -> flow_1 -> ... -> flow_9
, with each flow resolving the latest model at startup.
m
@Marvin isn't the automations solution what I proposed at the beginning (e.g. sequential chaining)?
m
✅ thought for 15.1 seconds
Yes — you were basically describing sequential chaining via events/automations from the start. And for the constraints you added later: - 10 long-running flows - each may run for ~1 day - each next flow should use the latest config/model at its own start time - you don’t want a 2-week controller flow sitting there your original instinct is the better fit So the refined answer is: - Yes, automations/events/triggers are a good solution here - But use them mainly for sequencing - And let each triggered flow resolve its own current parameters/config on startup That last part is the key. Why I initially pushed orchestration flow Because if you want: - strict ordering - easy branching - easy passing of previous outputs - centralized logic then a parent flow is usually the simplest mental model. But once you clarified: - runs are very long-lived - freshness matters at start time of each downstream flow - and you don’t want a long-lived parent that changes the recommendation. So yes: your original “flow 0 completes, then trigger flow 1, then trigger flow 2...” approach is the right direction. The important nuance What won’t work well is: - using automations to pass a giant, precomputed parameter bundle for the whole chain What will work well is: - automation only says “start the next deployment now” - the next flow looks up the latest model/config itself So the chain is event-driven, but parameter resolution is local and fresh. Good design for your case Each flow should receive only: - stable IDs - context keys - maybe the upstream run id if needed For example:
Copy code
python
@flow
def train_step(pipeline_id: str):
    config = load_current_pipeline_config(pipeline_id)
    model_info = load_latest_model_info(config)
    ...
Then the automation for 
flow_1
is basically: - when deployment/run for 
flow_0
completes successfully - run deployment 
flow_1
- pass 
pipeline_id
That means if a new model is released while 
flow_0
is running, 
flow_1
will pick it up when it starts. So to answer directly: Yes — the automation-based sequential chaining you proposed initially is likely the best architecture for this use case. I’d summarize it as: - Use automations for ordering - Use external state/config stores for freshness - Do not use a long-lived parent flow - Do not try to make automations carry all dynamic state If you want, I can give you a concrete Prefect 3 example of: - 2 chained deployments with an automation trigger, and - the downstream flow fetching fresh config at runtime and then you can replicate it for all 10.
m
If I need a bucket passed at a parameter from the beginning or such, how would I do that @Marvin
m
✅ thought for 14.5 seconds
Yes — pass the stable stuff up front. For values like: - bucket name - customer ID - dataset ID - environment - region - pipeline ID it’s perfectly fine to pass them as deployment parameters from the start and keep reusing them through the chain. Then each flow can also fetch dynamic values at runtime, like: - latest model version - current feature flags - active config revision So think of parameters in two categories: 1. Stable parameters Safe to pass through the chain: - 
bucket
- 
pipeline_id
- 
tenant_id
- 
region
2. Dynamic parameters Better resolved at flow start: - 
model_version
- 
config_revision
- 
latest_checkpoint
- 
active_ruleset
A good pattern is:
Copy code
python
from prefect import flow

@flow
def step_1(bucket: str, pipeline_id: str):
    runtime_config = fetch_latest_config(pipeline_id)
    model_version = runtime_config["model_version"]

    # use bucket + latest model_version
Then the automation that triggers 
step_2
passes along the same stable identifiers. Conceptually: - 
step_0(bucket="my-bucket", pipeline_id="abc")
- completion triggers 
step_1(bucket="my-bucket", pipeline_id="abc")
- completion triggers 
step_2(bucket="my-bucket", pipeline_id="abc")
while each step independently does:
Copy code
python
config = fetch_latest_config(pipeline_id)
How to pass stable parameters to the next deployment In Prefect automations, the 
RunDeployment
action can provide parameters to the triggered deployment. For a chain, you’d configure each downstream deployment with the stable values it needs. If the values are constant for that chain, common options are: Option A: bake them into the deployment If 
bucket
is always the same for that deployment/environment, just make it a default deployment parameter. Option B: pass them explicitly in the automation If 
bucket
varies by chain/run, set it in the automation-triggered deployment parameters. Option C: store chain context externally If you have many values to carry across 10 steps, store a record keyed by 
pipeline_id
, then each flow reads from it. For long multi-step chains, Option C is often the cleanest. For example:
Copy code
python
@flow
def step_3(pipeline_id: str):
    context = load_pipeline_context(pipeline_id)
    bucket = context["bucket"]
    latest_model = context["latest_model"]
That avoids having to thread many params through every automation. What I’d recommend If it’s just a few stable values like bucket + pipeline ID: - pass them as normal parameters If it’s a lot of shared context: - pass only 
pipeline_id
- store everything else in DB/object storage/config store - let each flow load what it needs Best practical design for your example Pass this kind of thing as parameters:
Copy code
python
{
  "pipeline_id": "abc123",
  "bucket": "my-bucket"
}
Then inside every flow:
Copy code
python
context = fetch_context(pipeline_id)
latest_model = context["latest_model"]
That gives you: - stable bucket available everywhere - dynamic model always fresh If you want, I can show you a Prefect 3 automation example where 
flow_0
completion triggers 
flow_1
and passes 
bucket
and 
pipeline_id
into the next run.
m
you keep showing me flows, I want to know how I would pass these through automations like you suggested @Marvin
m
✅ thought for 133.2 seconds
Yes — if you want to pass parameters through automations, the mechanism is the 
RunDeployment
action’s 
parameters
field, using Jinja templating to read values from the upstream 
flow_run
. So for stable params like 
bucket
or 
pipeline_id
, you can configure the downstream automation to copy them from the completed upstream run. Example shape:
Copy code
python
from prefect.automations import Automation, EventTrigger, RunDeployment, Posture
from prefect.events.schemas.events import ResourceSpecification

automation = Automation(
    name="flow-0 -> flow-1",
    trigger=EventTrigger(
        expect={"prefect.flow-run.Completed"},
        match_related=ResourceSpecification({
            "prefect.resource.role": "flow",
            "prefect.resource.name": "flow-0",
        }),
        posture=Posture.Reactive,
        threshold=1,
    ),
    actions=[
        RunDeployment(
            source="selected",
            deployment_id="<flow-1-deployment-id>",
            parameters={
                "bucket": {
                    "__prefect_kind": "jinja",
                    "template": "{{ flow_run.parameters['bucket'] }}",
                },
                "pipeline_id": {
                    "__prefect_kind": "jinja",
                    "template": "{{ flow_run.parameters['pipeline_id'] }}",
                },
            },
        )
    ],
)
That means: - 
flow-0
is started with 
bucket="my-bucket"
and 
pipeline_id="abc"
- when 
flow-0
completes - the automation triggers 
flow-1
- 
flow-1
gets those same values by reading them from `flow-0`’s parameters So yes, this is how you thread stable parameters through a chain. Important caveat about types Plain Jinja rendering can turn values into strings. If you need to preserve structured types, use the JSON hydration form. For strings like bucket names, this is fine:
Copy code
python
"bucket": {
    "__prefect_kind": "jinja",
    "template": "{{ flow_run.parameters['bucket'] }}",
}
For non-strings, use:
Copy code
python
"some_dict": {
    "__prefect_kind": "json",
    "value": {
        "__prefect_kind": "jinja",
        "template": "{{ flow_run.parameters['some_dict'] | tojson }}",
    },
}
What you can access in automation templates The upstream run is available as 
flow_run
, so common examples are:
Copy code
python
{{ flow_run.parameters['bucket'] }}
{{ flow_run.parameters['pipeline_id'] }}
{{ flow_run.id }}
You can also mix literal values with copied values:
Copy code
python
parameters={
    "bucket": {
        "__prefect_kind": "jinja",
        "template": "{{ flow_run.parameters['bucket'] }}",
    },
    "pipeline_id": {
        "__prefect_kind": "jinja",
        "template": "{{ flow_run.parameters['pipeline_id'] }}",
    },
    "environment": "prod",
}
How this fits your use case For your chain, a good split is: - pass 
bucket
, 
pipeline_id
, maybe 
tenant_id
through automations - do not pass highly dynamic things like 
latest_model_version
unless you really want to freeze them at trigger time - instead, let the downstream flow resolve those fresh at startup So the chain becomes: - start 
flow_0(bucket, pipeline_id)
- automation triggers 
flow_1(bucket, pipeline_id)
- automation triggers 
flow_2(bucket, pipeline_id)
- etc. with 
bucket
copied via:
Copy code
python
{{ flow_run.parameters['bucket'] }}
Docs / concepts - Automations - Events If you want, I can give you a full 2-step Prefect 3 example showing: - two deployments - the exact automation definition - and how 
bucket
/ 
pipeline_id
get passed from 
flow_0
to 
flow_1
.
m
@Marvin Alright, so let's say I have a lambda function or cron job polling a database, a new dataset is uploaded with a hashed value. I need to start the automation based on that hashed value and pass it to every step in the automation
m
✅ thought for 15.2 seconds
Yes — that’s a good fit for automations, if you introduce the hash as a stable parameter at the first run and then copy it forward in each 
RunDeployment
action. The pattern is: 1. your lambda / cron detects a new dataset 2. it starts flow 0 with parameters like: - 
dataset_hash
- maybe 
bucket
- maybe 
pipeline_id
3. when 
flow 0
completes, an automation triggers flow 1 4. the automation copies 
dataset_hash
from `flow 0`’s parameters into 
flow 1
5. repeat for all downstream steps So the hash becomes part of the chain context. What starts the chain? Your external poller would create the first flow run / deployment run for step 0. For example, conceptually:
Copy code
python
parameters = {
    "dataset_hash": discovered_hash,
    "bucket": "incoming-datasets",
    "pipeline_id": "dataset-processing",
}
Then every downstream automation copies those values from the triggering 
flow_run.parameters
. How to pass it through automations In each 
RunDeployment
action:
Copy code
python
parameters={
    "dataset_hash": {
        "__prefect_kind": "jinja",
        "template": "{{ flow_run.parameters['dataset_hash'] }}",
    },
    "bucket": {
        "__prefect_kind": "jinja",
        "template": "{{ flow_run.parameters['bucket'] }}",
    },
    "pipeline_id": {
        "__prefect_kind": "jinja",
        "template": "{{ flow_run.parameters['pipeline_id'] }}",
    },
}
That tells Prefect: - read these values from the completed upstream flow run - pass them into the next deployment So if 
flow_0
started with:
Copy code
python
{
  "dataset_hash": "abc123",
  "bucket": "incoming-datasets",
  "pipeline_id": "dataset-processing"
}
then 
flow_1
receives the same values, and then 
flow_2
, etc., as long as each automation copies them forward. Best design For your case, I’d treat 
dataset_hash
as a chain identifier. That means every step gets: - 
dataset_hash
- other stable identifiers and then each step can use that hash to: - query the DB for latest metadata - find the uploaded object - resolve current model/config - write step status back to the DB That is often cleaner than trying to pass a lot of context through each automation. So maybe every flow receives only:
Copy code
python
{
  "dataset_hash": "...",
}
and then does:
Copy code
python
record = load_dataset_record(dataset_hash)
bucket = record["bucket"]
latest_model = record["latest_model"]
...
That’s often the most robust design for long chains. Two good options Option A: pass the hash and a few stable params through every automation Good if the context is small. Option B: pass only the hash through every automation Best if the hash can be used to look up everything else. For most real systems, Option B is cleaner. Very important limitation Automations don’t maintain some built-in “workflow memory”. They only: - react to events - launch the next deployment - pass parameters you explicitly configure So if you want 
dataset_hash
at every step, you must either: - explicitly pass it in every automation, or - pass a single key like 
dataset_hash
and have each step look up the rest externally Recommended architecture for your case - Lambda/cron detects new dataset - Lambda launches 
step_0
deployment with 
dataset_hash
- Automation 
step_0 completed -> step_1
copies 
dataset_hash
- Automation 
step_1 completed -> step_2
copies 
dataset_hash
- ... - each step uses 
dataset_hash
to fetch the current record/config/state That gives you: - no long-lived orchestrator - a stable chain key - fresh runtime lookups at each step If you want, I can show you: 1. how the lambda should start the first deployment, and 2. the exact automation config for copying 
dataset_hash
to the next step.
m
@Marvin yes
m
✅ thought for 106.1 seconds
Yep — here’s the concrete pattern. 1. Start the first deployment from Lambda / cron If your poller finds a new dataset with hash 
abc123
, trigger the first deployment with that hash as a parameter. Using the CLI, the command exists as:
Copy code
bash
prefect deployment run <FLOW_NAME>/<DEPLOYMENT_NAME> --param key=value
Verified help: 
prefect deployment run --help
Example:
Copy code
bash
prefect deployment run flow-0/prod \
  --param dataset_hash="abc123" \
  --param bucket="incoming-datasets" \
  --param pipeline_id="dataset-processing"
You can do the same from Python:
Copy code
python
from prefect.deployments import run_deployment

run_deployment(
    name="flow-0/prod",
    parameters={
        "dataset_hash": "abc123",
        "bucket": "incoming-datasets",
        "pipeline_id": "dataset-processing",
    },
    timeout=0,  # return immediately
)
So your Lambda / cron is only responsible for starting step 0. 2. Pass those same parameters through the automation to step 1 This is the part you were asking about. In the automation’s 
RunDeployment
action, set 
parameters
with Jinja templates that read from the upstream 
flow_run.parameters
. Example:
Copy code
python
from datetime import timedelta
from prefect.automations import Automation, EventTrigger, RunDeployment, Posture
from prefect.events.schemas.events import ResourceSpecification

automation = Automation(
    name="flow-0 -> flow-1",
    trigger=EventTrigger(
        expect={"prefect.flow-run.Completed"},
        match_related=ResourceSpecification({
            "prefect.resource.role": "deployment",
            "prefect.resource.name": "flow-0",
        }),
        posture=Posture.Reactive,
        threshold=1,
        within=timedelta(seconds=0),
    ),
    actions=[
        RunDeployment(
            source="selected",
            deployment_id="<flow-1-deployment-id>",
            parameters={
                "dataset_hash": {
                    "__prefect_kind": "jinja",
                    "template": "{{ flow_run.parameters['dataset_hash'] }}",
                },
                "bucket": {
                    "__prefect_kind": "jinja",
                    "template": "{{ flow_run.parameters['bucket'] }}",
                },
                "pipeline_id": {
                    "__prefect_kind": "jinja",
                    "template": "{{ flow_run.parameters['pipeline_id'] }}",
                },
            },
        )
    ],
)
Then for 
flow-1 -> flow-2
, do the same thing again:
Copy code
python
parameters={
    "dataset_hash": {
        "__prefect_kind": "jinja",
        "template": "{{ flow_run.parameters['dataset_hash'] }}",
    },
    "bucket": {
        "__prefect_kind": "jinja",
        "template": "{{ flow_run.parameters['bucket'] }}",
    },
    "pipeline_id": {
        "__prefect_kind": "jinja",
        "template": "{{ flow_run.parameters['pipeline_id'] }}",
    },
}
So the values are copied forward from run to run. 3. Recommended design For your use case, I would usually pass: - 
dataset_hash
- 
bucket
- 
pipeline_id
through every automation, and let each flow use 
dataset_hash
to look up fresh metadata if needed. That gives you: - stable chain context passed forward via automation - fresh dynamic config loaded by each step at runtime 4. Important caveat: preserve types if needed For strings like 
dataset_hash
and 
bucket
, the simple Jinja form is fine. If you ever need to pass a dict/list/int and preserve the type, use the JSON wrapper:
Copy code
python
"payload": {
    "__prefect_kind": "json",
    "value": {
        "__prefect_kind": "jinja",
        "template": "{{ flow_run.parameters['payload'] | tojson }}",
    },
}
5. Alternative: start the chain from an event instead of directly running flow 0 If you want, your Lambda could emit a custom event instead, and an automation could start 
flow-0
using values from 
event.payload
. That would look like: ```python parameters={ "dataset_hash": {
"__prefect_kind": "jinja", "template": "{{ event.payload.dataset_hash }}", }, "bucket": { "__prefect_kind": "jinja", "template": "{{ event.payload.bucket }}", }, "pipeline_id": { "__prefect_kind": "jinja", "template": "{{ event.payload.pipeline_id }}", }, } ``` Then downstream automations use 
flow_run.parameters[...]
to keep forwarding them. That approach is nice if you want the whole thing to be fully event-driven. Summary Your end-to-end flow is: 1. Lambda/cron detects new dataset 2. it starts 
flow-0
with: - 
dataset_hash
- 
bucket
- 
pipeline_id
3. automation for 
flow-0 completed
starts 
flow-1
and copies those params 4. automation for 
flow-1 completed
starts 
flow-2
and copies those params 5. repeat through all steps Useful docs: - Automations - Events If you want, I can give you a full minimal working example with: - 
flow_0
- 
flow_1
- one automation - and the exact Lambda snippet to launch 
flow_0
.
2 Views
Open in Slack
PreviousNext
Powered by