0
# Variables Management
1

2
Prefect Variables are named, mutable JSON values that can be shared across tasks, flows, and deployments. They provide a centralized way to store and retrieve configuration values, secrets, and other data that needs to be accessed dynamically during workflow execution.
3

4
## Capabilities
5

6
### Variable Class
7

8
The main Variable class for creating, reading, updating, and managing variables.
9

10
```python
11
from prefect.variables import Variable
12
```
13

14
```python { .api }
15
class Variable(BaseModel):
16
    """Variables are named, mutable JSON values that can be shared across tasks and flows."""
17
    
18
    name: str                           # Variable name identifier
19
    value: StrictVariableValue         # Variable value (JSON-serializable)
20
    tags: Optional[List[str]]          # Optional tags for organization
21
    
22
    @classmethod
23
    def get(
24
        cls,
25
        name: str,
26
        default: StrictVariableValue = None,
27
    ) -> StrictVariableValue:
28
        """
29
        Get a variable's value by name.
30
        
31
        Args:
32
        - name: Name of the variable to retrieve
33
        - default: Default value if variable doesn't exist
34
        
35
        Returns:
36
        Variable value or default if not found
37
        """
38
    
39
    @classmethod
40
    async def aget(
41
        cls,
42
        name: str,
43
        default: StrictVariableValue = None,
44
    ) -> StrictVariableValue:
45
        """
46
        Asynchronously get a variable's value by name.
47
        
48
        Args:
49
        - name: Name of the variable to retrieve
50
        - default: Default value if variable doesn't exist
51
        
52
        Returns:
53
        Variable value or default if not found
54
        """
55
    
56
    @classmethod
57
    def set(
58
        cls,
59
        name: str,
60
        value: StrictVariableValue,
61
        tags: Optional[List[str]] = None,
62
        overwrite: bool = False,
63
    ) -> "Variable":
64
        """
65
        Set a new variable. Must pass overwrite=True if variable exists.
66
        
67
        Args:
68
        - name: Name of the variable
69
        - value: JSON-serializable value to store
70
        - tags: Optional tags for organization
71
        - overwrite: Whether to overwrite existing variable
72
        
73
        Returns:
74
        The newly created Variable object
75
        """
76
    
77
    @classmethod
78
    async def aset(
79
        cls,
80
        name: str,
81
        value: StrictVariableValue,
82
        tags: Optional[List[str]] = None,
83
        overwrite: bool = False,
84
    ) -> "Variable":
85
        """
86
        Asynchronously set a new variable. Must pass overwrite=True if variable exists.
87
        
88
        Args:
89
        - name: Name of the variable
90
        - value: JSON-serializable value to store
91
        - tags: Optional tags for organization
92
        - overwrite: Whether to overwrite existing variable
93
        
94
        Returns:
95
        The newly created Variable object
96
        """
97
```
98

99
## Usage Examples
100

101
### Basic Variable Operations
102

103
```python
104
from prefect.variables import Variable
105

106
# Set a variable
107
api_config = Variable.set(
108
    name="api_config",
109
    value={"base_url": "https://api.example.com", "timeout": 30},
110
    tags=["config", "api"]
111
)
112

113
# Get a variable value
114
config = Variable.get("api_config")
115
print(config)  # {"base_url": "https://api.example.com", "timeout": 30}
116

117
# Get with default value
118
debug_mode = Variable.get("debug_mode", default=False)
119

120
# Update an existing variable
121
Variable.set(
122
    name="api_config", 
123
    value={"base_url": "https://api.example.com", "timeout": 60},
124
    overwrite=True
125
)
126
```
127

128
### Using Variables in Flows and Tasks
129

130
```python
131
from prefect import flow, task
132
from prefect.variables import Variable
133

134
@task
135
def fetch_data():
136
    # Get API configuration from variables
137
    config = Variable.get("api_config")
138
    base_url = config["base_url"]
139
    timeout = config["timeout"]
140
    
141
    # Use configuration in task logic
142
    print(f"Fetching data from {base_url} with timeout {timeout}s")
143
    return {"data": "sample", "source": base_url}
144

145
@task
146
def process_data(raw_data: dict):
147
    # Get processing settings from variables
148
    batch_size = Variable.get("batch_size", default=100)
149
    
150
    print(f"Processing data in batches of {batch_size}")
151
    return {"processed": True, "batch_size": batch_size}
152

153
@flow
154
def data_pipeline():
155
    """Example flow using variables for configuration."""
156
    raw_data = fetch_data()
157
    result = process_data(raw_data)
158
    return result
159
```
160

161
### Async Variable Operations
162

163
```python
164
from prefect import flow, task
165
from prefect.variables import Variable
166
import asyncio
167

168
@task
169
async def async_fetch_data():
170
    # Asynchronously get variable value
171
    config = await Variable.aget("api_config")
172
    
173
    # Set a runtime variable asynchronously
174
    await Variable.aset(
175
        name="last_run_time",
176
        value=str(asyncio.get_event_loop().time()),
177
        overwrite=True
178
    )
179
    
180
    return {"data": "fetched", "config": config}
181

182
@flow
183
async def async_data_pipeline():
184
    """Example async flow using variables."""
185
    result = await async_fetch_data()
186
    return result
187
```
188

189
### Environment-Specific Configuration
190

191
```python
192
from prefect import flow
193
from prefect.variables import Variable
194
import prefect.runtime.deployment
195

196
@flow 
197
def environment_aware_flow():
198
    """Flow that adapts behavior based on environment variables."""
199
    
200
    # Get environment from deployment parameters or variables
201
    env = prefect.runtime.deployment.parameters.get("environment", "prod")
202
    
203
    # Get environment-specific configuration
204
    db_config = Variable.get(f"database_config_{env}")
205
    api_key = Variable.get(f"api_key_{env}")
206
    
207
    print(f"Running in {env} environment")
208
    print(f"Database config: {db_config}")
209
    
210
    return {"environment": env, "configured": True}
211
```
212

213
## Common Patterns
214

215
### Configuration Management
216

217
Variables are ideal for storing configuration that needs to be shared across multiple flows:
218

219
```python
220
# Set up shared configuration
221
Variable.set("app_config", {
222
    "database_url": "postgresql://localhost/mydb",
223
    "redis_url": "redis://localhost:6379",
224
    "log_level": "INFO"
225
})
226

227
Variable.set("feature_flags", {
228
    "enable_cache": True,
229
    "enable_notifications": False,
230
    "beta_features": True
231
})
232
```
233

234
### Dynamic Behavior Control
235

236
```python
237
@task
238
def conditional_task():
239
    # Use variables to control task behavior
240
    feature_flags = Variable.get("feature_flags", default={})
241
    
242
    if feature_flags.get("enable_cache", False):
243
        print("Using cached results")
244
        return "cached_result"
245
    else:
246
        print("Computing fresh results")
247
        return "fresh_result"
248
```
249

250
### Runtime State Management
251

252
```python
253
@task
254
def track_progress():
255
    # Update progress tracking variable
256
    current_progress = Variable.get("pipeline_progress", default=0)
257
    new_progress = current_progress + 1
258
    
259
    Variable.set("pipeline_progress", new_progress, overwrite=True)
260
    print(f"Pipeline progress: {new_progress}")
261
```
262

263
**Note:** Variables persist across flow runs and are accessible throughout your Prefect workspace. Use appropriate naming conventions and tags to organize variables effectively.