Good morning everyone question for y’all. I have a experimental set up where I have: 1. settings.py: Basically metadata in a dictionary 2. pipeline.py: where I define my flow 3. folder tasks: with several task.py (with different names of course) For now settings.py has the metadata of a single experiment including each experiment directory. Hence it is called/imported in each task file as well as the main pipeline file in order to have it as a global variable sharer across files. For example the results folder path is passed to tasks decorators so that each task knows where to save the result of the task and where to look when cacheing (i am using file based cacheing). The approach works perfectly when I just run one experiment, for which I detail its metadata in settings.py manually. However I have been trying to set a battery of experiments where only one or a few values of the metadata change. For that I attempted to set a second settings file (battery_experiments.py). This file contains lists of values for certain arguments in settings.py. Each of these are used in a for loop where I make several calls of the

@flow

in pipeline.py, each time updating settings.py metadata with values from battery_experiments.py. The issue is that even though the global variables within settings change in each iteration (I print them out), it seem like the task decorator keeps taking the first iteration value always. I am guessing that this happens because the decorators are compiled before anything is ran, so they keep the values in settings.py before any update is done. I am not sure how to make the iterative updates in global variables to be reflected in the task decorator input. Does any of you know if this is possible at all? Thanks a lot!

Hi there. I want to have a task that processes task states. Not sure why I'm getting this error:

prefect.exceptions.UnfinishedRun: Run is in PENDING state, its result is not available.

Given that I should be waiting for other tasks to finish. Full source code:

from typing import List

from prefect import task, flow, allow_failure
from prefect.task_runners import ConcurrentTaskRunner
from prefect.client.schemas import State


@task
def divide_by_zero(input: int):
    return input / 0


@task
def return_input(input: int):
    return input


@task
def process_futures(futures_list: List[State]):
    for future in futures_list:
        print(future.is_failed())

        if future.is_failed() == False:
            print(future.result())


@flow(task_runner=ConcurrentTaskRunner())
def test_wait_for():

    failed_future = divide_by_zero.submit(1, return_state=True)
    success_future = return_input.submit(1, return_state=True)

    futures_list=[failed_future, success_future]

    print(failed_future.is_failed())
    print(success_future.is_failed())

    process_futures(futures_list, wait_for=[allow_failure(failed_future), allow_failure(success_future)])

if __name__ == '__main__':
    test_wait_for()

Hey everyone, I am working on some flows using prefect 2.7.12, specifically sending an email when a flow fails. I'm having trouble accessing the name of the flow run instance (adjective, noun). I would like to be able to include this information in the email. I've tried a handful of things including: flow_run_name =context['flow_run_name'] flow_run_name = prefect.context.get("flow_run_name") flow_run_name = prefect.context.flow_run.name flow_name = get_flow_name() and probably a few that I am forgetting. Does anyone have any suggestions or advise? Anything you can provide is much appreciated!

I'm trying to set

PREFECT_API_URL

programmatically in my python code but I can't, is there any example regarding this?

hello, is there a way to disable a particular block?

I've got thousands of flows running in my deployment. I want to cancel all of them related to a specific flow name programmatically. How can I do this from Python?

Could anyone explain why a task run concurrency limit I have set still shows active task runs although no tasks are currently running? My limit is 8 slots and it shows 7 active task runs. If I were to run my flow it would run 1 task every 30 seconds. Thoughts?

I got a question, can we not see the Prefect UI unless we use Prefect Cloud?

Hello. In the article below, I understood how to activate the perfect flow in an event-driven manner. Is it possible to run flow(lambda) on a schedule set in prefect cloud? https://medium.com/the-prefect-blog/event-driven-data-pipelines-with-aws-lambda-prefect-and-github-actions-b3d9f84b1309

Hi guys, We are using version 2.10.12 and we found some problems in the UI. Under Flow Runs tab the graph is not responding well for sub-flow arrow button pushs. Furthermore, the name of the flow is written in white on a bright green line and it's hard to see.

Hi I have some questions related to disk space usage on prefect database. Mostly related to how block documents are stored. Details will be added soon in the thread.

Hi, Can I install worker on a Windows machine as well? or worker can be installed on Linux OS only?

Also, from some reason when I try to create a pool in the UI I can’t choose the infra:

The mostly UI is stateless, most filters and sorting rules are discarded at every click. Is there a way to solve this problem ? I'm using extensively the UI for debugging purpose... I'd love for instance to be able to keep "Newest to oldest" in the logs tab. While I understand that implementing session and user profile may not be the priority, maybe permanently through a config variable ?

Hi everyone! Does anyone know why I might be randomly getting these error messages from

prefect-gcp

? It is in an environment raising the flow in cloud run, on the machine where the prefect agent runs

prefect==2.10.7

and

prefect-gcp==0.4.1

are installed, it is a bit difficult to debug since the flow does not reach to be created in cloud run, however after a manual retry it works without problems, Thanks in advance!...

I am building deployments as subprocesses on my VM. I got it working so far, but I would like them to run within a specific

pyenv

now. Does anyone have experience with this or documentation that can point me in the correct direction? I tried using commands in the infrastructure block with no luck.

Sometimes, one of our flows will fail with a CRASHED state. We've learned that this is a Prefect infrastructure error, so retries will not happen. However, we need to figure out how to detect a crash and resend the flow. Any suggestions? For what it's work, we're seeing

