Hey, I have a flow running using Vertex Run - this is my run config -

flow.run_config = VertexRun(scheduling={'timeout': '3600s'},
                            machine_type='n2-highcpu-80', labels=["ml"],
                            service_account=PREFECT_SERVICRE_ACCOUNT)

It works without the scheduling parameter - I added it and used this documentation - https://docs.prefect.io/orchestration/flow_config/run_configs.html#vertexrun Now when registering to Prefect cloud and running I get this error -

Parameter to MergeFrom() must be instance of same class: expected google.protobuf.Duration got str.

Am I missing something? Thanks

Hello! I have a flow that uses DateTimeParameters, I would like to schedule it with one date being default to the day of the schedule, or first day of the month for example, is there a way to do it through schedules or should I handle this in my code ?

hello

Hi, I have chained tasks that are simply database operations, they don't return anything but they have to be run in a specific order. To this day I've simply used a boolean, I

return True

and pass it, unused, to the next task. Is that ok or did I miss a way to do that more "beautifully" in the docs ? (I don't find it particularly bad, + with the fact that I suffix it with "_done" I think it's explicit, but was still wondering)

t1_done = task1()
t2_done = task2(t1_done)
....

Hi all! It's our first ever launch week and to celebrate, in case you don't following the #announcements channel....we're giving away free pizza starting at 12p Eastern! https://prefect-community.slack.com/archives/CKNSX5WG3/p1647264267554719

Hi, Is it possible to have a callback function which is called every time a flow changes its state which runs on the system where I kick off the flow rather than on the system actually running the flow? I tried

Flow(..., state_handlers=[…])

but the callback is called on the system where the flow is executed. My idea is to use

StartFlowRun

e.g. on my local computer which starts the flow in AWS cloud. And every time the state changes (

submitted

,

running

,

canceled

, etc.) a local callback function is called where I see the

state

and

flow_run_id

.

Hi, I've what is probably a fairly basic question, but I can't find the right sequence of words to get an answer in the docs. Say I have a flow that takes 23 hours to run, and I kick it off at Midnight each night. Is it possible to block today's run if yesterdays is still running?

Free pizza is live! Get it while it's hot, just reply with your favorite open source project 🍕 https://twitter.com/PrefectIO/status/1503397645674504192

Apart from providing Pizza, we've also just launched this beautiful Community website http://prefect.community By far, the best part of it is the page featuring Top Prefectionists: https://www.prefect.io/community/top-contributors/ Kudos to all contributors! Thanks for being part of our community 🚀

FYI - two live events for Launch Week! Will we see you there? https://prefect-community.slack.com/archives/C036FRC4KMW/p1647289203959699

Hello everyone, how do you manage security requirements in Prefect core opensource version such as UI/API encrypted acces, authn/z etc..? is there any workaround applied by anyone? It seems cloud version solves all these problems. is there any way to handle these security requirements using in on prem?

Hello, I am trying to add slack notifications for a prefect flow using this: https://docs.prefect.io/core/advanced_tutorials/slack-notifications.html#installation-instructions I have a slack webhook URL that I've saved as a prefect secret. I am importing this:

from prefect.utilities.notifications import slack_notifier

and my tasks now look like this :

task(log_stdout=True, state_handlers=[slack_notifier])

install_snowflake_task = ShellTask(helper_script="pip install boto3 \
    snowflake-connector-python[pandas] \
    snowflake-ingest && pip install PyJWT==1.7.1", shell="bash", stream_output=True, return_all=True, state_handlers=[slack_notifier])

install_dbt_task = ShellTask(helper_script="pip install dbt==0.18.0", \
                             shell="bash", stream_output=True, return_all=True, state_handlers=[slack_notifier])

My flow looks like

with Flow("name-of-flow", state_handlers=[slack_notifier]) as flow:

The flow runs successfully but I see no notifications on the slack channel that I have authorized for this. What am I doing wrong?

Quick Orion question: is it possible to set the name of a task at run-time of a flow?

💃 Just arrived!

question: Let's say I have 2 separate repos where prefect lives...can I post a 'wait for done" flag type object (similar to airflow) that sends a signal to the 2nd Flow that it can be started?

Hi guys, I have a question regarding Prefect result. When the result is uploaded to Azure Blob Storage, does any encryption happen between transit? If yes, which encryption standard is that?

Can we create a task which runs before the flow ends, when all upstream task are successful. (without mentioning all upstream tasks name)

Hello Community, I am running into a snag and I was wondering if anyone had any insight on what is going on. It appears my flow is getting stuck in the

Submitted for execution:

phase. I am using

Docker

storage (The image is hosted in ECR and I’ve confirmed my execution role has access to the repo) and

ECS Run

. The flow runs locally but not in ECS. Dockerfile:

FROM prefecthq/prefect:latest-python3.8
WORKDIR /opt/prefect
COPY requirements.txt .
RUN pip install --upgrade pip && \
    pip install --no-cache-dir -r requirements.txt

Flow code:

