config = {
    "Hello World" : "hello_world",
    "Test": "test_flow",
}

if __name__ == "__main__":
    gitlab_repo = GitRepository(
        url = "<https://gitlab_host.com/repo.git>",
        credentials = {
            "access_token": Secret.load("gitlab-token")
        },
    )
    for flow_name, folder_name in config.items():
        flow.from_source(   
            source = gitlab_repo,
            entrypoint = f"orchestrator/{folder_name}/flow.py:run",
        ).deploy(
            name = flow_name,
            work_pool_name = "aws_ecs_dev",
        )

GitRepository

Secret

Secret.load("gitlab-token")

flow.from_source()

.deploy()

deploy

flow

serve

$ python orchestrator/deploy.py 
Traceback (most recent call last):
  File "<frozen importlib._bootstrap_external>", line 995, in exec_module
  File "<frozen importlib._bootstrap_external>", line 1132, in get_code
  File "<frozen importlib._bootstrap_external>", line 1190, in get_data
FileNotFoundError: [Errno 2] No such file or directory: '/var/folders/vg/dlzz1yq51d160q24h4c6f2n40000gp/T/tmpo_1zvf8t/orchestrator_beta/orchestrator/hello_world/flow.py'

GitRepository

GitRepository

entrypoint

entrypoint

flow.from_source()

orchestrator/hello_world/flow.py

entrypoint

flow.from_source()

entrypoint

flow.from_source()

<path_to_script>:<flow_func_name>

flow_func_name

@flow

entrypoint

python
from my_module import MyClass

def my_flow():
    instance = MyClass()
    instance.my_method()

entrypoint="my_script.py:my_flow"

flow.from_source().deploy()

DeploymentTriggerTypes

TriggerTypes

prefect.events

python
from prefect.events import DeploymentTriggerTypes, TriggerTypes

DeploymentScheduleCreate

DeploymentScheduleCreate

prefect.deployments

create_deployment_schedule_create

normalize_to_deployment_schedule_create

prefect.deployments.schedules

DeploymentScheduleCreate

prefect.server.schemas.schedules

CronSchedule

IntervalSchedule

RRuleSchedule

prefect.server.schemas.schedules

DeploymentScheduleCreate

base_job_template = jsonencode(
    jsondecode(
      templatefile(
        "${path.module}/templates/base-job-template.json",
        {
          ecs_cluster_arn      = var.ecs_cluster_arn
          environment          = var.environment
          prefect_docker_image = var.prefect_docker_image
          execution_role_arn   = var.execution_role_arn
          task_role_arn        = var.task_role_arn
        }
      )
    )
  )

aws esc hybrid

ecs_cluster_arn

templatefile

templatefile

hcl
base_job_template = jsonencode(
  jsondecode(
    templatefile(
      "${path.module}/templates/base-job-template.json",
      {
        ecs_cluster_arn      = var.ecs_cluster_arn,
        environment          = var.environment,
        prefect_docker_image = var.prefect_docker_image,
        execution_role_arn   = var.execution_role_arn,
        task_role_arn        = var.task_role_arn
      }
    )
  )
)

base-job-template.json

aws ecs hybrid

json
{
  "family": "my-task-family",
  "containerDefinitions": [
    {
      "name": "my-container",
      "image": "${prefect_docker_image}",
      "essential": true,
      "environment": [
        {
          "name": "ENVIRONMENT",
          "value": "${environment}"
        }
      ]
    }
  ],
  "executionRoleArn": "${execution_role_arn}",
  "taskRoleArn": "${task_role_arn}",
  "networkMode": "awsvpc",
  "requiresCompatibilities": ["FARGATE"],
  "cpu": "256",
  "memory": "512",
  "cluster": "${ecs_cluster_arn}"
}

${ecs_cluster_arn}

"cluster": "${ecs_cluster_arn}"

ecs_cluster_arn

terraform plan

base-job-template.json

templatefile

${}

