<@ULVA73B9P> is there a way to explicitly set a fl...
# ask-marvin
d
@Marvin is there a way to explicitly set a flow run as a subflow of another flow?
m
thought for 83.2 seconds
Yes — but not by directly setting a
parent_flow_run_id
. In Prefect 3.x, a flow run is considered a subflow when it is created with a
parent_task_run_id
. Normally Prefect does this automatically when one flow calls another flow. What that means in practice: - Automatic path:
parent_flow()
calls
child_flow()
→ Prefect creates the subflow relationship for you - Manual path: if you're creating flow runs through the client API, you can pass
parent_task_run_id
when creating the child flow run There is not a separate “mark this existing flow run as a subflow” operation after creation. The relationship has to be established when the child flow run is created. Example APIs that support this: -
client.create_flow_run(..., parent_task_run_id=...)
-
client.create_flow_run_from_deployment(..., parent_task_run_id=...)
So the answer is: Yes, but only at flow-run creation time via `parent_task_run_id`; you can’t retroactively assign a parent/subflow relationship to an existing run. If you just want normal subflow behavior, the recommended approach is still simply calling one flow from another. If you want, I can also show: 1. the normal subflow pattern, or 2. a manual client/API example for creating a child flow run under a parent.
d
@Marvin If I now run my subflow from the class like self.result = self.flow_func( ... ), how can I switch it to keep the parameters passed and parent_task_run_id to make it explicitly the child flow as well?
m
thought for 132.6 seconds
If you're already doing:
Copy code
self.result = self.flow_func(...)
and that call happens inside another running flow, then you're already getting subflow behavior in Prefect 3.x. So in most cases, you do not need to switch anything: - parameters are preserved automatically - the parent/child relationship is created automatically - Prefect sets the
parent_task_run_id
for you behind the scenes So the recommended pattern is still just:
Copy code
self.result = self.flow_func(**params)
or for positional args:
Copy code
self.result = self.flow_func(arg1, arg2)
If
self.flow_func
is a
@flow
, and you're calling it from within another
@flow
, Prefect will register it as a child flow run. Example:
Copy code
from prefect import flow

class Runner:
    def __init__(self, flow_func):
        self.flow_func = flow_func
        self.result = None

    @flow
    def parent(self, x, y):
        self.result = self.flow_func(x=x, y=y)
        return self.result

@flow
def child(x, y):
    return x + y

runner = Runner(child)
runner.parent(1, 2)
That should already create: - one parent flow run for
parent
- one child/subflow run for
child
When would you need to do something manual? Only if you are not actually invoking the child flow from inside a running parent flow, and instead are creating flow runs via the client API. In that case, you’d need to: 1. get the current parent flow run context 2. create a task run in the parent flow 3. create the child flow run with
parent_task_run_id=<that task run id>
That looks more like this:
Copy code
from prefect import flow
from prefect.context import FlowRunContext
from prefect.client.orchestration import get_client
from prefect.states import Pending

@flow
async def parent():
    ctx = FlowRunContext.get()
    if not ctx:
        raise RuntimeError("Not running inside a flow")

    async with get_client() as client:
        parent_task_run = await client.create_task_run(
            task={"name": "child-flow"},
            flow_run_id=ctx.flow_run.id,
            state=Pending(),
        )

        child_run = await client.create_flow_run(
            flow=child,
            parameters={"x": 1, "y": 2},
            parent_task_run_id=parent_task_run.id,
            state=Pending(),
        )

        return child_run.id
But that is advanced/manual orchestration, and usually not what you want. Bottom line: - If
self.flow_func(...)
is a flow call inside another flow: leave it as-is - You do not need to manually preserve params or set
parent_task_run_id
- Manual
parent_task_run_id
is only for client-created flow runs, not normal flow-to-flow calls If you want, paste a small version of your class and I can tell you whether your current call is already creating a real subflow or whether you need the manual API route.