What is the correct way of specifying a 'dummy' storage block for a deployment? My code that I have wrapped in prefect tasks correctly deals with absolute paths. Additionally, the data I am working with are on the order of 2TB, and are not really appropriately set up for the default local file system logic used when

--storage-block

is unset to copy. I thought I could set up a 'dummy' local file system block in the Orion UI (in my own managed server, not prefect cloud), however the

prefect deployment build

command says

'github', 's3',  'gcs', 'azure', 'smb'

are supported types. TL;DR - I need to set a

--storage-block

in my deployment, and I am reasonably certain in my situation I do not want to be copying anything to / from different file systems and blocks. I have a common underlying filesystem at the HPC center, and my data are pretty large -- large enough where I can not reasonably expect copying to and from the disk to be feasible

Hi , can someone tell me how to pass Service Account Info in the prefect_gcp block. I wanted to pass the values from the secrets block but prefect 2 doesn't seem to support json in secret block. What is the best way to get GCP BigQuery client from a dictionary. i.e a way to pass a dictionary from any of the blocks so that I can pass the Service Account Info as a parameter to the to obtain the big query client.

Hi everyone, i'm trying to use postgres database as prefect backend, I added the connection string and i stared the orion, all tables created at my db but missing fields, and got this error. can anyone help me?

Hi! When using the Github storage block to connect to a private repo (using Prefect version 2.6.6), I get a password related error (details in the first reply) when running a flow, what am I doing wrong?

Hi. I'm not missing anything specific from this, but wanted to make sure: Prefect 2 does not have an alternative for Prefect 1's

Flow(...).validate()

, right?

Hi, when running prefect deployment build with an infrastructure block I want to override the values of an array in the yaml, does anyone here know how I can do that? 🤔

Hi All, I’m working on moving away from using the Bitnami/Postgres subchart to an Azure DB for Postgres instance. After fixing the dialect name, I’m currently running into this error (will post in thread).

Hi Everyone, My team is getting k8 deployment jobs where the job log states the flow run status cannot transition to

RUNNING

Engine execution of flow run '148d81ea-3cfe-4db1-a0d4-3f3f17748fb0' aborted by orchestrator: This run cannot transition to the RUNNING state from the RUNNING state.

Will add more details to thread

Hello guys, ever since i updated prefect to 2.5+ i can not access the default blocks from Prefect UI. I dont know if that only occurs for me, but when i downgrade prefect to 2.3 the blocks seems to show up again. Any solutions for that?

Hi Everyone, I am trying to trigger prefect flow deployed on cloud whenever file is landed to S3 bucket using AWS Lambda function by referring this documentation from Chris. Lambda function is failing with

"errorMessage": "HTTP Error 404: Not Found"

I am passing API URL

<https://api.prefect.io>

. I understand that this documentation is not latest and I also tried URL

<https://api.prefect.cloud/>

and getting same error. Request your inputs regarding this error in Prefect version 2.0

Hi Prefect Folks - Is it possible to adjust the log level of a running flow? 🧵

Hi Everyone, is it possible to use the below thing for prefect 2.0? https://docs-v1.prefect.io/orchestration/recipes/configuring_storage.html#providing-the-dockerfile

Hi All, Would anyone have experience passing

io.StringIO

objects between tasks in prefect 2.0? Here's a minimal example I'm trying to get to work but getting weird results, appreciate any help on this.

import io
from prefect import task, flow

@task
def string_to_io():
    return io.StringIO("hello prefect")

@task
def consume(string_io):
    print(type(string_io))
    return string_io.read()

@flow
def test():
    # s = string()
    str_io = string_to_io()
    r = consume(str_io)
    print(r)

test()

when I run it I get

