0
# Prefect
1

2
Prefect is a workflow orchestration and management framework for building resilient data pipelines in Python. It provides a modern approach to workflow management with features like automatic retries, dynamic mapping, caching, and real-time observability, enabling robust and scalable data workflows.
3

4
## Package Information
5

6
- **Package Name**: prefect
7
- **Language**: Python
8
- **Installation**: `pip install prefect`
9

10
## Core Imports
11

12
```python
13
import prefect
14
from prefect import flow, task, get_run_logger, get_client, State
15
```
16

17
Common imports for workflow development:
18

19
```python
20
from prefect import flow, task
21
from prefect.tasks import task_input_hash
22
from prefect.deployments import deploy
23
from prefect.client.orchestration import get_client
24
from prefect.states import Completed, Failed, Running
25
from prefect.context import TaskRunContext
26
```
27

28
## Basic Usage
29

30
```python
31
from prefect import flow, task, get_run_logger
32

33
@task
34
def extract_data(url: str):
35
    """Extract data from a source."""
36
    logger = get_run_logger()
37
    logger.info(f"Extracting data from {url}")
38
    # Simulate data extraction
39
    return {"records": 100, "source": url}
40

41
@task
42
def transform_data(raw_data: dict):
43
    """Transform the extracted data."""
44
    logger = get_run_logger()
45
    logger.info("Transforming data")
46
    return {
47
        "processed_records": raw_data["records"],
48
        "status": "transformed"
49
    }
50

51
@task
52
def load_data(processed_data: dict):
53
    """Load data to destination."""
54
    logger = get_run_logger()
55
    logger.info(f"Loading {processed_data['processed_records']} records")
56
    return {"loaded": True}
57

58
@flow(name="ETL Pipeline")
59
def etl_flow(source_url: str):
60
    """Complete ETL workflow."""
61
    raw = extract_data(source_url)
62
    processed = transform_data(raw)
63
    result = load_data(processed)
64
    return result
65

66
if __name__ == "__main__":
67
    # Run the flow
68
    result = etl_flow("https://api.example.com/data")
69
    print(result)
70
```
71

72
## Architecture
73

74
Prefect's architecture is built around several key concepts:
75

76
- **Flows**: Top-level workflow containers that define the execution logic and dependencies between tasks
77
- **Tasks**: Individual units of work that can be cached, retried, and run in parallel
78
- **States**: Immutable snapshots representing the current status of flows and tasks (Pending, Running, Completed, Failed, etc.)
79
- **Deployments**: Infrastructure-aware flow configurations that enable scheduled and triggered execution
80
- **Work Pools**: Infrastructure abstraction layer for executing flows across different environments
81
- **Blocks**: Reusable configuration objects for credentials, connections, and infrastructure settings
82

83
This design enables building resilient workflows with automatic error handling, observability, and scalable execution across diverse infrastructure environments.
84

85
## Capabilities
86

87
### Core Workflows
88

89
Flow and task creation with decorators, execution control, serving deployments, and flow lifecycle management.
90

91
```python { .api }
92
def flow(
93
    fn=None,
94
    *,
95
    name: str = None,
96
    description: str = None,
97
    version: str = None,
98
    flow_run_name: str = None,
99
    task_runner: TaskRunner = None,
100
    timeout_seconds: Union[int, float] = None,
101
    validate_parameters: bool = True,
102
    retries: int = None,
103
    retry_delay_seconds: Union[int, float] = None,
104
    persist_result: bool = None,
105
    result_storage: ResultStorage = None,
106
    result_serializer: ResultSerializer = None,
107
    cache_result_in_memory: bool = True,
108
    on_completion: List[FlowStateHook] = None,
109
    on_failure: List[FlowStateHook] = None,
110
    on_cancellation: List[FlowStateHook] = None,
111
    on_crashed: List[FlowStateHook] = None,
112
    on_running: List[FlowStateHook] = None,
113
    log_prints: bool = None,
114
): ...
115

116
def task(
117
    fn=None,
118
    *,
119
    name: str = None,
120
    description: str = None,
121
    tags: Iterable[str] = None,
122
    version: str = None,
123
    cache_key_fn: Callable[[TaskRunContext, Dict[str, Any]], Optional[str]] = None,
124
    cache_expiration: datetime.timedelta = None,
125
    task_run_name: Union[str, Callable[[], str]] = None,
126
    retries: int = None,
127
    retry_delay_seconds: Union[int, float] = None,
128
    retry_condition_fn: RetryConditionCallable = None,
129
    persist_result: bool = None,
130
    result_storage: ResultStorage = None,
131
    result_serializer: ResultSerializer = None,
132
    cache_result_in_memory: bool = True,
133
    timeout_seconds: Union[int, float] = None,
134
    log_prints: bool = None,
135
    refresh_cache: bool = None,
136
    on_completion: List[StateHookCallable] = None,
137
    on_failure: List[StateHookCallable] = None,
138
): ...
139

140
def serve(*flows: Flow, **kwargs) -> None: ...
141
def aserve(*flows: Flow, **kwargs) -> None: ...
142
```
143