templatefile

json
{
  "cluster": "${ecs_cluster_arn}",
  "job_configuration": {
    "auto_deregister_task_definition": "{{ auto_deregister_task_definition }}",
    "aws_credentials": "{{ aws_credentials }}",
    "cloudwatch_logs_options": "{{ cloudwatch_logs_options }}",
    "cloudwatch_logs_prefix": "{{ cloudwatch_logs_prefix }}",
    "cluster": "{{ cluster }}",
    "command": "{{ command }}",
    "configure_cloudwatch_logs": "{{ configure_cloudwatch_logs }}",
    "container_name": "{{ container_name }}",
    "env": "{{ env }}",
    "labels": "{{ labels }}",
    "match_latest_revision_in_family": "{{ match_latest_revision_in_family }}",
    "name": "{{ name }}",
    "network_configuration": "{{ network_configuration }}"
  }
}

${placeholder}

templatefile

"cluster": "${ecs_cluster_arn}"

default

{{ placeholder }}

hcl
base_job_template = jsonencode(
  jsondecode(
    templatefile(
      "${path.module}/templates/base-job-template.json",
      {
        ecs_cluster_arn      = var.ecs_cluster_arn,
        environment          = var.environment,
        prefect_docker_image = var.prefect_docker_image,
        execution_role_arn   = var.execution_role_arn,
        task_role_arn        = var.task_role_arn
      }
    )
  )
)

{{ placeholder }}

"env": {
        "title": "Environment Variables",
        "description": "Environment variables to provide to the task run. These variables are set on the Prefect container at task runtime. These will not be set on the task definition.",
        "type": "object",
        "additionalProperties": {
          "type": "string"
        },
      },

json.dumps()

python
import json

# Your JSON object
default_json = {
    "key1": "value1",
    "key2": "value2"
}

# Serialize the JSON object to a string
default_json_string = json.dumps(default_json)

# Example of setting an environment variable
environment_variables = {
    "MY_JSON_VAR": default_json_string
}

# Assuming you are using Prefect, this can be passed to your task
# task.run(env=environment_variables)

base-job-template.json

templatefile

base-job-template.json.tmpl

json
   {
     "env": {
       "title": "Environment Variables",
       "description": "Environment variables to provide to the task run.",
       "type": "object",
       "additionalProperties": {
         "type": "string"
       },
       "MY_JSON_VAR": "${my_json_var}"
     }
   }

.tf

hcl
   variable "my_json_var" {
     description = "The JSON string to pass into the job template"
     type        = string
   }

templatefile

templatefile

hcl
   locals {
     my_json_var = jsonencode({
       key1 = "value1"
       key2 = "value2"
     })
   }

   output "rendered_json" {
     value = templatefile("${path.module}/base-job-template.json.tmpl", {
       my_json_var = local.my_json_var
     })
   }

terraform apply

templatefile

jsonencode

jsonencode

hcl
variable "my_json_var" {
  description = "The JSON string to pass into the job template"
  type        = string
}

locals {
  # Define your JSON structure using Terraform maps and objects
  base_job_template = {
    env = {
      title       = "Environment Variables",
      description = "Environment variables to provide to the task run.",
      type        = "object",
      additionalProperties = {
        type = "string"
      },
      MY_JSON_VAR = var.my_json_var
    }
  }
}

# Convert the local map to a JSON string
output "base_job_template_json" {
  value = jsonencode(local.base_job_template)
}

# Example usage in a Terraform resource
# resource "some_resource" "example" {
#   config = jsonencode(local.base_job_template)
# }

my_json_var

jsonencode

templatefile

@task

Task

Task

Task

run

run

@task

python
from prefect import Task, flow

class MyTask(Task):
    def run(self, x, y):
        # Task logic goes here
        return x + y

# Instantiate your task
my_task = MyTask()

@flow
def my_flow():
    # Use your task instance in the flow
    result = my_task.run(1, 2)
    print(f"Result: {result}")

