0
# Configuration
1
2
Prefect's configuration system provides comprehensive management of settings, blocks for reusable configurations, and variables for dynamic workflow parameterization. This enables secure credential management, environment-specific configurations, and flexible workflow customization.
3
4
## Capabilities
5
6
### Blocks
7
8
Reusable configuration objects for credentials, connections, and infrastructure settings that can be shared across flows and deployments.
9
10
```python { .api }
11
class Block(BaseModel):
12
"""
13
Base class for Prefect blocks - reusable configuration objects.
14
15
Blocks provide a way to store and share configuration, credentials,
16
and other reusable components across flows and deployments. They
17
support serialization, validation, and secure storage.
18
"""
19
20
_block_type_name: ClassVar[str]
21
_block_type_slug: ClassVar[str]
22
_logo_url: ClassVar[Optional[str]]
23
_description: ClassVar[Optional[str]]
24
_documentation_url: ClassVar[Optional[str]]
25
26
def save(
27
self,
28
name: str,
29
overwrite: bool = False,
30
) -> UUID:
31
"""
32
Save the block to Prefect storage.
33
34
Parameters:
35
- name: Name to save the block under
36
- overwrite: Whether to overwrite existing block with same name
37
38
Returns:
39
UUID of the saved block
40
41
Raises:
42
ValueError: If block with same name exists and overwrite=False
43
"""
44
45
@classmethod
46
def load(cls, name: str) -> "Block":
47
"""
48
Load a saved block by name.
49
50
Parameters:
51
- name: Name of the block to load
52
53
Returns:
54
Block instance loaded from storage
55
56
Raises:
57
ValueError: If block with given name doesn't exist
58
"""
59
60
@classmethod
61
def register_type_and_schema(cls) -> None:
62
"""Register the block type and schema with Prefect."""
63
64
def dict(self, **kwargs) -> Dict[str, Any]:
65
"""Convert block to dictionary representation."""
66
67
def json(self, **kwargs) -> str:
68
"""Convert block to JSON representation."""
69
70
@classmethod
71
def schema(cls, **kwargs) -> Dict[str, Any]:
72
"""Get the JSON schema for the block type."""
73
```
74
75
#### Usage Examples
76
77
```python
78
from prefect.blocks.core import Block
79
from pydantic import Field
80
from typing import Optional
81
82
# Custom block definition
83
class DatabaseConfig(Block):
84
"""Database connection configuration block."""
85
86
_block_type_name = "Database Config"
87
_block_type_slug = "database-config"
88
_description = "Configuration for database connections"
89
90
host: str = Field(description="Database host")
91
port: int = Field(default=5432, description="Database port")
92
database: str = Field(description="Database name")
93
username: str = Field(description="Database username")
94
password: str = Field(description="Database password")
95
ssl_mode: Optional[str] = Field(default="prefer", description="SSL mode")
96
97
def get_connection_string(self) -> str:
98
"""Generate connection string from configuration."""
99
return f"postgresql://{self.username}:{self.password}@{self.host}:{self.port}/{self.database}?sslmode={self.ssl_mode}"
100
101
# Save and load blocks
102
def setup_database_config():
103
# Create block instance
104
db_config = DatabaseConfig(
105
host="localhost",
106
port=5432,
107
database="myapp",
108
username="user",
109
password="secret"
110
)
111
112
# Save for reuse
113
block_id = db_config.save(name="production-db")
114
115
return block_id
116
117
# Use in flows
118
from prefect import flow, task
119
120
@task
121
def connect_to_database():
122
# Load the saved block
123
db_config = DatabaseConfig.load("production-db")
124
connection_string = db_config.get_connection_string()
125
126
# Use configuration
127
print(f"Connecting to: {connection_string}")
128
return "connected"
129
130
@flow
131
def database_workflow():
132
connection = connect_to_database()
133
return connection
134
```
135
136
### Settings Management
137
138
Access and management of Prefect settings and configuration profiles.
139
140
```python { .api }
141
def get_settings_context() -> SettingsContext:
142
"""
143
Get the current settings context.
144
145
Returns:
146
SettingsContext containing current profile and settings
147
"""
148
149
class SettingsContext:
150
"""
151
Context object containing Prefect settings and profile information.
152
153
Attributes:
154
- profile: Current configuration profile
155
- settings: Dictionary of current settings
156
"""
157
158
profile: Profile
159
settings: Dict[str, Any]
160
161
def update_settings(self, **kwargs) -> None:
162
"""Update settings in the current context."""
163
164
def get_setting(self, key: str, default: Any = None) -> Any:
165
"""Get a specific setting value."""
166
167
class Profile:
168
"""
169
Configuration profile containing Prefect settings.
170
171
Attributes:
172
- name: Profile name
173
- settings: Profile-specific settings
174
"""
175
176
name: str
177
settings: Dict[str, Any]
178
179
@classmethod
180
def load_from_file(cls, path: str) -> "Profile":
181
"""Load profile from configuration file."""
182
183
def save_to_file(self, path: str) -> None:
184
"""Save profile to configuration file."""
185
```
186
187
#### Usage Examples
188
189
```python
190
from prefect.context import get_settings_context
191
from prefect import flow, task
192
193
@task
194
def environment_aware_task():
195
# Access current settings
196
context = get_settings_context()
197
198
print(f"Current profile: {context.profile.name}")
199
print(f"API URL: {context.get_setting('PREFECT_API_URL')}")
200
201
# Conditional logic based on environment
202
if context.profile.name == "production":
203
return "production-mode-result"
204
else:
205
return "development-mode-result"
206
207
@flow
208
def settings_aware_flow():
209
context = get_settings_context()
210
211
# Log current configuration
212
print(f"Running with profile: {context.profile.name}")
213
214
result = environment_aware_task()
215
return result
216
```
217
218
### Variables
219
220
Dynamic variables for runtime parameterization and configuration management.
221
222
```python { .api }
223
class Variable:
224
"""
225
Prefect variable for dynamic configuration.
226
227
Variables provide a way to store and retrieve dynamic values
228
that can be updated without changing flow code.
229
"""
230
231
def __init__(
232
self,
233
name: str,
234
value: Any = None,
235
tags: List[str] = None,
236
):
237
"""
238
Initialize a variable.
239
240
Parameters:
241
- name: Variable name
242
- value: Variable value
243
- tags: Optional tags for organization
244
"""
245
246
def save(self, overwrite: bool = True) -> UUID:
247
"""
248
Save the variable to Prefect storage.
249
250
Parameters:
251
- overwrite: Whether to overwrite existing variable
252
253
Returns:
254
UUID of the saved variable
255
"""
256
257
@classmethod
258
def get(
259
cls,
260
name: str,
261
default: Any = None,
262
) -> Any:
263
"""
264
Get a variable value by name.
265
266
Parameters:
267
- name: Variable name to retrieve
268
- default: Default value if variable doesn't exist
269
270
Returns:
271
Variable value or default
272
"""
273
274
@classmethod
275
def set(
276
cls,
277
name: str,
278
value: Any,
279
tags: List[str] = None,
280
overwrite: bool = True,
281
) -> UUID:
282
"""
283
Set a variable value.
284
285
Parameters:
286
- name: Variable name
287
- value: Variable value
288
- tags: Optional tags
289
- overwrite: Whether to overwrite existing variable
290
291
Returns:
292
UUID of the variable
293
"""
294
295
@classmethod
296
def delete(cls, name: str) -> bool:
297
"""
298
Delete a variable.
299
300
Parameters:
301
- name: Variable name to delete
302
303
Returns:
304
True if variable was deleted
305
"""
306
```
307
308
#### Usage Examples
309
310
```python
311
from prefect.variables import Variable
312
from prefect import flow, task
313
314
@task
315
def configurable_task():
316
# Get dynamic configuration
317
batch_size = Variable.get("batch_size", default=100)
318
max_retries = Variable.get("max_retries", default=3)
319
320
print(f"Processing with batch_size={batch_size}, max_retries={max_retries}")
321
322
return {"batch_size": batch_size, "max_retries": max_retries}
323
324
@flow
325
def dynamic_workflow():
326
# Set runtime variables
327
Variable.set("batch_size", 500, tags=["processing"])
328
Variable.set("max_retries", 5, tags=["reliability"])
329
330
# Use variables in task
331
config = configurable_task()
332
333
return config
334
335
# Variable management
336
def setup_environment_variables():
337
# Production settings
338
Variable.set("api_endpoint", "https://api.production.com", tags=["production", "api"])
339
Variable.set("timeout_seconds", 30, tags=["production", "performance"])
340
Variable.set("log_level", "INFO", tags=["production", "logging"])
341
342
# Development settings
343
Variable.set("debug_mode", True, tags=["development"])
344
Variable.set("mock_external_apis", True, tags=["development", "testing"])
345
```
346
347
### Built-in Block Types
348
349
Prefect provides several built-in block types for common infrastructure and service integrations.
350
351
```python { .api }
352
# System blocks
353
class Secret(Block):
354
"""Secure storage for sensitive values."""
355
value: str = Field(description="Secret value")
356
357
class JSON(Block):
358
"""Storage for JSON data."""
359
value: Dict[str, Any] = Field(description="JSON data")
360
361
class String(Block):
362
"""Storage for string values."""
363
value: str = Field(description="String value")
364
365
# Notification blocks
366
class SlackWebhook(Block):
367
"""Slack webhook integration."""
368
url: str = Field(description="Slack webhook URL")
369
370
def notify(self, message: str) -> None:
371
"""Send notification to Slack."""
372
373
class EmailServer(Block):
374
"""Email server configuration."""
375
smtp_server: str = Field(description="SMTP server host")
376
smtp_port: int = Field(default=587, description="SMTP port")
377
username: str = Field(description="SMTP username")
378
password: str = Field(description="SMTP password")
379
380
def send_email(
381
self,
382
to: List[str],
383
subject: str,
384
body: str,
385
) -> None:
386
"""Send email notification."""
387
388
# Infrastructure blocks
389
class DockerContainer(Block):
390
"""Docker container configuration."""
391
image: str = Field(description="Container image")
392
command: Optional[List[str]] = Field(description="Container command")
393
env: Optional[Dict[str, str]] = Field(description="Environment variables")
394
395
class KubernetesJob(Block):
396
"""Kubernetes job configuration."""
397
namespace: str = Field(default="default", description="Kubernetes namespace")
398
image: str = Field(description="Container image")
399
job_manifest: Dict[str, Any] = Field(description="Kubernetes job manifest")
400
```
401
402
#### Usage Examples
403
404
```python
405
from prefect.blocks.system import Secret, JSON
406
from prefect.blocks.notifications import SlackWebhook
407
from prefect import flow, task
408
409
# Setup blocks
410
def setup_infrastructure_blocks():
411
# Save API key securely
412
api_secret = Secret(value="sk-1234567890abcdef")
413
api_secret.save("openai-api-key")
414
415
# Save configuration
416
config = JSON(value={
417
"model": "gpt-4",
418
"temperature": 0.7,
419
"max_tokens": 1000
420
})
421
config.save("llm-config")
422
423
# Setup notifications
424
slack = SlackWebhook(url="https://hooks.slack.com/services/...")
425
slack.save("team-notifications")
426
427
@task
428
def ai_task():
429
# Load secure credentials
430
api_key = Secret.load("openai-api-key").value
431
config = JSON.load("llm-config").value
432
433
# Use configuration
434
print(f"Using model: {config['model']}")
435
# API call logic here
436
437
return "AI processing complete"
438
439
@flow
440
def ai_workflow():
441
result = ai_task()
442
443
# Send notification
444
slack = SlackWebhook.load("team-notifications")
445
slack.notify(f"Workflow completed: {result}")
446
447
return result
448
```
449
450
### Environment Configuration
451
452
Utilities for managing environment-specific configurations and deployment settings.
453
454
```python { .api }
455
def get_current_profile() -> Profile:
456
"""Get the currently active configuration profile."""
457
458
def set_current_profile(profile_name: str) -> None:
459
"""Set the active configuration profile."""
460
461
def list_profiles() -> List[str]:
462
"""List all available configuration profiles."""
463
464
def create_profile(
465
name: str,
466
settings: Dict[str, Any],
467
activate: bool = False,
468
) -> Profile:
469
"""
470
Create a new configuration profile.
471
472
Parameters:
473
- name: Profile name
474
- settings: Profile settings dictionary
475
- activate: Whether to activate the new profile
476
477
Returns:
478
Created Profile object
479
"""
480
481
def delete_profile(name: str) -> bool:
482
"""
483
Delete a configuration profile.
484
485
Parameters:
486
- name: Profile name to delete
487
488
Returns:
489
True if profile was deleted
490
"""
491
```
492
493
#### Usage Examples
494
495
```python
496
from prefect.profiles import (
497
get_current_profile,
498
create_profile,
499
set_current_profile,
500
list_profiles
501
)
502
503
# Profile management
504
def setup_environments():
505
# Create development profile
506
dev_profile = create_profile(
507
name="development",
508
settings={
509
"PREFECT_API_URL": "http://localhost:4200/api",
510
"PREFECT_LOGGING_LEVEL": "DEBUG",
511
"PREFECT_LOCAL_STORAGE_PATH": "./dev-storage"
512
}
513
)
514
515
# Create production profile
516
prod_profile = create_profile(
517
name="production",
518
settings={
519
"PREFECT_API_URL": "https://api.prefect.cloud/api/accounts/.../workspaces/...",
520
"PREFECT_API_KEY": "pnu_...",
521
"PREFECT_LOGGING_LEVEL": "INFO"
522
}
523
)
524
525
return dev_profile, prod_profile
526
527
def switch_environment(env_name: str):
528
"""Switch to specified environment profile."""
529
available = list_profiles()
530
531
if env_name in available:
532
set_current_profile(env_name)
533
current = get_current_profile()
534
print(f"Switched to profile: {current.name}")
535
else:
536
print(f"Profile '{env_name}' not found. Available: {available}")
537
```
538
539
## Types
540
541
Types related to configuration and blocks:
542
543
```python { .api }
544
from typing import Any, Dict, List, Optional, ClassVar, Union
545
from uuid import UUID
546
from pydantic import BaseModel, Field
547
from datetime import datetime
548
549
class BlockDocument:
550
"""Document representing a saved block."""
551
id: UUID
552
name: str
553
block_type_id: UUID
554
block_schema_id: UUID
555
data: Dict[str, Any]
556
created: datetime
557
updated: datetime
558
559
class BlockType:
560
"""Block type registration information."""
561
id: UUID
562
name: str
563
slug: str
564
logo_url: Optional[str]
565
documentation_url: Optional[str]
566
description: Optional[str]
567
code_example: Optional[str]
568
569
class BlockSchema:
570
"""JSON schema for a block type."""
571
id: UUID
572
checksum: str
573
fields: Dict[str, Any]
574
block_type_id: UUID
575
version: str
576
577
class VariableDocument:
578
"""Document representing a saved variable."""
579
id: UUID
580
name: str
581
value: Any
582
tags: List[str]
583
created: datetime
584
updated: datetime
585
586
# Configuration types
587
class ProfileSettings:
588
"""Settings within a configuration profile."""
589
590
PREFECT_API_URL: Optional[str]
591
PREFECT_API_KEY: Optional[str]
592
PREFECT_LOGGING_LEVEL: Optional[str]
593
PREFECT_LOCAL_STORAGE_PATH: Optional[str]
594
# ... many other settings
595
596
class DeploymentConfiguration:
597
"""Configuration for deployment environments."""
598
work_pool_name: str
599
job_variables: Dict[str, Any]
600
build_steps: List[Dict[str, Any]]
601
push_steps: List[Dict[str, Any]]
602
pull_steps: List[Dict[str, Any]]
603
```