hi, is there an good way to have exponential delay in a task retry (v1)?

Hi, I'm hitting a recurring error when trying to run DbtShellTask:

AttributeError: module 'prefect.tasks' has no attribute 'dbt'

. I've tried to run this locally and within the jupyter-scipy Docker image. Any thoughts why Prefect cannot find the module?

# dbt_shell_task.py

import sys
import prefect
from prefect import task, flow, get_run_logger


@flow
def dbt_flow(cmd='dbt run'):

    # Execute specified command
    task = prefect.tasks.dbt.dbt.DbtShellTask(
        command=cmd,
        profile_name='default',
        environment='Development',
        dbt_kwargs={'type': 'bigquery'},
        overwrite_profiles=False,
        profiles_dir='/home/jovyan/.dbt/profiles.yml'
    )
    logger = get_run_logger()
    <http://logger.info|logger.info>("Command Run: %s!", name)
    return task

if __name__ == "__main__":
    cmd = sys.argv[1]
    dbt_flow(cmd)

# packages
prefect==2.6.6
prefect-dbt==0.2.4
prefect-shell==0.1.3
dbt-bigquery==1.3.0
dbt-core==1.3.0
dbt-extractor==0.4.1

Is there a best practice for removing data in the orion db older than x amount of time?

Hi! I'm on Prefect 1.2 and use function tasks with

@task

annotation. I have defined two tasks in a flow like this:

@task(name="Backup product catalogue", log_stdout=True)
def backup_task():
    backup_product_catalogue_to_gcs()


@task(name="Update Product Catalogue", log_stdout=True, trigger=triggers.all_finished)
def upsert_pack_products_task():
    upsert_pack_products()


with Flow(
    name="Product catalogue",
    run_config=get_run_config(),
    storage=get_storage("product_catalogue/flow.py"),
) as product_catalogue_flow:
    backup_task()
    upsert_pack_products_task(upstream_tasks=[backup_task])

I renamed the function name

upsert_pack_products_task

and re-deployed. Now I can see two

Backup product catalogue

in my flow on Prefect and they both run (see image) when the flow is triggered. How can I remove that extra task? And how can I debug what is going on?

Can we set a task to success state, ie. based on a condition execute a task else set it to success and trigger the downstream?

Hello all, I'm running a Prefect flow in a Kubernetes cluster via DaskTaskRunner. My task is erroring out with

Crash detected! Execution was interrupted by an unexpected exception.

(full logs in thread). I'm trying to figure out the best way to debug this further. In general – and to help diagnose this particular case – I'd like to shuttle logs to our DataDog instance: I found this discussion on the topic, but is there something more recent / more complete for Prefect v2? Aside from getting logs out, any other ideas for diagnosing this terse error?

Getting this error with a fivetran task, even though I defined it exactly as in the docs (https://github.com/fivetran/prefect-fivetran). Any ideas?

sync = await fivetran_sync_flow(
                ^
SyntaxError: 'await' outside async function

Hey everyone, I seem to be getting an

Orion logging error

with 2.6.6 for our k8 jobs Will post logs in the thread

Hi all! I was curious if there is anything in the works for creating add-ons to the prefect_azure collection for interacting/accessing Azure Key Vaults. I know I can go ahead and create it myself, though I didn't want to reinvent the wheel if it was forthcoming. 🙂

Hello! I have added a slack notification for my flow but when the flow times out I don't get the notification, what state does timed out map to? I have selected all states from the notification menu

who can I talk to with a query regarding prefect-sqlalchemy

Side question: is this still the official Slack for Prefect after moving to 2.0? I was here before but not sure where to ask those questions after we migrate 🙂

When flows crash (according to their logs), Orion UI keeps showing the flows as running, and their run time keeps increasing. If the flow isn't running on the agent, shouldn't it show as "Failed" and not running? Still looking into what actually causes the crash - this is more about how Prefect handles them. We do have flows that runs successful as well, this is just a single flow or two crashing every time.

Hi does anyone know in Prefect 2 why there would be a 30 second delay between the agent picking up a flow run and starting to execute it? Many thanks.

Hi, is there any way to add description of a task in schematic (flow visualisation)

Hi , is there a way to deploy multiple data pipelines(one for daily and other pipeline for weekly etc) , would appreciate the help , thanks in advance

Hi. What I should use instead of wait_for_flow_run(Prefect 1) task in Prefect 2?

In Prefect Live at 3PM Eastern today (Nov 9th), we will host @ben from Club 42. We will talk about how he uses Prefect to coordinate workflows in his college and his experience transitioning from Prefect 1.0 to 2.0. Come join us live on Twitch.

Hi guys, Is there a way to import a function form another file as a task? I was trying something like this:

from tasks.date import get_start_date
@task(name='Generate start date', retries=2, retry_delay_seconds=10)(get_start_date)

It fails with `RuntimeError: Tasks cannot be run outside of a flow. To call the underlying task function outside of a flow use

task.fn()

.` Thanks!

There seems to be a problem with Prefect v1 where new schedules are not being added to flows. Is there an outage on our side?

Can anyone point me towards documentation on setting up a profile to attach to the cloud API from your local machine? Thank you

What's the best way to define dependencies in 2.0?

Hello! I am using Prefect 1.x and store my flows using the Git storage option. I’m trying to “load additional files” from the repo using the steps from the storage docs under Loading Additional Files with Git Storage. However, the file I am looking for isn’t on the file system and I get a “FileNotFoundError: [Errno 2] No such file or directory” error. Looking at the source code for Prefect 1.4.0, the Git storage implementation uses a ‘TemporaryGitRepo’ object to get the flow definition - see line 148 here https://github.com/PrefectHQ/prefect/blob/1.4.0/src/prefect/storage/git.py#L148) which seems like it deletes the repo contents from the file system after reading the flow definition file. I think I’m following the docs correctly, but I might be missing something. I’d appreciate any help and/or suggestions on what to try. Thanks in advance!

Hi! What’s the best way to see error logs or otherwise debug an MS Teams notification? Running Prefect 2.6.4 and I see notification webhooks coming across for flow

Running

states but not for

Failed

— details in thread:

I was able to setup a prefect agent in kubernetes (GKE), and now I'm a little confused on next steps. If you are to create a flow deployment, does that code deployment need to be in a storage block? I'm assuming it will have to be. I tried a flow run without the storage block and got an error saying it couldn't find the flow (and it referenced my local computer's path).

Also, is it possible to keep the agent separate from all the code dependencies by running flows in Docker containers and building all the dependencies within the Docker container? If so, is there a tutorial fo rthis?

Returning Manual State from a Task I want to run a scheduled flow that deals with an intermittent service, and I want to handle the exceptions gracefully. The common idiom seems to be raising

ValueError

to trigger a failure, and thats how to manage the state of subsequent tasks or flows. Here's the thing: I don't want to see a stack trace unless it is some new failure that I need to troubleshoot. For the expected exceptions I want to post an error the log and proceed with my flow according to logic I design. I'm having trouble returning a manual state of Cancelled, concrete example and logs posted in thread. Can anyone guide me toward causing the entire flow to be cancelled without raising an exception?

is there a way to execute long sql queries from a .sql file with sqlalchemy_execute ?

Can we get a /health + /stats endpoint on an agent so we can monitor and cycle if needed