expose job variables in runtime

PrefectHQ · Dec 2, 2024 · 055cfa2 · 055cfa2
1 parent b52c2c9
commit 055cfa2
Show file tree

Hide file tree

Showing 2 changed files with 25 additions and 0 deletions.
diff --git a/src/prefect/runtime/flow_run.py b/src/prefect/runtime/flow_run.py
@@ -44,6 +44,7 @@
     "run_count",
     "api_url",
     "ui_url",
+    "job_variables",
 ]
 
 
@@ -317,6 +318,11 @@ def get_flow_run_ui_url() -> Optional[str]:
     return f"{PREFECT_UI_URL.value()}/flow-runs/flow-run/{flow_run_id}"
 
 
+def get_job_variables() -> Optional[Dict[str, Any]]:
+    flow_run_ctx = FlowRunContext.get()
+    return flow_run_ctx.flow_run.job_variables if flow_run_ctx else None
+
+
 FIELDS = {
     "id": get_id,
     "tags": get_tags,
@@ -331,4 +337,5 @@ def get_flow_run_ui_url() -> Optional[str]:
     "api_url": get_flow_run_api_url,
     "ui_url": get_flow_run_ui_url,
     "flow_version": get_flow_version,
+    "job_variables": get_job_variables,
 }
diff --git a/tests/runtime/test_flow_run.py b/tests/runtime/test_flow_run.py
@@ -645,3 +645,21 @@ async def test_url_pulls_from_api_when_needed(
         monkeypatch.setenv(name="PREFECT__FLOW_RUN_ID", value=str(run.id))
 
         assert getattr(flow_run, url_type) == expected_url
+
+
+class TestJobVariables:
+    async def test_job_variables_is_attribute(self):
+        assert "job_variables" in dir(flow_run)
+
+    async def test_job_variables_is_none_when_not_set(self):
+        assert flow_run.job_variables is None
+
+    async def test_job_variables_returns_variables_when_present_dynamically(self):
+        assert flow_run.job_variables is None
+
+        with FlowRunContext.model_construct(
+            flow_run=FlowRun.model_construct(job_variables={"foo": "bar"})
+        ):
+            assert flow_run.job_variables == {"foo": "bar"}
+
+        assert flow_run.job_variables is None