# Run the flow
my_flow()

Task

run

@task
def task_a():
    pass

@task
def task_b():
    pass

@task
def task_c():
    pass
    
@task
def task_d():
    pass

@flow
def my_flow():
    a = task_a.submit()
    b = task_b.submit()
    # Wait for task_a and task_b to complete
    c = task_c.submit(wait_for=[a, b])
    # task_d will wait for task_c to complete
    # Note: If waiting for one task it must still be in a list.
    d = task_d(wait_for=[c])

submit()

submit()

submit()

python
from prefect import task, flow

@task
def task_a(param1):
    print(f"Task A received: {param1}")

@task
def task_b(param2):
    print(f"Task B received: {param2}")

@task
def task_c(param3):
    print(f"Task C received: {param3}")

@task
def task_d(param4):
    print(f"Task D received: {param4}")

@flow
def my_flow():
    a = task_a.submit(param1="Hello from A")
    b = task_b.submit(param2="Hello from B")
    # Wait for task_a and task_b to complete
    c = task_c.submit(param3="Hello from C", wait_for=[a, b])
    # task_d will wait for task_c to complete
    d = task_d.submit(param4="Hello from D", wait_for=[c])

# Run the flow
my_flow()

submit()

submit()

wait_for

wait_for

Crash detected! Execution was cancelled by the runtime environment.

asyncio

time.sleep()

async

await asyncio.sleep()

ProcessPoolExecutor

fork

spawn

Crash detected! Execution was cancelled by the runtime environment

def fibonacci(n):
    if n <= 1:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

@task(log_prints=True)
def dummy_task(task_definition: dict):
    fib_number = 37
    fibonacci(fib_number)
    print(task_definition)

for task in parallel_tasks:
    dummy_task.with_options(name=task_name).submit(something_to_print, wait_for=deps)

fibonacci

n

python
  def fibonacci(n, memo={}):
      if n in memo:
          return memo[n]
      if n <= 1:
          return n
      memo[n] = fibonacci(n-1, memo) + fibonacci(n-2, memo)
      return memo[n]

bash
# Set a concurrency limit of 8 for tasks with the "my_task" tag
prefect concurrency-limit create --tag my_task --limit 8

python
from prefect import task, flow

@task(tags=["my_task"])
def example_task():
    # Your task logic here
    pass

@flow
def my_flow():
    # Submit tasks with concurrency control
    futures = [example_task.submit() for _ in range(20)]  # Example of 20 tasks
    for future in futures:
        future.result()  # Wait for all tasks to complete

# Run the flow
my_flow()

@task

@flow

submit()

wait_for

python
from prefect import task, flow

@task
def task_a():
    print("Running Task A")
    return "Result from A"

@task
def task_b():
    print("Running Task B")
    return "Result from B"

@task
def task_c(a_result, b_result):
    print(f"Running Task C with {a_result} and {b_result}")
    return "Result from C"

@task
def task_d(c_result):
    print(f"Running Task D with {c_result}")

@flow
def my_flow():
    # Submit tasks without dependencies
    a_future = task_a.submit()
    b_future = task_b.submit()

    # Submit task C, which depends on A and B
    c_future = task_c.submit(a_result=a_future, b_result=b_future)

    # Submit task D, which depends on C
    task_d.submit(c_result=c_future)

# Run the flow
my_flow()

.submit()

wait_for

wait_for

.submit()

@task(log_prints=True)
def say_hello(name: str) -> None:
    print(f"Hello, {name}!")


@flow(name="Hello World")
def run(names: list[str]) -> None:
    say_hello.map(names) # most common parallel task

class A:
    @flow(name="Hello World")
    def run(self, names: list[str]) -> None:
        say_hello.map(names) # most common parallel task

self

cls

self

cls

python
from prefect import task, flow

@task(log_prints=True)
def say_hello(name: str) -> None:
    print(f"Hello, {name}!")

