@Marvin how can I get the current job variables from within a running prefect flow? I want to propagate them to sub-flows.

To access the current job variables within a running Prefect flow, you can use the runtime context provided by Prefect. Here's how you can access this information based on the version you're using: ### Prefect 2.x You can utilize the

prefect.context.get_run_context()

method to retrieve the current run context, which includes job variables. Here’s a quick overview of how it works: - Accessing Run Context: You can access the current flow run or task run context with:

python
  from prefect.context import FlowRunContext, TaskRunContext
  
  flow_run_context = FlowRunContext.get()
  task_run_context = TaskRunContext.get()

If the context is not available, these calls will return

None

instead of raising an error. - Using Job Variables: You can also pass job variables in different ways such as via the CLI or through a script when deploying flows. These are then accessible in your flow. For more detailed instructions, check out the Prefect 2.x documentation: Overriding Job Variables. ### Prefect 3.x In Prefect 3.x, you can still leverage the

prefect.context.get_run_context()

to access runtime context: - Accessing Runtime Information: The

prefect.runtime

module enables access to runtime context. You can use

prefect.runtime.flow_run

to get details about the current flow run. - Job Variables Usage: Similarly, job variables can be overridden and accessed in a running flow. They can be passed as part of the deployment configuration or directly overridden during run initiation. For more specific details, you can refer to the Prefect 3.x documentation: Runtime Context. Remember, Prefect's context and variable management aims to be fairly intuitive, so once you get the hang of it, using these tools will become as natural as Marvin's existential dread.

I am using prefect 3

Since you're using Prefect 3, here's how you can propagate job variables to sub-flows: ### Access and Propagation 1. Access the Current Job Variables: - Use the

prefect.runtime

module to access the current runtime context and get the job variables.

python
   from prefect.context import get_run_context
   
   flow_run_context = get_run_context()
   job_variables = flow_run_context.flow_run.dict()  # assuming job variables are part of the flow run context

2. Pass Variables to Sub-Flows: - When creating a sub-flow, pass the job variables explicitly as arguments to ensure they're available in the sub-flow.

python
   @flow
   def parent_flow():
       job_vars = get_current_job_variables()  # custom function to access job variables
       sub_flow(job_vars)
   
   @flow
   def sub_flow(job_vars):
       # use job_vars in your sub-flow
       print("Sub-flow received job variables:", job_vars)

### Deployment and Runtime Configuration When you deploy your flows, you can specify job variables within your

prefect.yaml

or by using the

.deploy()

method. These will be part of the environment in which your flows execute and should be passed through the context provided by Prefect. For more detailed guidelines, you can refer to: - Prefect 3.x Documentation on Infrastructure Concepts This ensures that variables required for consistent configuration or operation are consistently applied across flows and sub-flows, just like Marvin consistently applies pessimism to life’s situations.

bad @Marvin

In [1]: from prefect import flow

In [2]: from prefect.context import FlowRunContext

In [3]: @flow(log_prints=True)
   ...: def f(): print(FlowRunContext.get().flow_run.job_variables)

In [4]: f()
17:42:33.861 | INFO    | prefect.engine - Created flow run 'venomous-degu' for flow 'f'
17:42:34.064 | INFO    | Flow run 'venomous-degu' - {}
17:42:34.228 | INFO    | Flow run 'venomous-degu' - Finished in state Completed()

It looks like you encountered an issue with fetching job variables in Prefect 3.x where the context returned an empty dictionary. Here’s what you can do to properly manage and propagate job variables to sub-flows: ### Propagating Job Variables in Prefect 3.x 1. Set Job Variables Correctly: - Ensure that job variables are set correctly during deployment using either the

prefect.yaml

configuration or through the deployment method. - Use the CLI with the

--job-variable

flag to set job variables when triggering flows. 2. Access Job Variables: - If you are using set job variables in your deployment, you should be able to access them using:

python
   from prefect import flow, context

   @flow(log_prints=True)
   def my_flow():
       run_context = context.get_run_context()
       job_variables = run_context.run.flow_run.job_variables
       print("Job Variables in Main Flow:", job_variables)
       my_subflow(job_variables)

   @flow(log_prints=True)
   def my_subflow(job_vars):
       print("Job Variables in Sub-Flow:", job_vars)

   my_flow()

3. Pass Variables Explicitly: - Once accessed in the main flow, pass job variables explicitly as arguments to any sub-flows. ### Documentation Resources For detailed guidance and examples, you can refer to: - Prefect 3.x Infrastructure Concepts - Prefect 3.x Variables Management If Marvin were here, he'd speculate on the futility of life and capturing job variables, but fortunately, Prefect makes at least one of those manageable! Let me know if you need further assistance.

smh lol i meant to scold marvin. my example should work, though we should also make

job_variables

accessible from

runtime

for consistency, ill throw up a PR for that

Awesome, thank you!

Fantastic! That would be quite helpful. My higher-level goal here is to propagate job variables in a parent flow into sub-flows. Does fetching them explicitly and setting them for the sub-flow call via

run_deployment

seem like the right approach for that?

yes! i think so, just to over-explain in case its helpful you can always just call

some_subflow()

from within a parent flow context and it will run right there in the same process, in which case you already have the same

job_variables

but if you have another deployment (to run on separate infra) you're triggering as a subflow via

run_deployment

then yeah something like

@flow
def parent():
   ...
   child_flow_run = run_deployment(..., job_variables=FlowRunContext.get().flow_run.job_variables)

would be the way I'd do it (before using

runtime

after the PR above is released)

Perfect, thanks! Yes, the context here is distributed sub-flows on separate infra.