144
[Core Workflows](./core-workflows.md)
145

146
### State Management
147

148
State creation functions, state utilities, and flow run lifecycle control.
149

150
```python { .api }
151
def Completed(cls: type = State, **kwargs: Any) -> State: ...
152

153
def Failed(cls: type = State, **kwargs: Any) -> State: ...
154

155
def Running(cls: type = State, **kwargs: Any) -> State: ...
156

157
def Pending(cls: type = State, **kwargs: Any) -> State: ...
158

159
def Cancelled(cls: type = State, **kwargs: Any) -> State: ...
160

161
def Crashed(cls: type = State, **kwargs: Any) -> State: ...
162

163
def Scheduled(
164
    scheduled_time: datetime = None,
165
    name: str = None,
166
    message: str = None,
167
    type: StateType = None,
168
) -> State: ...
169

170
def Paused(
171
    cls: type = State,
172
    timeout_seconds: Optional[int] = None,
173
    pause_expiration_time: Optional[datetime] = None,
174
    reschedule: bool = False,
175
    pause_key: Optional[str] = None,
176
    **kwargs: Any,
177
) -> State: ...
178

179
def Suspended(
180
    cls: type = State,
181
    timeout_seconds: Optional[int] = None,
182
    pause_expiration_time: Optional[datetime] = None,
183
    pause_key: Optional[str] = None,
184
    **kwargs: Any,
185
) -> State: ...
186

187
def AwaitingRetry(
188
    cls: type = State,
189
    scheduled_time: Optional[datetime] = None,
190
    **kwargs: Any,
191
) -> State: ...
192

193
def Retrying(cls: type = State, **kwargs: Any) -> State: ...
194

195
def pause_flow_run(
196
    wait_for_input: Optional[type] = None,
197
    timeout: int = 3600,
198
    poll_interval: int = 10,
199
    key: Optional[str] = None,
200
) -> Any: ...
201

202
def resume_flow_run(
203
    flow_run_id: UUID,
204
    run_input: Optional[Dict[str, Any]] = None,
205
) -> None: ...
206
```
207

208
[State Management](./state-management.md)
209

210
### Deployments
211

212
Deploy flows to work pools, manage deployments, and run deployed workflows.
213

214
```python { .api }
215
def deploy(
216
    *deployments: RunnerDeployment,
217
    work_pool_name: Optional[str] = None,
218
    image: Optional[Union[str, DockerImage]] = None,
219
    build: bool = True,
220
    push: bool = True,
221
    print_next_steps_message: bool = True,
222
    ignore_warnings: bool = False,
223
) -> List[UUID]: ...
224

225
def initialize_project(
226
    name: str = None,
227
    recipe: str = None,
228
) -> None: ...
229

230
def run_deployment(
231
    name: str,
232
    parameters: Dict[str, Any] = None,
233
    scheduled_time: datetime = None,
234
    flow_run_name: str = None,
235
    timeout: int = None,
236
    poll_interval: int = 10,
237
    tags: List[str] = None,
238
    idempotency_key: str = None,
239
    work_queue_name: str = None,
240
) -> FlowRun: ...
241
```
242

243
[Deployments](./deployments.md)
244

245
### Client API
246

247
HTTP clients for interacting with the Prefect API server and cloud services.
248

249
```python { .api }
250
def get_client(
251
    httpx_settings: dict = None,
252
    sync_client: bool = None,
253
) -> Union[PrefectClient, SyncPrefectClient]: ...
254

255
class PrefectClient:
256
    def __init__(
257
        self,
258
        api: str = None,
259
        api_key: str = None,
260
        api_version: str = None,
261
        httpx_settings: dict = None,
262
    ): ...
263

264
    async def create_flow_run(
265
        self,
266
        flow: Flow,
267
        name: str = None,
268
        parameters: Dict[str, Any] = None,
269
        context: Dict[str, Any] = None,
270
        tags: List[str] = None,
271
        parent_task_run_id: UUID = None,
272
        state: State = None,
273
    ) -> FlowRun: ...
274
```
275

276
[Client API](./client-api.md)
277

278
### Context & Utilities
279

280
Context management, logging, annotations, and transaction support.
281

282
```python { .api }
283
def get_run_logger(name: str = None) -> logging.Logger: ...
284

285
def tags(*tags: str, **kwargs) -> ContextManager: ...
286

287
class unmapped:
288
    def __init__(self, value: Any): ...
289

290
class allow_failure:
291
    def __init__(self, value: Any): ...
292

293
class Transaction:
294
    def __init__(
295
        self,
296
        key: Optional[str] = None,
297
        timeout: Optional[float] = None,
298
    ): ...
299
    
300
    def __enter__(self) -> "Transaction": ...
301
    def __exit__(self, exc_type, exc_val, exc_tb) -> None: ...
302
```
303

304
[Context & Utilities](./context-utilities.md)
305

306
### Runtime Context Access
307

308
Access current execution context including deployment parameters, flow run metadata, and task run information during workflow execution.
309