class A:
    @staticmethod
    @flow(name="Hello World")
    def run(names: list[str]) -> None:
        say_hello.map(names)

# Usage
A.run(["Alice", "Bob", "Charlie"])

python
from prefect import task, flow

@task(log_prints=True)
def say_hello(name: str) -> None:
    print(f"Hello, {name}!")

class A:
    @classmethod
    @flow(name="Hello World")
    def run(cls, names: list[str]) -> None:
        say_hello.map(names)

# Usage
A.run(["Alice", "Bob", "Charlie"])

@staticmethod

self

@classmethod

cls

self

self

cls

python
from prefect import task, flow

@task(log_prints=True)
def say_hello(name: str) -> None:
    print(f"Hello, {name}!")

class A:
    @classmethod
    @flow(name="Hello World")
    def run(cls, names: list[str]) -> None:
        say_hello.map(names)

# Usage
A.run(["Alice", "Bob", "Charlie"])

python
from prefect import task, flow

@task(log_prints=True)
def say_hello(name: str) -> None:
    print(f"Hello, {name}!")

class A:
    @classmethod
    @flow(name="Hello World")
    def run(cls, names: list[str]) -> None:
        say_hello.map(names)

# Entry point
if __name__ == "__main__":
    A.run(["Alice", "Bob", "Charlie"])

if __name__ == "__main__":

A.run()

{'task_1': {'task_2', 'task_3'}, 'task_4': {'task_3'}, 'task_5': {'task_4'}, 'task_2': {}, 'task_3': {'task_2'}}

@task

submit()

wait_for

python
from prefect import task, flow

@task
def task_1():
    print("Executing Task 1")

@task
def task_2():
    print("Executing Task 2")

@task
def task_3():
    print("Executing Task 3")

@task
def task_4():
    print("Executing Task 4")

@task
def task_5():
    print("Executing Task 5")

dependency_graph = {
    'task_1': {'task_2', 'task_3'},
    'task_4': {'task_3'},
    'task_5': {'task_4'},
    'task_2': {},
    'task_3': {'task_2'}
}

@flow
def dynamic_flow():
    # Dictionary to hold task futures
    task_futures = {}

    # Iterate over the dependency graph
    for task_name, dependencies in dependency_graph.items():
        # Retrieve the task function by name
        task_func = globals()[task_name]
        
        # Gather futures for dependencies
        wait_for = [task_futures[dep] for dep in dependencies]
        
        # Submit the task with its dependencies
        task_futures[task_name] = task_func.submit(wait_for=wait_for)

# Run the flow
dynamic_flow()

@task

wait_for

dependency_graph

{'task_1': {'task_2', 'task_3'}, 'task_4': {'task_3'}, 'task_5': {'task_4'}, 'task_2': {}, 'task_3': {'task_2'}}

@task(log_prints=True)
def dummy_task(task_definition: dict):
    print(task_definition)

wait_for

wait_for

python
from prefect import task, flow

@task(log_prints=True)
def dummy_task(task_definition: dict):
    print(f"Executing task with definition: {task_definition}")

# Example dependency graph
dependency_graph = {
    'task_1': {'task_2', 'task_3'},
    'task_4': {'task_3'},
    'task_5': {'task_4'},
    'task_2': {},
    'task_3': {'task_2'}
}

# Task definitions
task_definitions = {
    'task_1': {'name': 'Task 1'},
    'task_2': {'name': 'Task 2'},
    'task_3': {'name': 'Task 3'},
    'task_4': {'name': 'Task 4'},
    'task_5': {'name': 'Task 5'},
}

@flow
def dynamic_flow():
    # Dictionary to hold task futures
    task_futures = {}

    # Iterate over the dependency graph
    for task_name, dependencies in dependency_graph.items():
        # Gather futures for dependencies
        wait_for = [task_futures[dep] for dep in dependencies]
        
        # Submit the task with its dependencies
        task_futures[task_name] = dummy_task.submit(task_definitions[task_name], wait_for=wait_for)

