0
# Runtime Context Access
1
2
Prefect provides runtime context modules that allow flows and tasks to access information about their current execution environment. These modules provide dynamic access to deployment parameters, flow run metadata, and task run details during execution.
3
4
## Capabilities
5
6
### Deployment Runtime Context
7
8
Access current deployment information including parameters, ID, and version during flow execution.
9
10
```python
11
import prefect.runtime.deployment
12
```
13
14
```python { .api }
15
# Available attributes on prefect.runtime.deployment:
16
id: str # The deployment's unique ID
17
name: str # The deployment's name
18
version: str # The deployment's version
19
flow_run_id: str # The current flow run ID for this deployment
20
parameters: Dict[str, Any] # Parameters passed to this deployment run
21
```
22
23
**Usage Example:**
24
25
```python
26
from prefect import flow
27
import prefect.runtime.deployment
28
29
@flow
30
def my_flow():
31
# Access deployment information
32
deployment_id = prefect.runtime.deployment.id
33
params = prefect.runtime.deployment.parameters
34
version = prefect.runtime.deployment.version
35
36
print(f"Running deployment {deployment_id} version {version}")
37
print(f"Parameters: {params}")
38
```
39
40
### Flow Run Runtime Context
41
42
Access current flow run information including metadata, parameters, and execution details.
43
44
```python
45
import prefect.runtime.flow_run
46
```
47
48
```python { .api }
49
# Available attributes on prefect.runtime.flow_run:
50
id: str # The flow run's unique ID
51
tags: List[str] # The flow run's set of tags
52
scheduled_start_time: datetime # Expected scheduled start time
53
name: str # Name of the flow run
54
flow_name: str # Name of the flow
55
flow_version: str # Version of the flow
56
parameters: Dict[str, Any] # Parameters passed to this run
57
parent_flow_run_id: Optional[str] # ID of parent flow run (if subflow)
58
parent_deployment_id: Optional[str] # ID of parent deployment (if any)
59
root_flow_run_id: str # ID of root flow run in hierarchy
60
run_count: int # Number of times this flow run has been executed
61
```
62
63
**Usage Example:**
64
65
```python
66
from prefect import flow
67
import prefect.runtime.flow_run
68
69
@flow
70
def my_flow():
71
# Access flow run context
72
flow_run_id = prefect.runtime.flow_run.id
73
tags = prefect.runtime.flow_run.tags
74
parameters = prefect.runtime.flow_run.parameters
75
run_count = prefect.runtime.flow_run.run_count
76
77
print(f"Flow run {flow_run_id} (attempt {run_count})")
78
print(f"Tags: {tags}, Parameters: {parameters}")
79
```
80
81
### Task Run Runtime Context
82
83
Access current task run information and metadata during task execution.
84
85
```python
86
import prefect.runtime.task_run
87
```
88
89
```python { .api }
90
# Available attributes on prefect.runtime.task_run:
91
id: str # The task run's unique ID
92
name: str # Name of the task run
93
task_key: str # Unique key identifying the task
94
flow_run_id: str # ID of the parent flow run
95
task_inputs: Dict[str, Any] # Input parameters for the task
96
run_count: int # Number of times this task run has been executed
97
```
98
99
**Usage Example:**
100
101
```python
102
from prefect import task
103
import prefect.runtime.task_run
104
105
@task
106
def my_task():
107
# Access task run context
108
task_run_id = prefect.runtime.task_run.id
109
task_name = prefect.runtime.task_run.name
110
task_key = prefect.runtime.task_run.task_key
111
run_count = prefect.runtime.task_run.run_count
112
113
print(f"Task {task_name} ({task_key}) run {run_count}")
114
return f"Executed task run {task_run_id}"
115
```
116
117
## Common Patterns
118
119
### Conditional Logic Based on Runtime Context
120
121
```python
122
from prefect import flow, task
123
import prefect.runtime.deployment
124
import prefect.runtime.flow_run
125
126
@task
127
def process_data(data_source: str):
128
# Use deployment parameters to configure processing
129
config = prefect.runtime.deployment.parameters.get("processing_config", {})
130
131
if "debug" in prefect.runtime.flow_run.tags:
132
print(f"Debug mode: processing {data_source} with config {config}")
133
134
return f"Processed {data_source}"
135
136
@flow
137
def data_pipeline():
138
# Access runtime context for dynamic behavior
139
env = prefect.runtime.deployment.parameters.get("environment", "prod")
140
data_source = f"s3://bucket/{env}/data.csv"
141
142
return process_data(data_source)
143
```
144
145
### Testing with Mock Runtime Context
146
147
Runtime attributes can be mocked using environment variables:
148
149
```python
150
import os
151
152
# Mock deployment context
153
os.environ["PREFECT__RUNTIME__DEPLOYMENT__ID"] = "test-deployment-id"
154
os.environ["PREFECT__RUNTIME__DEPLOYMENT__PARAMETERS"] = '{"env": "test"}'
155
156
# Mock flow run context
157
os.environ["PREFECT__RUNTIME__FLOW_RUN__ID"] = "test-flow-run-id"
158
os.environ["PREFECT__RUNTIME__FLOW_RUN__TAGS"] = '["test", "debug"]'
159
```
160
161
**Note:** Runtime context modules return empty values when not executing within a Prefect flow run context.