wait_for

subflow

fails

import asyncio
from prefect import flow, get_run_logger

@flow
async def test_async_flow():
    await asyncio.sleep(3)
    logger = get_run_logger()
    logger.debug("a debug")
    <http://logger.info|logger.info>("my info")
    logger.warning("some warning")
    logger.critical("super critical")
    logger.error("error!")
    print('all done')

@flow
def test_sync_flow():
    test_async_flow()

test_sync_flow()

timeout_seconds=n

n

block.save("name", overwrite=True)

storage_block = RemoteFileSystem(
        basepath="<s3://prefect-flows>",
        key_type="hash",
        settings=dict(
            use_ssl=False,
            key=config.STORAGE_MINIO_USER,
            secret=config.STORAGE_MINIO_USER,
            client_kwargs=dict(endpoint_url=f"http://{config.STORAGE_MINIO_URL}"),
        ),
    )
    storage_block.save("flow-storage", overwrite=True)
    <http://logging.info|logging.info>("Created storage block")

import helper_file.py

'''
flow that uses helper_file
'''

from prefect import flow, get_run_logger

@flow
def cow_say():
    logger = get_run_logger()
    <http://logger.info|logger.info>("I'm in [red]danger[/red]!")

if __name__ == "__main__":
    cow_say()

wait until finish

register_type_and_schema()

persist_result=True

one

10

5

lambda *args: "one"

lambda *args: "one1"

@task(persist_result=True, cache_key_fn=lambda *args: "one")
async def one(): 
    return 5

@task(persist_result=True, cache_key_fn=lambda *args: "two")
async def two(): 
    return 2

@flow(cache_result_in_memory=False)
async def flow_test():
    a = await one()
    b = await two()
    return a + b 

await flow_test()

/opt/miniconda3/envs/flywheel/lib/python3.10/site-packages/prefect/tasks.py:206: UserWarning: A task named 'one' and defined at '/var/folders/5c/hf9_blgj3jbbxxvkw_s9snsw0000gn/T/ipykernel_90535/3130328684.py:1' conflicts with another task. Consider specifying a unique `name` parameter in the task definition:

 `@task(name='my_unique_name', ...)`
  warnings.warn(
/opt/miniconda3/envs/flywheel/lib/python3.10/site-packages/prefect/tasks.py:206: UserWarning: A task named 'two' and defined at '/var/folders/5c/hf9_blgj3jbbxxvkw_s9snsw0000gn/T/ipykernel_90535/3130328684.py:5' conflicts with another task. Consider specifying a unique `name` parameter in the task definition:

/var/folders/5c/hf9_blgj3jbbxxvkw_s9snsw0000gn/T/ipykernel_90535/3130328684.py

/var/folders/5c/hf9_blgj3jbbxxvkw_s9snsw0000gn/T/

ipykernel_<some_number>

prefect orion database reset -y