# Run the flow
dynamic_flow()

dummy_task

task_definition

dependency_graph

wait_for

wait_for

dependency_graph

task_definitions

prefect._internal.concurrency - Service 'EventsWorker' failed with 16 pending items

Service 'EventsWorker' failed with 16 pending items

PREFECT_DEBUG_MODE=1

python
from prefect import task, flow

@task(log_prints=True)
def dummy_task(task_definition: dict):
    print(f"Executing task with definition: {task_definition}")

# Example dependency graph
dependency_graph = {
    'task_1': {'task_2', 'task_3'},
    'task_4': {'task_3'},
    'task_5': {'task_4'},
    'task_2': {},
    'task_3': {'task_2'}
}

# Task definitions
task_definitions = {
    'task_1': {'name': 'Task 1'},
    'task_2': {'name': 'Task 2'},
    'task_3': {'name': 'Task 3'},
    'task_4': {'name': 'Task 4'},
    'task_5': {'name': 'Task 5'},
}

@flow
def dynamic_flow():
    # Dictionary to hold task futures
    task_futures = {}

    # Iterate over the dependency graph
    for task_name, dependencies in dependency_graph.items():
        # Gather futures for dependencies
        wait_for = [task_futures[dep] for dep in dependencies]
        
        # Submit the task with its dependencies
        task_futures[task_name] = dummy_task.submit(task_definitions[task_name], wait_for=wait_for)

# Run the flow
dynamic_flow()

@flow
def dynamic_flow():
    # Dictionary to hold task futures
    task_futures = {}

    # Iterate over the dependency graph
    for task_name, dependencies in dependency_graph.items():
        # Gather futures for dependencies
        wait_for = [task_futures[dep] for dep in dependencies]
        
        # Submit the task with its dependencies
        task_futures[task_name] = dummy_task.submit(task_definitions[task_name], wait_for=wait_for)
    for name, link in task_futures.items():
        link.wait()

python
from prefect import task, flow

@task(log_prints=True)
def dummy_task(task_definition: dict):
    print(f"Executing task with definition: {task_definition}")

# Example dependency graph
dependency_graph = {
    'task_1': {'task_2', 'task_3'},
    'task_4': {'task_3'},
    'task_5': {'task_4'},
    'task_2': {},
    'task_3': {'task_2'}
}

# Task definitions
task_definitions = {
    'task_1': {'name': 'Task 1'},
    'task_2': {'name': 'Task 2'},
    'task_3': {'name': 'Task 3'},
    'task_4': {'name': 'Task 4'},
    'task_5': {'name': 'Task 5'},
}

@flow
def dynamic_flow():
    # Dictionary to hold task futures
    task_futures = {}

    # Iterate over the dependency graph
    for task_name, dependencies in dependency_graph.items():
        # Gather futures for dependencies
        wait_for = [task_futures[dep] for dep in dependencies]
        
        # Submit the task with its dependencies
        task_futures[task_name] = dummy_task.submit(task_definitions[task_name], wait_for=wait_for)

# Run the flow
dynamic_flow()

for name, link in task_futures.items():
        link.wait()

from prefect import task, flow

@task(log_prints=True)
def dummy_task(task_definition: dict):
    print(f"Executing task with definition: {task_definition}")
dependency_graph = {
    'task_1': {'task_2', 'task_3'},
    'task_4': {'task_3'},
    'task_5': {'task_4'},
    'task_2': {},
    'task_3': {'task_2'}
}

# Task definitions
task_definitions = {
    'task_1': {'name': 'Task 1'},
    'task_2': {'name': 'Task 2'},
    'task_3': {'name': 'Task 3'},
    'task_4': {'name': 'Task 4'},
    'task_5': {'name': 'Task 5'},
}