9:38:59.511 | INFO    | prefect.engine - Created flow run 'congenial-longhorn' for flow 'test'
19:38:59.708 | INFO    | Flow run 'congenial-longhorn' - Created task run 'string_to_io-78a25967-0' for task 'string_to_io'
19:38:59.709 | INFO    | Flow run 'congenial-longhorn' - Executing 'string_to_io-78a25967-0' immediately...
19:38:59.768 | INFO    | Task run 'string_to_io-78a25967-0' - Finished in state Completed()
19:38:59.789 | INFO    | Flow run 'congenial-longhorn' - Created task run 'consume-884672dc-0' for task 'consume'
19:38:59.790 | INFO    | Flow run 'congenial-longhorn' - Executing 'consume-884672dc-0' immediately...
<class 'list'>
19:38:59.828 | ERROR   | Task run 'consume-884672dc-0' - Encountered exception during execution:
Traceback (most recent call last):
  File "/home/lukasz/.cache/pypoetry/virtualenvs/zlotyryjek-eFYuV0Po-py3.10/lib/python3.10/site-packages/prefect/engine.py", line 1222, in orchestrate_task_run
    result = await run_sync(task.fn, *args, **kwargs)
  File "/home/lukasz/.cache/pypoetry/virtualenvs/zlotyryjek-eFYuV0Po-py3.10/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 68, in run_sync_in_worker_thread
    return await anyio.to_thread.run_sync(call, cancellable=True)
  File "/home/lukasz/.cache/pypoetry/virtualenvs/zlotyryjek-eFYuV0Po-py3.10/lib/python3.10/site-packages/anyio/to_thread.py", line 31, in run_sync
    return await get_asynclib().run_sync_in_worker_thread(
  File "/home/lukasz/.cache/pypoetry/virtualenvs/zlotyryjek-eFYuV0Po-py3.10/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 937, in run_sync_in_worker_thread
    return await future
  File "/home/lukasz/.cache/pypoetry/virtualenvs/zlotyryjek-eFYuV0Po-py3.10/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 867, in run
    result = context.run(func, *args)
  File "/home/lukasz/projects/zlotyryjek/crap.py", line 18, in consume
    return string_io.read()
AttributeError: 'list' object has no attribute 'read'

So for some reason

stringIO

is converted to an empty list 🤔

I have noticed that flow runs seem to get stuck in a running state. This seems to be happening more and more frequently. We are using Prefect 2.4.0 and using K8s and GCS for infrastructure. I have found that for these flow runs I am seeing logs on the agent like this

prefect.agent - An error occured while monitoring flow run <flow_run_id> The flow run will not be marked as failed, but an issue may have occurred.

previously I had something along the lines of

with Flow(
    "...",
    state_handlers=[utils.notify_completion, utils.notify_running],...

with

def notify_running(flow: Flow, old_state, new_state) -> State:

And I could ask what the new state is doing. eg if

if new_state.is_running()

I'm reading https://docs.prefect.io/concepts/states/to try to work out how to capture these flow level state changes and am struggling. Any pointers?

hi folks! long time listener, first time caller 😄 — i have a question about upgrading and downgrading the database (we’re using a postgres container) along side prefect versions. i ran into an issue when trying to rollback from 2.6.6 to 2.6.4 and got this error on orion startup:

alembic.util.exc.CommandError: Can't locate revision identified by '41e5ed9e1034'

i’ll post some details in a thread about what i tried before nuking the database and starting over, but 1) is there guidance on how i should be using

prefect orion database stamp

or

… revision

before upgrading orion from 2.6.4 -> 2.6.6 (or any other minor version) ? and 2) should i be running

… downgrade

from the 2.6.6 package or 2.6.4 package if i wanted to go from 2.6.6->2.6.4?

Does the Orion Server produce any more logs beyond the ones that appear in the agents logs? I can't seem to work out what is causing my flow to crash. I have a flow that runs every minute but at least once a day it will fail due to:

Crash detected! Execution was interrupted by an unexpected exception.

Looking over the logs produced by the agent I can't see the Exception thrown. The flow looks like below. I've changed the variables used but the premise is that two lambdas are invoked but the 2nd needs to wait until the first is completed.