import os
from prefect import Flow, task
from prefect.run_configs import ECSRun
from prefect.storage import Docker


@task(log_stdout=True)
def extract():
    x = [4, 5, 6]
    print("Starting: {}".format(x))
    return x


@task
def transform(y):
    return [i * 10 for i in y]


@task(log_stdout=True)
def load(z):
    print("Received: {}".format(z))


with Flow("Test Flow", storage=Docker(dockerfile="Dockerfile", registry_url=os.getenv("REGISTRY_URL"), image_name=os.getenv("IMAGE_NAME"))) as flow:
    e = extract()
    t = transform(e)
    l = load(t)

flow.run_config = ECSRun(
    task_role_arn=os.getenv("TASK_ROLE_ARN"),
    execution_role_arn=os.getenv("EXECUTION_ROLE_ARN")
)

Hi Folks - I have a flow that uses the Prefect provided

RenameFlowRun

. Would it be expected that in a downstream task if I call

prefect.context.get("flow_run_name")

the new/updated name is returned?

How to call Prefect Python APIs within Prefect Orion? Also, how to call Prefect Orion Python APIs from a Java application? Do we have any tutorial or doc for both cases?

Hi folks, is Prefect SOC 2 compliant?

Hey all! I am working on more complex flow tasks that require external libraries like pandas and scikit-learn. I've had trouble deploying before due to pickling not supporting all libraries. Do you have any general guidance, or a list of "known pickleable" libraries?

Is there an easy way to get my secrets from local .env file into Prefect Cloud?

Hi Everyone, I'm new to prefect and I'm running into the following issue. I defined a task and inside a flow block I call this task. It happens that I expect the return of this task to be a list of urls(variable "extratos"), but don't know why I'm getting the following error message when "len(arquivos_extrato" is run: "object of type 'GetItem' has no len()". sharepoint4 is a customized class that i have built and the function listar_arquivos() is part of this class. My code is something like this: Can someone help me to understand why am I getting a GetItem object and not a list ?

Okay so I have my container built on azure container registry, but Docker() storage refuses to connect. How do I choose the container in Azure Container registry?

Hi folks, is it possible to call a task inside another task? Something like this: def function_a(): c = a + b return c task def do_something(): y = function(a) return y

so I have my flow in one file and then the tasks in another file. I finally got them to play nicely with Docker storage, but now in the RUN python /opt/prefect/healthcheck.py, it can't find the file that is in the same folder as the flow.

System Version check: OK
/opt/prefect/healthcheck.py:130: UserWarning: Flow uses module which is not importable. Refer to documentation on how to import custom modules <https://docs.prefect.io/api/latest/storage.html#docker>
  flows = cloudpickle_deserialization_check(flow_file_paths)
Traceback (most recent call last):
  File "/opt/prefect/healthcheck.py", line 130, in <module>
    flows = cloudpickle_deserialization_check(flow_file_paths)
  File "/opt/prefect/healthcheck.py", line 43, in cloudpickle_deserialization_check
    flows.append(cloudpickle.loads(flow_bytes))
ModuleNotFoundError: No module named "Module"

hey team - orion 2.0b1 is throwing some typer errors on success (I can see they are succeeding but it looks potentially misleading)

Set default storage to 'local'.
Traceback (most recent call last):
  File "/home/brad/.local/pipx/venvs/prefect/lib/python3.9/site-packages/prefect/cli/base.py", line 58, in wrapper
    return fn(*args, **kwargs)
  File "/home/brad/.local/pipx/venvs/prefect/lib/python3.9/site-packages/prefect/utilities/asyncio.py", line 120, in wrapper
    return run_async_in_new_loop(async_fn, *args, **kwargs)
  File "/home/brad/.local/pipx/venvs/prefect/lib/python3.9/site-packages/prefect/utilities/asyncio.py", line 67, in run_async_in_new_loop
    return anyio.run(partial(__fn, *args, **kwargs))
  File "/home/brad/.local/pipx/venvs/prefect/lib/python3.9/site-packages/anyio/_core/_eventloop.py", line 56, in run
    return asynclib.run(func, *args, **backend_options)
  File "/home/brad/.local/pipx/venvs/prefect/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 233, in run
    return native_run(wrapper(), debug=debug)
  File "/usr/lib/python3.9/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/usr/lib/python3.9/asyncio/base_events.py", line 642, in run_until_complete
    return future.result()
  File "/home/brad/.local/pipx/venvs/prefect/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 228, in wrapper
    return await func(*args)
  File "/home/brad/.local/pipx/venvs/prefect/lib/python3.9/site-packages/prefect/cli/storage.py", line 136, in create
    exit_with_success(f"Set default storage to {name!r}.")
  File "/home/brad/.local/pipx/venvs/prefect/lib/python3.9/site-packages/prefect/cli/base.py", line 193, in exit_with_success
    raise typer.Exit(0)
click.exceptions.Exit: 0
An exception occurred.

Orion Question - is there a way to see flow schematic in Orion UI outside of flow run like Prefect 1.0 ?