@flow
def dynamic_flow():
    # Dictionary to hold task futures
    task_futures = {}

    # Iterate over the dependency graph
    for task_name, dependencies in dependency_graph.items():
        # Gather futures for dependencies
        wait_for = [task_futures[dep] for dep in dependencies]
        
        # Submit the task with its dependencies
        task_futures[task_name] = dummy_task.submit(task_definitions[task_name], wait_for=wait_for)

# Run the flow
dynamic_flow()

for name, link in task_futures.items():
        link.wait()

from prefect import task, flow

@task(log_prints=True)
def dummy_task(task_definition: dict):
    print(f"Executing task with definition: {task_definition}")

dependency_graph = {
    'task_1': {'task_2', 'task_3'},
    'task_4': {'task_3'},
    'task_5': {'task_4'},
    'task_2': {},
    'task_3': {'task_2'}
}

task_definitions = {
    'task_1': {'name': 'Task 1'},
    'task_2': {'name': 'Task 2'},
    'task_3': {'name': 'Task 3'},
    'task_4': {'name': 'Task 4'},
    'task_5': {'name': 'Task 5'},
}

@flow
def dynamic_flow():
    task_futures = {}
    for task_name, dependencies in dependency_graph.items():
        wait_for = [task_futures[dep] for dep in dependencies]  
        task_futures[task_name] = dummy_task.submit(task_definitions[task_name], wait_for=wait_for)

dynamic_flow()

for name, link in task_futures.items():
        link.wait()

from prefect import task, flow

@task(log_prints=True)
def dummy_task(task_definition: dict):
    print(f"Executing task with definition: {task_definition}")
dependency_graph = {
    'task_1': {'task_2', 'task_3'},
    'task_4': {'task_3'},
    'task_5': {'task_4'},
    'task_2': {},
    'task_3': {'task_2'}
}
task_definitions = {
    'task_1': {'name': 'Task 1'},
    'task_2': {'name': 'Task 2'},
    'task_3': {'name': 'Task 3'},
    'task_4': {'name': 'Task 4'},
    'task_5': {'name': 'Task 5'},
}
@flow
def dynamic_flow():
    task_futures = {}
    for task_name, dependencies in dependency_graph.items():
        wait_for = [task_futures[dep] for dep in dependencies]  
        task_futures[task_name] = dummy_task.submit(task_definitions[task_name], wait_for=wait_for)

dynamic_flow()

for name, link in task_futures.items():
        link.wait()

from prefect import task, flow

@task(log_prints=True)
def dummy_task(task_definition: dict):
    print(f"Executing task with definition: {task_definition}")
dependency_graph = {
    'task_1': {'task_2', 'task_3'},
    'task_4': {'task_3'},
    'task_2': {},
    'task_3': {'task_2'}
}
task_definitions = {
    'task_1': {'name': 'Task 1'},
    'task_2': {'name': 'Task 2'},
    'task_3': {'name': 'Task 3'},
    'task_4': {'name': 'Task 4'},
}
@flow
def dynamic_flow():
    task_futures = {}
    for task_name, dependencies in dependency_graph.items():
        wait_for = [task_futures[dep] for dep in dependencies]  
        task_futures[task_name] = dummy_task.submit(task_definitions[task_name], wait_for=wait_for)

dynamic_flow()

for name, link in task_futures.items():
        link.wait()

{'task_1': {'task_2', 'task_3'}, 'task_4': {'task_3'}, 'task_5': {'task_4'}, 'task_2': {}, 'task_3': {'task_2'}}

@task(log_prints=True)
def dummy_task(task_definition: dict):
    print(task_definition)

wait_for

wait_for

python
from prefect import task, flow

@task(log_prints=True)
def dummy_task(task_definition: dict):
    print(f"Executing task with definition: {task_definition}")

# Example dependency graph
dependency_graph = {
    'task_1': {'task_2', 'task_3'},
    'task_4': {'task_3'},
    'task_5': {'task_4'},
    'task_2': {},
    'task_3': {'task_2'}
}