Flow run infrastructure exited with non-zero status code -9.

with the crash.

Is anyone using Github to manage there flow development and can you specify the branch and SHA for the flow to use for execution

Probably a dumb question, but want to ask before I go all the way to trying it out with a deployment: Can a task make use of a non-task function in the same file? I think my tasks will be a lot easier to read and maintain if I don't have to have too much going on all in one function. For example:

def little_function(thing):
  print(thing)

@task(name='big function')
def big_function(list_of_things):
  for thing in list_of_things:
    little_function(thing)

Hi Community, I have encountered a very strange behavior with retries on the flow level. Here is my flow that connects to an SFTP server and loads data from it. If a file is not found on the SFTP server, I close the connection and raise an exception. I expect that the flow will retry to re-run from the beginning, starting from the connection to the SFTP server. But it fails with the error “Socket is closed.” It appears to me that Prefect caches the connection and does not re-run the flow from the beginning. What is the best practice for restarting connection on retry?

class SFTPClient(Base):
    def __init__(self, block_name: str, broker_name: str = None) -> None:
        self.rfs = RemoteFileSystem.load(block_name)
    
    def load_file_data(self, remote_path: str) -> pd.DataFrame:
        ...


@flow(
    name="Transfer files to S3 bucket",
    retries=20,
    retry_delay_seconds=5,
)
def flowrun(date: str = None):
    context = get_run_context()

    sftclient = SFTPClient(block_name="rjo-sftp", broker_name="RJO")
    data = sftclient.load_file_data(remote_path="/tmp/test.csv")

    if not data:
        sftclient.rfs.filesystem.client.close()
        sftclient = None
        raise Exception("No data")

Is it possible there is a bug with legacy (0.15x) flow scheduling as of last night? Despite not making any changes yesterday, we saw jobs start piling up late in the evening (UTC), where previously everything had been running fine for many months. I know the deadline for switching over to 2.x is fast approaching, and we are underway now with our migration, but as i say we didn't touch anything at all yesterday so fairly sure this isn't something on our side.

Hi everyone, I am trying to run a job on databricks with

jobs_runs_submit_by_id_and_wait_for_completion

function from the

prefect_databricks

package, but I am getting the below error at the end of flow run although the job in databricks succeeded. It is peculiar that the databricks job has only one task and the error refers to multiple tasks error:

Flow run 'vermilion-manatee' - Created task run 'jobs_runs_get_output-0' for task 'jobs_runs_get_output'
12:37:18.270 | INFO    | Flow run 'vermilion-manatee' - Submitted task run 'jobs_runs_get_output-0' for execution.
12:37:19.736 | ERROR   | Task run 'jobs_runs_get_output-0' - Encountered exception during execution:

raise HTTPStatusError(message, request=request, response=self)
httpx.HTTPStatusError: Client error '400 Bad Request' for url 'https://{removed}.<http://cloud.databricks.com/api/2.1/jobs/runs/get-output?run_id=5681931|cloud.databricks.com/api/2.1/jobs/runs/get-output?run_id=5681931>'

raise httpx.HTTPStatusError(
httpx.HTTPStatusError: A job run with multiple tasks was provided.JSON response: {'error_code': 'INVALID_PARAMETER_VALUE', 'message': 'Retrieving the output of runs with multiple tasks is not supported. Please retrieve the output of each individual task run instead.'}

Does anybody else have this problem using

jobs_runs_submit_by_id_and_wait_for_completion

?

Hello. I see the prefect.io website mentions event-based workflows, but I can't find much in the documentation about it. I recall some blog post by prefect staff at one point which described an attempt at event-driven flows by making use of AWS and/or some other external system, however, the text on the website seems to suggest there is now an out-of-the-box feature for this within prefect. Is this correct? We have a highly event-driven infrastructure so this is a highly sought-after feature for us.

Hi, I want to access a file from github in the flow. I am already using github storage block for flow deployment and reading flow file. Can I do this with current setup or need to perform additional steps like use a github module?

Hi everyone, Is there any way to delete logs using the prefect python api client?

What is the Prefect way to implement SFTP sensor similar to Airflow sensor?

Hello, I am having a problem with a flow run execution. I am trying to execute a flow run using the following API endpoint /api/deployments/{id}/create_flow_run. It is working as expected but when I tried to pass an infrastructure_document_id (which refers to a ECSrun task block) still executing it on local instead of trying to execute on AWS. I don't know if am missing something.

Hey everyone, I'm having some trouble, and I'm not sure what the cause is. I have my cron schedule set in my deployment file, and it is showing up on my UI Dashboard. However, Prefect is not adhering to it. Any ideas what might cause this? I know the time zone is correct, that's about all I can think to check.

Question regarding task scheduling and execution. I’m using

RayTaskScheduler

, but it could be applied to any, I suppose. Assume there is only one worker (node, process, or whatever fits the

Scheduler

class), and the computation graph consists of multiple identical branches (

A -> B -> C

x3). Is there a way to tell Prefect to prefer to go DFS and complete one such branch, before attempting to continue to the other branches? I’m aware of the

wait_for

argument, is there anything else one can do?

Hi everyone! How to you declare a parameter of a deployment which must be chosen from a list of values in the class prefect.deployments.Deployment in Prefect 2.8.3?