run_deployment

flow_run.state.result(fetch=True)

@flow(persist_result=False)

prefect.exceptions.MissingResult: State data is missing. Typically, this occurs when result persistence is disabled and the state has been retrieved from the API.

S3Bucket

prefect_aws.s3

python
from prefect_aws.s3 import S3Bucket

s3_bucket = S3Bucket.load("my-bucket")

S3Bucket

0.1.1

prefect-aws

s3

prefect_aws

S3Bucket

0.1.1

prefect_aws

S3Bucket

0.1.1

boto3

boto3

bash
    pip install boto3

python
    import boto3
    from prefect import task, Flow

    @task
    def upload_to_s3(bucket_name, key, data, aws_access_key_id, aws_secret_access_key):
        s3_client = boto3.client(
            "s3",
            aws_access_key_id=aws_access_key_id,
            aws_secret_access_key=aws_secret_access_key,
        )
        s3_client.put_object(Bucket=bucket_name, Key=key, Body=data)
    
    @task
    def download_from_s3(bucket_name, key, aws_access_key_id, aws_secret_access_key):
        s3_client = boto3.client(
            "s3",
            aws_access_key_id=aws_access_key_id,
            aws_secret_access_key=aws_secret_access_key,
        )
        response = s3_client.get_object(Bucket=bucket_name, Key=key)
        return response['Body'].read()

    with Flow("s3-example") as flow:
        # Replace these with your actual bucket name, key, data, and AWS credentials
        upload_to_s3("my-bucket", "my-key", "my-data", "your-access-key", "your-secret-key")
        content = download_from_s3("my-bucket", "my-key", "your-access-key", "your-secret-key")
    
    flow.run()

boto3

prefect_aws

S3Bucket

prefect_aws

@flow(persist_result=True, result_storage="s3-bucket/s3-test")

No class found for dispatch key 's3-bucket' in registry for type 'Block'

No class found for dispatch key 's3-bucket' in registry for type 'Block'

s3-bucket

prefect_aws 0.1.1

prefect_aws

bash
    prefect block register -m prefect_aws

bash
    prefect block types ls

python
    from prefect import flow
    from prefect_aws.s3 import S3Bucket

    @flow(persist_result=True, result_storage=S3Bucket.load("s3-bucket/s3-test"))
    def my_flow():
        return "Persist this result!"

    my_flow()

prefect_aws

S3Bucket

prefect_aws

boto3

S3Bucket

prefect_aws

boto3

prefect

boto3

bash
    pip install prefect boto3

python
    import os
    import boto3
    from prefect.filesystems import WritableFileSystem
    from prefect import flow, task

    class S3Storage(WritableFileSystem):
        def __init__(self, bucket_name, aws_access_key_id, aws_secret_access_key, folder=""):
            self.bucket_name = bucket_name
            self.aws_access_key_id = aws_access_key_id
            self.aws_secret_access_key = aws_secret_access_key
            self.folder = folder
            self.client = boto3.client(
                "s3", 
                aws_access_key_id=self.aws_access_key_id, 
                aws_secret_access_key=self.aws_secret_access_key
            )
        
        def write(self, content, path):
            full_path = os.path.join(self.folder, path)
            self.client.put_object(Bucket=self.bucket_name, Key=full_path, Body=content)
            return full_path

        def read(self, path):
            full_path = os.path.join(self.folder, path)
            obj = self.client.get_object(Bucket=self.bucket_name, Key=full_path)
            return obj['Body'].read()

    # Example usage in a flow
    @flow(persist_result=True, result_storage=S3Storage("your_bucket", "your_access_key", "your_secret_key", "your_folder"))
    def my_flow():
        return "Persist this result to S3!"

    my_flow()

S3Storage

boto3

result_storage

flow

S3Bucket

prefect_aws

"your_bucket"

"your_access_key"

"your_secret_key"

"your_folder"

prefect_aws

bash
   pip install -U prefect

python
   from prefect import flow, task

   @task
   def say_hello():
       print("Hello, world!")

   @flow
   def my_flow():
       say_hello()

   if __name__ == "__main__":
       my_flow()

.py

bash
   python my_flow_script.py

python
from prefect import flow

@flow
def my_job_flow():
    # Your flow logic here

if __name__ == "__main__":
    my_job_flow()