# Task definitions
task_definitions = {
    'task_1': {'name': 'Task 1'},
    'task_2': {'name': 'Task 2'},
    'task_3': {'name': 'Task 3'},
    'task_4': {'name': 'Task 4'},
    'task_5': {'name': 'Task 5'},
}

@flow
def dynamic_flow():
    # Dictionary to hold task futures
    task_futures = {}

    # Iterate over the dependency graph
    for task_name, dependencies in dependency_graph.items():
        # Gather futures for dependencies
        wait_for = [task_futures[dep] for dep in dependencies]
        
        # Submit the task with its dependencies
        task_futures[task_name] = dummy_task.submit(task_definitions[task_name], wait_for=wait_for)

    # Ensure all tasks complete
    for future in task_futures.values():
        future.wait()

# Run the flow
dynamic_flow()

dummy_task

task_definition

dependency_graph

wait_for

wait_for

dependency_graph

task_definitions

prefect-aws

AwsSecret

AwsSecret

on_failure=[send_slack_notification]

def send_slack_notification(flow: Flow, flow_run: FlowRun, state: State) -> None:
    something

send_slack_notification

PrefectClient

python
from prefect.client.orchestration import PrefectClient

def send_slack_notification(flow: Flow, flow_run: FlowRun, state: State) -> None:
    # Initialize the Prefect client
    with PrefectClient() as client:
        # Fetch the flow run details
        flow_run_details = client.read_flow_run(flow_run.id)
        
        # Extract the deployment name
        deployment_name = flow_run_details.deployment_id
        
        # Use the deployment name as needed
        print(f"Deployment Name: {deployment_name}")

flow_run.id

FlowRun

.map()

Complete

wait_for

wait_for

wait_for

wait_for

map()

map

wait_for

python
from prefect import flow, task

@task
def my_task(x):
    return x * 2

@flow
def my_flow():
    results = my_task.map([1, 2, 3], wait_for=None)
    # Perform any additional logic if needed
    return results

if __name__ == "__main__":
    my_flow()

.map()

.result()

wait_for

wait_for

wait_for

wait_for

map

wait_for

python
from prefect import flow, task

@task
def task_1():
    # Some processing
    return "Task 1 complete"

@task
def task_2(x):
    return f"Task 2 complete with {x}"

@flow
def my_flow():
    x = task_1.submit()
    # task_2 will wait for task_1 to complete
    results = task_2.map([1, 2, 3], wait_for=[x])
    return results

if __name__ == "__main__":
    my_flow()

task_2

task_1

wait_for

.result()

PREFECT_RUNNER_HEARTBEAT_FREQUENCY

PREFECT_RUNNER_HEARTBEAT_FREQUENCY

.bashrc

.zshrc

base-job-template.json

from datetime import timedelta

from prefect.automations import Automation
from prefect.client.schemas.objects import StateType
from prefect.events.actions import ChangeFlowRunState
from prefect.events.schemas.automations import EventTrigger, Posture
from prefect.events.schemas.events import ResourceSpecification


my_automation = Automation(
    name="Crash zombie flows",
    trigger=EventTrigger(
        after={"prefect.flow-run.heartbeat"},
        expect={
            "prefect.flow-run.heartbeat",
            "prefect.flow-run.Completed",
            "prefect.flow-run.Failed",
            "prefect.flow-run.Cancelled",
            "prefect.flow-run.Crashed",
        },
        match=ResourceSpecification({"prefect.resource.id": ["prefect.flow-run.*"]}),
        for_each={"prefect.resource.id"},
        posture=Posture.Proactive,
        threshold=1,
        within=timedelta(seconds=90),
    ),
    actions=[
        ChangeFlowRunState(
            state=StateType.CRASHED,
            message="Flow run marked as crashed due to missing heartbeats.",
        )
    ],
)

if __name__ == "__main__":
    my_automation.create()

null_resource

external

null_resource

external

deployment_name

flow_run

prefect.runtime