import os
from prefect import flow, get_run_logger, task
from prefect.deployments import Deployment
from prefect.filesystems import S3
from prefect.orion.schemas.schedules import CronSchedule
from utils.aws import invoke_lambda

# --- Flow Details --- #
client = "example_client"
project = "example_project" 
flow_name = "Example Name"
description = """
    Example Description
"""
tags = ["example"]  # Optional list of string tags for the flow. Helps to filter in UI
schedule = CronSchedule(cron="* 7-19 * * MON-FRI", timezone="Australia/Sydney")

# --- Other Flow Details (Leave as is) --- #
work_queue = "example_workqueue"
storage = S3.load("s3-storage")
output_file_name = f"{os.path.splitext(os.path.basename(__file__))[0]}.yaml"

def invoke_lambda(function_full_name, payload, blocking=False, **kwargs):
    """
    Invokes the given lambda function in AWS.

    Args:
        function_full_name (str): Exact name of the lambda function
        payload (dict): Dictionary of data you may want to pass to lambda
        blocking (bool, optional): If you want to wait for a response from the lambda. Defaults to False.
    """
    config = Config(max_pool_connections=250, read_timeout=900, connect_timeout=900)
    lambda_client = boto3.client("lambda", config=config)
    # Change client's region
    if kwargs.get("region"):
        lambda_client = boto3.client(
            "lambda", config=config, region_name=kwargs.get("region")
        )

    # DO NOT BLOCK IF WILL RUN OVER 30 SECONDS
    payload = bytes(json.dumps(payload), encoding="utf8")

    if blocking:
        response = lambda_client.invoke(
            FunctionName=function_full_name,
            InvocationType="RequestResponse",
            LogType="Tail",
            Payload=payload,
        )
        res_payload = response.get("Payload").read()
        return json.loads(res_payload)
    else:
        response = lambda_client.invoke(
            FunctionName=function_full_name,
            InvocationType="Event",
            LogType="None",
            Payload=payload,
        )
        return response

@task
def first_lambda_to_invoke():
    logger = get_run_logger()
    res = invoke_lambda("first_lambda_to_invoke", {}, True)
    <http://logger.info|logger.info>(res)
    return ""


@task
def second_lambda_to_invoke():
    logger = get_run_logger()
    res = invoke_lambda("second_lambda_to_invoke", {}, True)
    <http://logger.info|logger.info>(res)
    return ""


@flow
def example_flow():
    x = first_lambda_to_invoke.submit()
    y = second_lambda_to_invoke.submit(wait_for=[x])


if __name__ == "__main__":
    Deployment.build_from_flow(
        flow=example_flow,
        name=flow_name,
        work_queue_name=work_queue,
        tags=[client, project] + tags,
        schedule=schedule,
        description=description,
        output=output_file_name,
        storage=storage,
        apply=True,
        path="/",
    )

Here is an example of the log output from a failed run

Created task run 'first_lambda_to_invoke-ec90046a-0' for task 'first_lambda_to_invoke'
09:32:10 AM
Submitted task run 'first_lambda_to_invoke-ec90046a-0' for execution.
09:32:10 AM
Created task run 'second_lambda_to_invoke-a66b51c9-0' for task 'second_lambda_to_invoke'
09:32:10 AM
Submitted task run 'second_lambda_to_invoke-a66b51c9-0' for execution.
09:32:10 AM
Created task run 'first_lambda_to_invoke-ec90046a-0' for task 'first_lambda_to_invoke'
09:32:10 AM
Submitted task run 'first_lambda_to_invoke-ec90046a-0' for execution.
09:32:10 AM
Created task run 'second_lambda_to_invoke-a66b51c9-0' for task 'second_lambda_to_invoke'
09:32:10 AM
Submitted task run 'second_lambda_to_invoke-a66b51c9-0' for execution.
09:32:10 AM
Crash detected! Execution was interrupted by an unexpected exception.
09:32:10 AM
first_lambda_to_invoke-ec90046a-0
{'statusCode': 200}
09:32:21 AM
first_lambda_to_invoke-ec90046a-0
Crash detected! Execution was interrupted by an unexpected exception.