bash
export EXECUTION_ENVIRONMENT=local
python my_job_flow_script.py

prefect config set

PREFECT_API_URL

prefect config set

bash
   prefect config set PREFECT_API_URL=""

@task

.submit

await asyncio.gather

.wait()

.submit()

asyncio.gather()

python
import asyncio
from prefect import task, flow

@task
async def run_me_task(input_data):
    # Simulate some asynchronous work
    await asyncio.sleep(1)
    return f"Processed {input_data}"

@flow
async def main_flow(inputs):
    # Submit tasks
    task_futures = [run_me_task.submit(input_data) for input_data in inputs]
    
    # Await their completion
    results = await asyncio.gather(*[future.result() for future in task_futures])

    # Continue with the rest of the flow
    for result in results:
        print(result)

# Run the flow with some example inputs
if __name__ == "__main__":
    inputs = ["data1", "data2", "data3"]
    asyncio.run(main_flow(inputs))

@task

run_me_task

.submit()

await asyncio.gather()

resources.requests.memory

resources.limits.memory

memory

psutil

python
import psutil
import logging
import time
from prefect import task, flow

# Configure logging to print to stdout
logging.basicConfig(level=<http://logging.INFO|logging.INFO>)

@task
def log_memory_usage():
    process = psutil.Process()
    memory_info = process.memory_info().rss / 1024 ** 2  # Convert to MB
    <http://logging.info|logging.info>(f"Current memory usage: {memory_info:.2f} MB")

@task
def perform_task():
    # Simulate a memory-intensive task
    time.sleep(5)
    return "Task completed"

@flow
def main_flow():
    log_memory_usage.run()
    result = perform_task()
    log_memory_usage.run()
    return result

if __name__ == "__main__":
    main_flow()

log_memory_usage

psutil

log_memory_usage

python
     from prefect import task, flow

     @task(tags=["memory_intensive"])
     def intensive_task():
         pass  # Some memory-intensive operation

     @flow
     def my_flow():
         # Submit tasks, possibly as sub-tasks
         for _ in range(100):
             intensive_task.submit()

     # Set concurrency limits for tasks with this tag elsewhere in your Prefect configuration or via UI

@flow

prefect concurrency-limit

Global

Not ready

successfully created with id

prefect.yaml

asyncio.gather()

python
import asyncio
from prefect import task, flow

@task
async def my_task(n):
    await asyncio.sleep(n)  # Simulate some async work
    return f"Task completed in {n} seconds"

@flow
async def my_flow():
    # Submit multiple tasks
    task_durations = [2, 3, 1, 4]  # Example durations
    task_futures = [my_task.submit(duration) for duration in task_durations]
    
    # Gather results of all tasks
    results = await asyncio.gather(*[future.result() for future in task_futures])
    
    # Print results
    for result in results:
        print(result)

# Entry point
if __name__ == "__main__":
    asyncio.run(my_flow())

my_task

@task

async def

my_flow

@flow

submit()

await asyncio.gather()

asyncio.run(my_flow())

@task

flow

@flow

@flow

@task

Finished in state Failed('Flow run encountered an exception. CrashedRun: Flow run infrastructure exited with non-zero status code -1.')

raise await get_state_exception(state)

final_results = await asyncio.gather(*[flow_run.state.result(fetch=True) for flow_run in flow_runs])

flow_runs

await asyncio.gather(*[run_deployment(...) for params in params_list])

time.sleep

await run_deployment(...)

python
import asyncio
from prefect import flow
from prefect.deployments import run_deployment
from prefect.states import State

async def poll_flow_run_completion(flow_run):
    while True:
        # Refresh the state from the server
        state = await flow_run.state.refresh()
        if state.is_completed():
            # Break when the sub-flow is complete
            break
        elif state.is_failed():
            raise RuntimeError(f"Sub-flow {flow_run.id} failed.")
        await asyncio.sleep(2)  # Wait a bit before checking again

@flow
async def my_parent_flow(params_list):
    flow_runs = await asyncio.gather(*[run_deployment(..., parameters=params) for params in params_list])
    
    # Polling for flow run completion
    await asyncio.gather(*[poll_flow_run_completion(flow_run) for flow_run in flow_runs])

    # Fetch and process results safely
    final_results = [await flow_run.state.result(fetch=True) for flow_run in flow_runs]
    
    # Use final_results as needed

# To trigger the example
if __name__ == "__main__":
    asyncio.run(my_parent_flow([...]))

poll_flow_run_completion()

await asyncio.sleep(2)

fetch=True

result(fetch=True)

state = await flow_run.state.refresh()
AttributeError: 'State' object has no attribute 'refresh'

State

python
from prefect.client import get_client
from prefect.orion.schemas.states import State

async def poll_flow_run_completion(flow_run_id):
    async with get_client() as client:
        while True:
            # Fetch the latest state of the flow run
            state = await client.read_flow_run(flow_run_id=flow_run_id)

            if state.state.is_completed():
                break
            elif state.state.is_failed():
                raise RuntimeError(f"Sub-flow {flow_run_id} failed.")
            await asyncio.sleep(2)  # Wait a bit before checking again

python
@flow
async def my_parent_flow(params_list):
    flow_runs = await asyncio.gather(*[run_deployment(..., parameters=params) for params in params_list])
    
    # Polling for flow run completion
    await asyncio.gather(*[poll_flow_run_completion(flow_run.id) for flow_run in flow_runs])

    # Fetch and process results safely
    final_results = []  # Initialize your final results list
    for flow_run in flow_runs:
        result = await client.read_flow_run(flow_run.id).state.result()
        final_results.append(result)
    
    # Use final_results as needed

id

get_state_exception

run_deployment

run_deployment

run_deployment

run_deployment

python
async def poll_flow_run_completion(flow_run):
    state = flow_run.state
    # Use state directly if provided, or a client for fresh state (preferably)
    if state.is_completed():
        return True
    elif state.is_failed():
        raise RuntimeError(f"Sub-flow {flow_run.id} failed.")
    return False

# Within your flow, handle based on API support from Prefect
async def monitor_flows(flow_runs):
    for flow_run in flow_runs:
        # Polling mechanism to check if flow runs are complete
        while not await poll_flow_run_completion(flow_run):
            await asyncio.sleep(2)

# Triggered flow with awaitable `run_deployment`
flow_runs = await asyncio.gather(*[run_deployment(...) for params in params_list])
await monitor_flows(flow_runs)

state

state

python
from prefect.client import get_client

async def poll_flow_run_completion(flow_run_id):
    async with get_client() as client:
        while True:
            # Fetch the latest state of the flow run
            flow_run = await client.read_flow_run(flow_run_id)  # Fetch the latest flow run
            state = flow_run.state

            if state.is_completed():
                return True
            elif state.is_failed():
                raise RuntimeError(f"Sub-flow {flow_run_id} failed.")
            await asyncio.sleep(2)

# Usage example
flow_runs = await asyncio.gather(*[run_deployment(...) for params in params_list])
await asyncio.gather(*[poll_flow_run_completion(flow_run.id) for flow_run in flow_runs])

httpx.ConnectError: All connection attempts failed

run_deployment

httpx.ConnectError: All connection attempts failed

httpx

httpx

run_deployment

@flow

python
from prefect import flow, task
import asyncio

@task(retries=3, retry_delay_seconds=10)
async def my_sub_task(id):
    # Simulate work that might fail
    print(f"Running task {id}")
    if id % 2 == 0:
        raise Exception("Simulated failure")

@flow(retries=2, retry_delay_seconds=20)
async def my_sub_flow(id):
    await my_sub_task(id)

@flow
async def main_flow():
    flow_run_ids = []
    for i in range(10):  # assuming 10 sub-flows
        sub_flow_run = await run_deployment(name="my_sub_flow", parameters={"id": i})
        flow_run_ids.append(sub_flow_run.id)
    
    # Assume you have a way to collect these and check their state
    await asyncio.gather(*[poll_flow_run_completion(flow_run_id) for flow_run_id in flow_run_ids])

if __name__ == "__main__":
    asyncio.run(main_flow())

retries

retry_delay_seconds

@task

@flow

exponential_backoff

@flow

@flow

python
   from prefect import flow, task
   import boto3  # Example library for interacting with S3

   @task
   def check_cache_and_process(item_id):
       # Create a check to see if this ID has already been processed
       s3_client = boto3.client('s3')
       try:
           s3_client.head_object(Bucket='my-bucket', Key=f'results/{item_id}.json')
           print(f"{item_id} already processed. Skipping.")
           return
       except s3_client.exceptions.NoSuchKey:
           pass  # Continue to process if not found

       # Do processing here
       result = process_item(item_id)
       
       # Save result to cache
       s3_client.put_object(Body=result, Bucket='my-bucket', Key=f'results/{item_id}.json')

   @flow(retries=3, retry_delay_seconds=10)
   async def process_items(item_ids):
       for item_id in item_ids:
           check_cache_and_process(item_id)

   @flow
   async def main_flow(items_batch):
       await process_items(items_batch)

   if __name__ == "__main__":
       asyncio.run(main_flow([...]))

run_deployment

run_deployment

run_deployment

run_deployment

@flow

@task

Job reached backoff limit.

backoffLimit

backoffLimit

backoffLimit

backoffLimit

yaml
apiVersion: batch/v1
kind: Job
metadata:
  name: my-prefect-job
spec:
  template:
    spec:
      containers:
      - name: my-container
        image: my-image
      restartPolicy: Never
  backoffLimit: 5  # Increase this where applicable

run_deployment

parameters=

parameters

run_deployment

deployment.yaml

deployment.yaml

deployment.yaml

yaml
     parameters:
       param1: default_value1
       param2: default_value2

run_deployment

python
from prefect import flow, task

@flow
def my_flow(param1, param2="default_value2"):
    print(f"Param 1: {param1}, Param 2: {param2}")
    # Further operations on parameters

if __name__ == "__main__":
    # Example deployment parameter call
    run_deployment("my-flow-deployment", parameters={"param1": "value1", "param2": "custom_value2"})

deployment.yaml

Job reached backoff limit.

httpx

asyncio

asyncio.Semaphore

python
import asyncio
from prefect import flow, task
from prefect.deployments import run_deployment

async def run_job_with_semaphore(sem, job_params, deployment_name):
    async with sem:
        return await run_deployment(deployment_name, parameters=job_params)

@flow
async def manage_sub_flows(job_chunks, max_concurrent_jobs=5):
    sem = asyncio.Semaphore(max_concurrent_jobs)
    tasks = []
    
    for job_params in job_chunks:
        tasks.append(run_job_with_semaphore(sem, job_params, "my-subflow-deployment"))

    # Await completion of all jobs
    await asyncio.gather(*tasks)

# Example Usage
if __name__ == "__main__":
    job_chunks = [...]  # List of job-specific parameters
    asyncio.run(manage_sub_flows(job_chunks))

asyncio.Semaphore

max_concurrent_jobs

3

bash
prefect work-pool inspect --name <your-work-pool-name>

run_deployment

run_deployment

run_deployment

run_deployment

Reported flow run '71d0a29d-896e-41b0-9c37-f59d3dbb9e56' as crashed: Flow run infrastructure exited with non-zero status code -1.

job_watch_timeout_seconds

await asyncio.gather(*[poll_flow_run_completion(flow_run.id) for flow_run in flow_runs_in_region])

async def poll_flow_run_completion(flow_run_id: uuid.UUID, poll_wait_seconds: int = 20):
    async with get_client() as client:
        while True:
            # Fetch the latest state of the flow run
            flow_run = await client.read_flow_run(flow_run_id)
            state = flow_run.state
            if state.is_completed():
                return True
            elif state.is_failed():
                raise RuntimeError(f"Sub-flow {flow_run_id} failed.")
            await asyncio.sleep(poll_wait_seconds)

region_final_results = await asyncio.gather(*[flow_run.state.result(fetch=True) for flow_run in flow_runs_in_region])

prefect/states.py", line 91, in _get_state_result
    raise await get_state_exception(state)
prefect.exceptions.CrashedRun: Flow run infrastructure exited with non-zero status code -1.

time.sleep

Hey

await asyncio.gather(*[poll_flow_run_completion(flow_run.id) for flow_run in flow_runs_in_region])

async def poll_flow_run_completion(flow_run_id: uuid.UUID, poll_wait_seconds: int = 20):
    async with get_client() as client:
        while True:
            # Fetch the latest state of the flow run
            flow_run = await client.read_flow_run(flow_run_id)
            state = flow_run.state
            if state.is_completed():
                return True
            elif state.is_failed():
                raise RuntimeError(f"Sub-flow {flow_run_id} failed.")
            await asyncio.sleep(poll_wait_seconds)

region_final_results = await asyncio.gather(*[flow_run.state.result(fetch=True) for flow_run in flow_runs_in_region])

prefect/states.py", line 91, in _get_state_result
    raise await get_state_exception(state)
prefect.exceptions.CrashedRun: Flow run infrastructure exited with non-zero status code -1.

await asyncio.gather(*[poll_flow_run_completion(flow_run.id) for flow_run in flow_runs_in_region])

region_final_results = await asyncio.gather(*[flow_run.state.result(fetch=True) for flow_run in flow_runs_in_region])

prefect/states.py", line 91, in _get_state_result
    raise await get_state_exception(state)
prefect.exceptions.CrashedRun: Flow run infrastructure exited with non-zero status code -1.

time.sleep

python
import asyncio
import time

async def poll_flow_run_completion(flow_run_id):
    # Sample completion function
    while True:
        state = ...  # Fetch the state
        if state.is_completed():
            return True
        elif state.is_failed():
            raise RuntimeError(f"Flow run {flow_run_id} failed.")
        await asyncio.sleep(2)

flow_runs_in_region = [...]  # Your setup for regions
await asyncio.gather(*[poll_flow_run_completion(flow_run.id) for flow_run in flow_runs_in_region])

time.sleep(10)  # Increase sleep time to allow full state availability

region_final_results = await asyncio.gather(*[flow_run.state.result(fetch=True) for flow_run in flow_runs_in_region])

httpx

httpx.ConnectError: All connection attempts failed

await asyncio.gather

run_deployment

httpx.ConnectError: All connection attempts failed

httpx

state.is_completed()

flow_run.state.result(fetch=True)

line 91, in _get_state_result
    raise await get_state_exception(state)

COMPLETED

COMPLETED

await

python
import asyncio

async def fetch_results_with_retries(flow_run, max_attempts=3, delay=5):
    attempts = 0
    while attempts < max_attempts:
        try:
            return await flow_run.state.result(fetch=True)
        except Exception as e:
            print(f"Attempt {attempts + 1} failed with error: {e}")
            attempts += 1
            await asyncio.sleep(delay)
    raise RuntimeError(f"Failed to fetch results after {max_attempts} tries.")

# Use it within your fetch logic
region_final_results = await asyncio.gather(
    *[fetch_results_with_retries(flow_run) for flow_run in flow_runs_in_region]
)

flow_run

flow_run.state.result(fetch=True)

State data is missing

state

result

flow_run

state

flow_run

python
from prefect.client import get_client

async def fetch_latest_state(flow_run_id):
    async with get_client() as client:
        flow_run = await client.read_flow_run(flow_run_id)
        return flow_run.state

async def fetch_results_with_retries(flow_run_id, max_attempts=3, delay=5):
    attempts = 0
    while attempts < max_attempts:
        try:
            state = await fetch_latest_state(flow_run_id)
            return await state.result(fetch=True)
        except Exception as e:
            print(f"Attempt {attempts + 1} failed with error: {e}")
            attempts += 1
            await asyncio.sleep(delay)
    raise RuntimeError(f"Failed to fetch results after {max_attempts} tries.")

# Usage with flow run IDs
region_final_results = await asyncio.gather(
    *[fetch_results_with_retries(flow_run.id) for flow_run in flow_runs_in_region]
)

completed

.state.result(fetch=True)

run_deployment

entrypoint

deployment.yaml

persist_results=True

yaml
description: Deployment using S3 for result persistence
storage:
  - aws-s3:
      bucket: your-bucket-name
      key: subflow-results/
parameter_openapi_schema:
  type: object
  properties:
    param1:
      type: string
# Additional configuration specific to your flow

bash
  prefect storage create aws-s3 --bucket your-bucket-name --key your-key-path

python
  @task(persist_result=True)
  async def my_task():
      # Task logic

  @flow(persist_result=True)
  async def my_flow():
      result = await my_task()
      return result

persist_result=True

result_storage

persist_result=True

None

result_storage

persist_result

result_storage

persist_result=True

persist_result=True

python
from prefect import task, flow

@task(persist_result=True)
async def my_task():
    # Task logic

@flow(persist_result=True, result_storage=my_s3_storage_block)
async def my_flow():
    return await my_task()

# Ensure your deployment configuration contains these flags

run_deployment

run_deployment

persist_result=True

deployment.yaml

result_storage

python
from prefect import flow

@flow(persist_result=True)
def my_sub_flow(param):
    # Logic for sub-flow
    return {"key": "value"}

yaml
name: my-sub-flow
flow_name: my_sub_flow
parameters:
  param: default_value
tags: []
entrypoint: flow_script.py:my_sub_flow
result_storage:
  aws_s3:
    bucket: "your-bucket-name"
    key_prefix: "results/"

python
from prefect.deployments import run_deployment
from prefect import flow

@flow
def parent_flow():
    # Run sub-flow with deployment
    result = run_deployment(
        "my-sub-flow",
        parameters={"param": "my_value"},
        wait=True  # Ensures parent flow waits for sub-flow completion
    )
    # process results or other logic

result_storage

persist_result

WritableFileSystem

flow_run.state.result(fetch=True)

"No class found for dispatch key 'mys3storage' in registry for type 'Block'."

WritableFileSystem

flow_run.state.result(fetch=True)

python
  from my_custom_persistence import MyS3Storage

run_deployment

run_deployment

python
from prefect.client import get_client
from prefect.orion.schemas.states import Failed
import asyncio

async def mark_flow_run_as_failed(flow_run_id):
    async with get_client() as client:
        await client.set_flow_run_state(flow_run_id=flow_run_id, state=Failed())

# Call this function with the flow run ID you need to mark
# asyncio.run(mark_flow_run_as_failed(flow_run_id))

from prefect.orion

Failed

Failed

python
from prefect import get_client
from prefect.states import Failed
import asyncio

async def mark_flow_run_as_failed(flow_run_id):
    async with get_client() as client:
        await client.set_flow_run_state(flow_run_id, state=Failed())

# Example usage:
# asyncio.run(mark_flow_run_as_failed(flow_run_id))

prefect

get_client()

pod eviction

kubectl

bash
# Example bash script to check pod status and react
POD_NAME="your-pod-name"
if kubectl get pod $POD_NAME | grep -q "Evicted"; then
  # Consider invoking a Prefect API call here to mark the flow run as failed
  echo "Pod $POD_NAME evicted, needs intervention"
fi

python
from prefect.client import get_client

async def download_logs_to_file(flow_run_id, output_file):
    async with get_client() as client:
        logs = await client.read_logs(flow_run_id=flow_run_id)
        with open(output_file, 'w') as f:
            for log in logs:
                f.write(log.message + '\n')

# Example usage
# asyncio.run(download_logs_to_file(flow_run_id="<flow_run_id>", output_file="flow_logs.txt"))

PREFECT_RUNNER_HEARTBEAT_FREQUENCY

task_function

await

submit

result

object PrefectFuture can't be used in 'await' expression

wait_for_result()

PrefectFuture

python
from prefect import flow, task

@task
def task_function():
    return "Task completed"

@flow
def my_flow():
    future = task_function.submit()
    result = future.wait_for_result()  # Wait for the task to complete and get the result
    print(result)

my_flow()

task_function.submit()

PrefectFuture

wait_for_result()

await

PrefectFuture

AttributeError: 'PrefectFuture' object has no attribute 'wait_for_result'

PrefectFuture

wait_for_result()

result()

python
from prefect import flow, task

@task
def task_function():
    return "Task completed"

@flow
def my_flow():
    future = task_function.submit()
    result = future.result()  # Wait for the task to complete and get the result
    print(result)

my_flow()

future.result()

submit

submit

submit

submit

PrefectFuture

python
   result = task_function()  # Executes immediately, no Prefect orchestration

submit

python
   future = task_function.submit()
   result = future.result()  # Prefect manages execution, allowing orchestration features

submit

get_run_logger()

python
from prefect import task, get_run_logger

@task
def my_task():
    logger = get_run_logger()
    <http://logger.info|logger.info>("This is a log message from the task.")

get_run_logger()

get_run_logger()

get_run_logger()

submit

gather

submit

gather

submit

gather

python
from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner

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

@flow(task_runner=ConcurrentTaskRunner())
def my_flow():
    futures = [my_task.submit(i) for i in range(5)]
    results = [future.result() for future in futures]
    print(results)

if __name__ == "__main__":
    my_flow()

my_task.submit(i)

ConcurrentTaskRunner

future.result()

bash
   prefect config view --show-defaults

send_chat_message

prefect_slack

python
from prefect_slack import SlackCredentials
from prefect_slack.messages import send_chat_message

@flow
def example_send_message_flow():
    context = get_run_context()
    token = "xoxb-your-bot-token-here"
    send_chat_message(
        slack_credentials=SlackCredentials(token),
        channel="#prefect",
        text=f"Flow run {context.flow_run.name} completed :tada:"
    )

example_send_message_flow()

prefect_slack

send_chat_message

prefect_slack.messages

python
from prefect import get_client

async def cancel_all_active_runs():
    async with get_client() as client:
        # Fetch all active flow runs
        flow_runs = await client.read_flow_runs(state_names=["Running"])
        
        for flow_run in flow_runs:
            # Cancel each flow run
            await client.set_flow_run_state(flow_run.id, state="Cancelled")

        # Fetch all active task runs
        task_runs = await client.read_task_runs(state_names=["Running"])
        
        for task_run in task_runs:
            # Cancel each task run
            await client.set_task_run_state(task_run.id, state="Cancelled")

# Run the cancellation process
import asyncio
asyncio.run(cancel_all_active_runs())

"Running"

python
import anyio
from prefect import get_client
from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterState

async def cancel_all_active_flow_runs():
    async with get_client() as client:
        # Define a filter for active flow runs
        filter = FlowRunFilter(
            state=FlowRunFilterState(
                names=["Pending", "Running", "Scheduled", "Late"]
            )
        )
        # Retrieve active flow runs
        flow_runs = await client.read_flow_runs(flow_run_filter=filter)
        
        for flow_run in flow_runs:
            # Cancel each flow run
            await client.set_flow_run_state(flow_run.id, state="Cancelled")

# Run the cancellation process
anyio.run(cancel_all_active_flow_runs)

prefect.states.State

"Cancelled"

State

python
import anyio
from prefect import get_client
from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterState
from prefect.states import Cancelled

async def cancel_all_active_flow_runs():
    async with get_client() as client:
        # Define a filter for active flow runs
        filter = FlowRunFilter(
            state=FlowRunFilterState(
                names=["Pending", "Running", "Scheduled", "Late"]
            )
        )
        # Retrieve active flow runs
        flow_runs = await client.read_flow_runs(flow_run_filter=filter)
        
        for flow_run in flow_runs:
            # Cancel each flow run
            await client.set_flow_run_state(flow_run.id, state=Cancelled())

# Run the cancellation process
anyio.run(cancel_all_active_flow_runs)

Cancelled

Cancelled

prefect.states

Running

Failed

python
import anyio
from prefect import get_client
from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterState
from prefect.states import Cancelled

async def cancel_sub_flows_of_failed_parents():
    async with get_client() as client:
        # Define a filter for failed parent flow runs
        parent_filter = FlowRunFilter(
            state=FlowRunFilterState(
                names=["Failed"]
            )
        )
        # Retrieve failed parent flow runs
        failed_parents = await client.read_flow_runs(flow_run_filter=parent_filter)

        for parent_flow in failed_parents:
            # Retrieve running sub-flows
            sub_flow_filter = FlowRunFilter(
                parent_flow_run_id=parent_flow.id,
                state=FlowRunFilterState(
                    names=["Running"]
                )
            )
            sub_flows = await client.read_flow_runs(flow_run_filter=sub_flow_filter)
            
            for sub_flow in sub_flows:
                # Cancel each running sub-flow
                await client.set_flow_run_state(sub_flow.id, state=Cancelled())

# Run the cancellation process
anyio.run(cancel_sub_flows_of_failed_parents)

filter = FlowRunFilter(
...             state=FlowRunFilterState(
...                 names=["Pending", "Running", "Scheduled", "Late"]
...             )
...         )

FlowRunFilter

Cancelled

FlowRunFilter

FlowRunFilter

read_deployments

read_deployments

aiohttp.client_exceptions.ClientPayloadError: Response payload is not completed:

aiohttp.client_exceptions.ClientPayloadError: Response payload is not completed