t1 = task1()
t2 = task2(t2)
while condition:
  t3 = task3(a)
  t4 = task4(t3)
  t5 = task5(t4)

docker run

Failed to load and execute Flow's environment: FlowStorageError('An error occurred while unpickling the flow:\n  ModuleNotFoundError("No module named \'my_package\'")\nThis may be due to a missing Python module in your current environment. Please ensure you have all required flow dependencies installed.')

prefect.tasks.prefect.create_flow_run

Task

Failed to load and execute Flow's environment: FlowStorageError('An error occurred while unpickling the flow:\n AttributeError("Can\'t get attribute \'create_params_file\' on <module \'our_filename.py\' from \'/our_filename.py\'>")')

create_params_file

def create_params_file(base_filename: str, **kwargs) -> str:

MY_RESULTS = S3Result(bucket='my_bucket_without_s3_prefix',location='my_output_folder')
prefect.config.flows.checkpointing = True
!export PREFECT__FLOWS__CHECKPOINTING=true
with Flow("please work", result=MY_RESULTS) as f:
    t1 = my_task()
state = f.run()
!aws s3 ls <s3://my_bucket_name_witohut_s3_prefix/my_output_folder>  # nothing is here

line 282, in get_profile_context raise MissingContextError("No profile context found.")

create_flow_run

map

f

wait

LocalDaskExecutor

f

wait

from prefect import Flow, task
import time
from prefect.executors import LocalDaskExecutor

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

@task
def wait(x):
    time.sleep(x)

with Flow("test") as flow:
    a = list(range(4))
    wait.map(a)
    result = f.map(a)

flow.executor = LocalDaskExecutor()

target

pip install prefect[azure]

new_flow_context=prefect.context.get('config')