Note the

{'statusCode': 200}

response from the Lambda showing that it ran successfully.

Hey - what is the prefect 2.0 equivalent of how I would do `

.set_dependencies(
            upstream_tasks=[my_upstream_task]
        )

Oh - I can see in the message above there is a

wait_for

kwarg in the submit handler... Ignore me 👀

Prefect 2.6.6, local server, docker containers. Hi all, I am getting two errors consistently after some time in agent logs, not running any flows. Theses errors later on fial my flows sometime.

BrokenPipeError: [Errno 32] Broken pipe

ConnectionResetError: [Errno 104] Connection reset by peer

. Any leads would be much appreciated, cheers

I wonder if there is a recommended way to clear task input cache? Details below. Suppose I've set up and ran a task with

cache_key_fn=task_input_hash

and

cache_expiration=timedelta(days=1)

. Now for some reason I want to clear cache and run the task again (suppose I've changed the logic in a module I'm using from the task). My options (including non-working): 1. Deleting cache files doesn't work. If I run a task after deleting files I get an error. Prefect doesn't check if the file exists before deciding on using cache. 2. Deleting flow runs or task runs does work. But it's not convenient for two reasons: a. I should delete history of runs. b. I have to remember which flow/task run to delete or delete all of them. 3. Adding extra parameter

cache_num

to task should work. I can just change it's value every time I want to avoid using cache. But it generates extra boilerplate code and I have to change flow/task code to change parameter value. It looks like a dirty duct tape to me) 4. Changing data in OrionDB maybe possible, but I'm sure that it's strongly not recommended. And I'd like to avoid it. Maybe there is a better option I don't know about? A perfect solution would be CLI or API with flow/task name as an input parameter.

I wonder if anyone has a similar use-case to ours, where we want to let an application (so end-users indirectly) trigger tasks in our Prefect installation. So basically event-driven. I know it’s possible, but I wonder what the best practice is? Currently we are thinking we should add an abstraction layer, especially so that we have a consistent API since the workings of it might change (the way we query a flow/deployment I believe changes when we switch to using Kubernetes for example, and we want our application back-end to be agnostic of that, especially because different people are responsible for that). But is it silly to put a FastAPI in front of Prefect, which under water also uses FastAPI? Or does it make perfect sense from the perspective of abstraction?

Hi all, I wonder the

is_schedule_active

means status of flow (turn on or turn off) ? I got a some flow’s`is_schedule_active` bool value as true by using graphql, but on my prefect UI this flow is turned off state. Am i think wrong?

Hi All,

I have a flow that executes sql query on redshift using prefect-sqlalchemy and my queries are always aborted

Hi, I have a question about Prefect working with Ray https://prefecthq.github.io/prefect-ray/ It is possible to run task on many machines using this setup? Is it creating a Ray Cluster then?

Hi again! I'm trying to set up SQLAlchemy Database Credentials through Prefect UI. What is the format for the Connect Args field? Is it

param=value

,

'param': 'value'

or something else? Anyway after editing this field I cannot create the Block. Chome gives me the following error. UPD. I've found out that I have to fill Query and Connect Args either with some values or with

null

. And the right format for Connect Args is JSON:

{
  "param": "value"
}

P.S. Btw, I cannot create Block in Firefox. Is this browser not supported?

Hello, what is the proper way of triggering the same flow with different parameters in prefect 2, and then wait for all of them to finish. Example of what I want to achieve:

flow_run_ids=[]
for param in params:
    flow_run_id = create_flow_run(
            flow=flow_model,
            parameters=run_params,
            name=run_name,
            state=state
        )
    flow_run_ids.append(flow_run_id)

wait_for_flows(flow_run_ids)

Thanks for any help