flow_run

prefect.runtime.flow_run

deployment_name

deployment_name

name

prefect.runtime.deployment

python
from prefect.runtime.deployment import name as deployment_name

# Access the deployment name
print(deployment_name)

ValueError: Flows defined interactively cannot be deployed. Check out the quickstart guide for help getting started: <https://docs.prefect.io/latest/get-started/quickstart>

ValueError: Flows defined interactively cannot be deployed

Running

Running

Running

on_completion

on_failure

on_running

StateHookCallable

python
from prefect import flow
from prefect.states import StateHookCallable

def custom_on_completion(state):
    print("Flow completed with state:", state)

def custom_on_failure(state):
    print("Flow failed with state:", state)

def custom_on_running(state):
    print("Flow is running with state:", state)

@flow
def my_flow():
    # Your flow logic here
    pass

# Patch the hooks for testing
my_flow.on_completion(custom_on_completion)
my_flow.on_failure(custom_on_failure)
my_flow.on_running(custom_on_running)

# Now you can run your flow and observe the patched behavior
my_flow()

on_completion

on_failure

on_running

Data integrity
│ conflict. This usually means a unique or foreign key constraint was
│ violated. See server logs for details.

python
   from prefect import flow
   from prefect.deployments import Deployment
   from prefect.server.schemas.schedules import CronSchedule

   @flow
   def my_flow():
       # Flow logic here
       pass

   # Define a cron schedule with a specific timezone
   cron_schedule = CronSchedule(cron="0 0 * * *", timezone="America/New_York")

   # Create a deployment with the cron schedule
   deployment = Deployment.build_from_flow(
       flow=my_flow,
       name="my_flow_deployment",
       schedule=cron_schedule
   )

   deployment.apply()

prefect.yaml

yaml
   schedules:
     - cron: "0 0 * * *"
       timezone: "America/New_York"

prefect deploy

flow.from_source(
            source=gitlab_repo,
            entrypoint=deployment_config.entrypoint,
        ).deploy(something)

flow.from_source

python
from prefect import flow
from prefect.deployments import Deployment
from prefect.server.schemas.schedules import CronSchedule

# Define your flow
@flow
def my_flow():
    # Flow logic here
    pass

# Define your cron schedule with a timezone
cron_schedule = CronSchedule(cron="0 0 * * *", timezone="America/New_York")

# Deploy the flow from source with the cron schedule
Deployment.from_source(
    flow=my_flow,
    source=gitlab_repo,
    entrypoint=deployment_config.entrypoint,
    name="my_flow_deployment",
    schedule=cron_schedule
).apply()

CronSchedule

timezone

Deployment.from_source

schedule

CronSchedule

Deployment.from_source

python
from prefect import flow
from prefect.deployments import Deployment
from prefect.server.schemas.schedules import CronSchedule
from prefect_gitlab import GitLabCredentials

# Define your flow
@flow
def my_flow():
    # Flow logic here
    pass

# Define your cron schedule with a timezone
cron_schedule = CronSchedule(cron="0 0 * * *", timezone="America/New_York")

# GitLab credentials (if required)
gitlab_credentials = GitLabCredentials(
    access_token="your_access_token"  # Only if your repo is private
)

# Deploy the flow from source with the cron schedule
Deployment.from_source(
    flow=my_flow,
    source={
        "gitlab_repo": "your-gitlab-repo-url",
        "entrypoint": "path/to/your/flow.py:my_flow",
        "gitlab_credentials": gitlab_credentials
    },
    name="my_flow_deployment",
    schedule=cron_schedule
).apply()

CronSchedule

prefect-gitlab

CronSchedule

deployment_name

deployment_name

{{ deployment_name }}

deployment_name

@task

@task

@task

python
# Original task function
# from prefect import task

# @task
def my_function(x, y):
    return x + y

# Running as an ordinary function
result = my_function(3, 4)
print(result)  # Outputs: 7

my_function