Hey guys, has anyone ever used Prefect to manage AWS DMS tasks before?

Hi everyone ! Since yesterday, all my Airbyte tasks are failing returning this kind of errors :

######## ERROR 1  
<...>
  File "/root/.local/lib/python3.9/site-packages/prefect/tasks/airbyte/airbyte.py", line 345, in run
    job_id, job_created_at = self._trigger_manual_sync_connection(
TypeError: cannot unpack non-iterable NoneType object 

######## ERROR 2 
<...>
  File "/root/.local/lib/python3.8/site-packages/prefect/tasks/airbyte/airbyte.py", line 333, in run
    job_status, job_created_at, job_updated_at = self._get_job_status(
TypeError: cannot unpack non-iterable NoneType object

The task failed even if the Airbyte sync is correctly launched. That’s odd because I did not make any modifications since yesterday on either P Prefect or Airbyte. I first encountered this issue with prefect core v1.1.0, and I face the same behavior after upgrading my core version to v1.3.0. Have other people experienced this behavior ?

Hey folks, What’s the expected behaviour when there’s a cleanup task waiting on a set of tasks and one of the upstream task fails? How does the overall state of the flow get resolved? Code below to clarify this.

@flow
def test_flow(nums: list = None):
    test_futures = [square.submit(num) for num in nums]
    return do_cleanup.submit(wait_for=test_futures)

In the above does do_cleanup get invoked or does the flow directly go into a failed state?

Hello, When we build a deployment, the deployment gets the name of either the function that starts the flow, or the flow name argument in the function itself. But it doesn't take the deployment name itself that we specify with the

--name

argument, and there isn't an argument to specify the flow name in the CLI (unless we do it by hand after the creation of the file). I believe that's not a good behaviour since it complicates managing lot of flows. For instance let's assume we have this structure:

- Flows
--| - flow_1.py
--| - flow_2.py

If both files look like this

@task
def test_task():
   return 'hello'
@flow
def main():
   message = test_task()
   print(message)
if __name__= '__main__':
  main()

The flow name in the yaml file will be the same (main), and that will put confusion on the prefect UI. As we'll have the same flow but with different deployment names, but it's not the same flow code under the hood. Is there a better way to handle this apart from changing the name after the creation of the yaml file ?

Hi, does this look familiar?

Failed to load and execute flow run: ModuleNotFoundError("No module named '/root/'")

I am running the Prefect Agent on a separate machine than the Prefect Server, but the communication works (The Flow starts on the Agent) Thanks!

Hi 👋 I cannot access the logs of any job since this morning ...

Hi Prefect team, I can't seem to figure out how to override the customizations using the cli, I always end up with an error. Is it even possible to override the customizations using the cli?

deployment build --override customizations = resource_object

Hi there, can I run a async task like

prefect-dbt

's

trigger_dbt_cli_command()

inside another task? This does not work for

async

tasks. Thanks

okay, here is an interesting question, I have the following example flow

@task
def required():
    return [
        1,
        2,
        3,
    ]


@task
def optional_case():
    return True


@task
def optional(data):
    data.append(4)
    return data


@task
def report_case():
    return False


@task(skip_on_upstream_skip=False)
def report():
    print('all done!')


with Flow(
    'test',
    executor=LocalDaskExecutor(),
) as flow:

    data = required()

    with case(optional_case, True):
        opt_data = optional(data)

    with case(report_case, True):
        report(upstream_tasks=(data, opt_data,))

The idea is there there are 2 variables, summarized as: • variable 1: should optional data processing be run • variable 2: should reporting be run at the end (note that reporting does not take data as an explicit input) I am trying to achieve this in prefect one using

case

statements. The issue is that I need to set

skip_on_upstream_skip

on report, so that it runs even if optional data processing is not run. I just want to ensure that report is run if its case is True, otherwise it is not run, but it must only be run AFTER

data

and

optional

have run

Hi! I want to set the runtime name of a task depending on one of the arguments, without having to set name on every function call. Basically what is the Prefect 2 equivalent of the prefect 1 code below

from prefect import task, Flow


def foo(x):
    return _foo(x, task_args=dict(name=x))


@task(task_run_name="{x}")
def _foo(x):
    print(x)


with Flow("flow") as flow:
    foo('a') # should be named 'a' 
    foo('b') # should be named 'b'

flow()

Hi All, I have a question about setting upstream tasks. Here’s some pseudo code:

@task()
def task1():
	# do stuff 

@task()
def task2():
	# do other stuff 

ids = ['aaa', 'bbb', 'ccc'] 

for id in ids:
	x = task1()

y = task2(upstream_tasks=[x])  

flow.run(executor=LocalDaskExecutor)

My problem is that

task2()

fires when the first instance of

task1()

completes successfully - I need

task2()

to fire only once all of

task1()

instances have completed - what am I doing wrong here?

If there’s a task failure in an upstream task and I am waiting on it to complete (using

wait_for

), it’s never triggered, rather goes into an undocumented

NotReady

state. To overcome this, I have to set

.result(raise_on_failure=False)

on the upstream task. See images below for states. See screengrabs 👇 - left fails, right works. This seems like a common DAG use-case which should be handled well by prefect, how do we wait on a set of upstream tasks to complete and then perform a cleanup / notification at the end (with an aggregate of results from all upstream tasks).

It would be nice if the schedule didn't wipe every time you re-apply a deployment

Hey everyone! Trying to figure something out - in Prefect 1, is there a way to restart flow runs from the API? I am able to cancel flows via the API using the

cancel_flow_run

function but I don’t see a way to restart the run once cancelled

Error downloading Flow from S3: An error occurred (AccessDenied) when calling the GetObject operation: Access Denied

I keep getting this error when flows are supposed to download their definitions from storage. I have both a prefect user and prefect role specified in the bucket's policy.

I have a question about dynamically generating tasks. 🧵

Anyone else experiencing an error trying to pip install the latest prefect release (2.3.2)? https://github.com/PrefectHQ/prefect/blob/main/RELEASE-NOTES.md#232. Maybe not available yet?

pip install "prefect==2.3.2"

ERROR: Could not find a version that satisfies the requirement prefect==2.3.2 (from versions: 0.5.0, 0.5.1, 0.5.2, 0.5.3, 0.5.4, 0.5.5, 0.6.0, 0.6.1, 0.6.2, 0.6.3, 0.6.4, 0.6.5, 0.6.6, 0.6.7, 0.7.0, 0.7.1, 0.7.2, 0.7.3, 0.8.0, 0.8.1, 0.9.0, 0.9.1, 0.9.2, 0.9.3, 0.9.4, 0.9.5, 0.9.6, 0.9.7, 0.9.8, 0.10.0, 0.10.1, 0.10.2, 0.10.3, 0.10.4, 0.10.5, 0.10.6, 0.10.7, 0.11.0, 0.11.1, 0.11.2, 0.11.3, 0.11.4, 0.11.5, 0.12.0, 0.12.1, 0.12.2, 0.12.3, 0.12.4, 0.12.5, 0.12.6, 0.13.0, 0.13.1, 0.13.2, 0.13.3, 0.13.4, 0.13.5, 0.13.6, 0.13.7, 0.13.8, 0.13.9, 0.13.10, 0.13.11, 0.13.12, 0.13.13, 0.13.14, 0.13.15, 0.13.16, 0.13.17, 0.13.18, 0.13.19, 0.14.0, 0.14.1, 0.14.2, 0.14.3, 0.14.4, 0.14.5, 0.14.6, 0.14.7, 0.14.8, 0.14.9, 0.14.10, 0.14.11, 0.14.12, 0.14.13, 0.14.14, 0.14.15, 0.14.16, 0.14.17, 0.14.18, 0.14.19, 0.14.20, 0.14.21, 0.14.22, 0.15.0, 0.15.1, 0.15.2, 0.15.3, 0.15.4, 0.15.5, 0.15.6, 0.15.7, 0.15.8, 0.15.9, 0.15.10, 0.15.11, 0.15.12, 0.15.13, 1.0rc1, 1.0.0, 1.1.0, 1.2.0, 1.2.1, 1.2.2, 1.2.3, 1.2.4, 1.3.0, 2.0a1, 2.0a2, 2.0a3, 2.0a4, 2.0a5, 2.0a6, 2.0a7, 2.0a8, 2.0a9, 2.0a10, 2.0a11, 2.0a12, 2.0a13, 2.0b1, 2.0b2, 2.0b3, 2.0b4, 2.0b5, 2.0b6, 2.0b7, 2.0b8, 2.0b9, 2.0b10, 2.0b11, 2.0b12, 2.0b13, 2.0b14, 2.0b15, 2.0b16, 2.0.0, 2.0.1, 2.0.2, 2.0.3, 2.0.4, 2.1.0, 2.1.1, 2.2.0, 2.3.0, 2.3.1)

ERROR: No matching distribution found for prefect==2.3.2

it looks like

fastapi

has a way to override the default

operationId

which helps give endpoints better fn names. any chance orion could implement this? as they’re currently implemented, the operation ids (though descriptive) are pretty bad from a usability-perspective 😕

is anyone else having issues with app.prefect.cloud? no workspaces showing up - 422 errors in the console - also getting 422 errors from the api

Have issues and flows are now late

yeah I think something is down

yeah; could anyone from Prefect confirm?

I submitted a ticket via their website chat

@Jeff Hale sorry for tagging you out of the blue, just saw your posts and was looking for someone from Prefect who is online to check if the API is up on Prefect end

There is something wrong with the release, we are trying to rollback now

Got it, thank you so much! Appreciate it

thanks!

To clarify, the Prefect 2.3.2 release is fine, but Prefect Cloud was also being released and that release is being rolled back at the moment.

The issue with Prefect Cloud should be fixed now

thanks team!