310
```python { .api }
311
# Runtime context modules for accessing execution information
312
import prefect.runtime.deployment
313
import prefect.runtime.flow_run  
314
import prefect.runtime.task_run
315

316
# Deployment context attributes
317
prefect.runtime.deployment.id: str
318
prefect.runtime.deployment.name: str
319
prefect.runtime.deployment.parameters: Dict[str, Any]
320
prefect.runtime.deployment.version: str
321

322
# Flow run context attributes  
323
prefect.runtime.flow_run.id: str
324
prefect.runtime.flow_run.name: str
325
prefect.runtime.flow_run.parameters: Dict[str, Any]
326
prefect.runtime.flow_run.tags: List[str]
327

328
# Task run context attributes
329
prefect.runtime.task_run.id: str
330
prefect.runtime.task_run.name: str
331
prefect.runtime.task_run.task_key: str
332
```
333

334
[Runtime Context Access](./runtime-context.md)
335

336
### Variables Management
337

338
Named, mutable JSON values that can be shared across tasks, flows, and deployments for configuration and data storage.
339

340
```python { .api }
341
from prefect.variables import Variable
342

343
class Variable(BaseModel):
344
    name: str
345
    value: StrictVariableValue
346
    tags: Optional[List[str]]
347
    
348
    @classmethod
349
    def get(
350
        cls,
351
        name: str,
352
        default: StrictVariableValue = None,
353
    ) -> StrictVariableValue: ...
354

355
    @classmethod
356
    async def aget(
357
        cls,
358
        name: str, 
359
        default: StrictVariableValue = None,
360
    ) -> StrictVariableValue: ...
361

362
    @classmethod
363
    def set(
364
        cls,
365
        name: str,
366
        value: StrictVariableValue,
367
        tags: Optional[List[str]] = None,
368
        overwrite: bool = False,
369
    ) -> "Variable": ...
370

371
    @classmethod
372
    async def aset(
373
        cls,
374
        name: str,
375
        value: StrictVariableValue,
376
        tags: Optional[List[str]] = None,
377
        overwrite: bool = False,
378
    ) -> "Variable": ...
379
```
380

381
[Variables Management](./variables.md)
382

383
### Configuration
384

385
Settings, blocks, and configuration management for Prefect infrastructure and credentials.
386

387
```python { .api }
388
class Block(BaseModel):
389
    def save(
390
        self,
391
        name: str,
392
        overwrite: bool = False,
393
    ) -> UUID: ...
394
    
395
    @classmethod
396
    def load(cls, name: str) -> "Block": ...
397

398
def get_settings_context() -> SettingsContext: ...
399
```
400

401
[Configuration](./configuration.md)
402

403
## Types
404

405
Core types used throughout the Prefect API:
406

407
```python { .api }
408
from typing import Any, Callable, Dict, List, Optional, Union, TypeVar, Generic
409
from datetime import datetime, timedelta
410
from uuid import UUID
411
import logging
412

413
# Generic type parameters
414
P = TypeVar("P")  # Parameters
415
R = TypeVar("R")  # Return type
416

417
class Flow(Generic[P, R]):
418
    name: str
419
    fn: Callable[P, R]
420
    description: Optional[str]
421
    version: Optional[str]
422

423
class Task(Generic[P, R]):
424
    name: str
425
    fn: Callable[P, R]
426
    description: Optional[str]
427
    version: Optional[str]
428

429
class State:
430
    type: StateType
431
    name: Optional[str]
432
    message: Optional[str]
433
    data: Any
434
    timestamp: datetime
435

436
class FlowRun:
437
    id: UUID
438
    name: str
439
    flow_id: UUID
440
    state: State
441
    parameters: Dict[str, Any]
442

443
class TaskRun:
444
    id: UUID
445
    name: str
446
    task_key: str
447
    flow_run_id: UUID
448
    state: State
449

450
# Enums and Constants
451
class StateType(str, Enum):
452
    SCHEDULED = "SCHEDULED"
453
    PENDING = "PENDING"
454
    RUNNING = "RUNNING"
455
    COMPLETED = "COMPLETED"
456
    FAILED = "FAILED"
457
    CANCELLED = "CANCELLED"
458
    CRASHED = "CRASHED"
459
    PAUSED = "PAUSED"
460
    SUSPENDED = "SUSPENDED"
461
    AWAITING_RETRY = "AWAITING_RETRY"
462
    RETRYING = "RETRYING"
463

464
# Context types
465
class TaskRunContext:
466
    """Context information available during task execution."""
467
    pass
468

469
# Callable types
470
FlowStateHook = Callable[[Flow, FlowRun, State], None]
471
StateHookCallable = Callable[[Task, TaskRun, State], None]
472
RetryConditionCallable = Callable[[Task, TaskRun, State], bool]
473

474
# Configuration types
475
class TaskRunner:
476
    pass
477

478
class ResultStorage:
479
    pass
480

481
class ResultSerializer:
482
    pass
483

484
# Deployment types
485
class RunnerDeployment:
486
    """Deployment configuration for flows."""
487
    pass
488

489
class DockerImage:
490
    """Docker image configuration for deployments."""
491
    pass
492

493
class DeploymentImage:
494
    """Deployment image configuration."""
495
    pass
496
```