Good day! I'm trying to run a flow using azure container instance (ACI) and I'm getting a strange missing context error that I never bumped into during my docker and kubernetes days. Has anyone seen this before?

Hi there, I have a prefect orion server running in kubernetes and I am trying to deploy some flows on it from my local machine. When I run:

❯ PREFECT_API_URL=<https://prefect.xxx.xxx/api> prefect deployment ls

I am getting:

File "/home/patrick/miniconda3/envs/prefect/lib/python3.8/ssl.py", line 944, in do_handshake
    self._sslobj.do_handshake()
ssl.SSLCertVerificationError: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1131)
An exception occurred.

• I've tried set envs to disable certificate verification: PYTHONHTTPSVERIFY=false • Tried to add the server certificate (*.xxx.xxx.crt) on the server copying the CRT file

/usr/local/share/ca-certificates/

and update the CA store:

sudo update-ca-certificates

Nothing works. Any tip to solve this?

Hey all,Can you run Prefect 1.0 agents on the same machine as Prefect 2.0 Agents (While migrating from 1.0 to 2.0) or do they need their own machines?

Hi everyone! 👋 S Today we are adding 11 new channels to our Slack Community to open up conversations and discussions relating not only Prefect-specific topics but data engineering and data science. We want this Community to be a place where you can collaborate and get help with your data regardless of the stack or tool you are using. We can all share tips, tricks, and learn together. We have new cloud-specific channels for #prefect-aws, #prefect-gcp, #prefect-azure Channels dedicated to #prefect-kubernetes, #prefect-docker, #prefect-dbt, and #prefect-cloud New channels for #prefect-integrations, #prefect-recipes and #prefect-getting-started that will have live stream events for Build-a-Block or Code with Marvin. Lastly, we are so grateful for all of you in our Community that we wanted to create a #gratitude channel where we can all share our thanks for one another. So make sure to check out the new channels, join the conversations, and look for new events and announcements. tl;dr - 11 new channels in slack and lots of new events or watch the video 😉 📺 https://www.loom.com/share/5ad9da90d6ab4beeadcbd7de6ef7a527

Hello, is there a way to prevent deployments from overriding whether or not a flow is enabled in Prefect 1? In this case, I have deployed flows that have been turned off in the UI that will automatically be turned on again the next time I deploy changes even for something unrelated. This means having to remember to turn a flow off again after every deployment which can be annoying and create issues if a flow runs by accident when it shouldn't.

Hey, I have a flow that is triggered by an API call from an external process. Sometimes, the process fails, and the API call doesn’t send, meaning the flow is not triggered. Is there a way to monitor those delays? I mean - If the flow doesn’t triggered by 8 am, is there a way to monitor it and send a Slack notification or an email? Thanks. @Taylor Curran

Hi!

I'm trying to use Prefect 2.6.5 to run a Flow locally. The Flow has >10 Tasks and quite a bit of concurrency (>1000). It's an async Flow and all Tasks are async although several use

<http://anyio.to|anyio.to>_thread

internally. I'm using the ConcurrentTaskRunner. The flow is ran like this:

anyio.run(
        generate_report,
        datetime.date.fromisoformat(run_date),
        backend_options={"debug": True, "use_uvloop": True},
    )

The problem I' having is that some Tasks finish but they remain in state Running forever. There's no output I could see to understand the problem. Help, please.

Hello, is there a way to 'pause' a task depending on the outcome of some runtime process? I don't want the task to retry except in this one specific runtime result (normally the task could just fail and retry according to args) Perhaps I could return a manual state

AwaitingRetry

?

@task
    # some code, oh found out a blocker I'd like to wait for
     time_2_hours_from_now=  .... some date function...
     if function_is_blocked:
           return AwaitingRetry(scheduled_time=time_2_hours_from_now

hey all, is there a way to dynamically name flows in v1?

with prefect.Flow(
    name="SOME_DYNAMIC_NAME" # instead of hardcoding this, i want to pass a value when registering / calling it

Hi, for prefect 2 self hosted where would I find logs around notifications being fired?

Hi. How to see RRuleSchedule in UI? I just see 'Every year' on Deployment tab and it is not clear when scheduled flow runs will appear and how many?

can you pass a task as a flow parameter in prefect 1?

How can I fail a task? I am try/catching exceptions, which works file, but I still want the task to fail. The docs didn't quite show this example:

@task
def fail_task():
    return Failed()

@flow
def fail_flow_i_hope():
    fail_task()

fail_flow_i_hope()

11:42:32.292 | DEBUG   | Task run 'always_fail_task-d83fd752-0' - Beginning execution...
11:42:32.302 | INFO    | Task run 'always_fail_task-d83fd752-0' - Finished in state Completed()
...
11:42:32.313 | INFO    | Flow run 'ludicrous-lemur' - Finished in state Completed('All states completed.')

I'd like a task to fail, and then do retries (or not) but end the steps in the flow, with the flow also failed.

Traceback (most recent call last):
  File "/Users/jonyoung/Dev/workflows/.venv/lib/python3.9/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "/Users/jonyoung/Dev/workflows/.venv/lib/python3.9/site-packages/prefect/engine/flow_runner.py", line 569, in get_flow_run_state
    executors.prepare_upstream_states_for_mapping(
  File "/Users/jonyoung/Dev/workflows/.venv/lib/python3.9/site-packages/prefect/utilities/executors.py", line 682, in prepare_upstream_states_for_mapping
    value = upstream_state.result[i]
KeyError: 0
└── 17:51:25 | ERROR   | Unexpected error occured in FlowRunner: KeyError(0)

any help what this error means?

I have multiple deployments in a work queue with unlimited concurrency, but they still run sequentially and one will not start until the other is finished. Am I misunderstanding how concurrency works and it only applies to the same flow? Do I need a separate work queue for each of my deployments? Because what has been happening is we occasionally have a flow that takes a really long time, and all other flows wait until the long-running one finishes, with any missed jobs showing up as late and then running once the long one finishes. Two issues with that: having time-sensitive jobs not run because they are waiting, and then having hundreds of late jobs in the queue and no way to get rid of them other than deleting one by one

Hi, I have a problem while editing any block from Prefect UI. if the field value matches the default value it'll not set and the value will not be sent on the PATCH request making it impossible to edit it back to the default one

Hi Community - me again. I have a strange one, I have set up an agent in ECS with a custom image that has

prefect-aws

deps installed. I have an infrastructure

ECSTask

block that pulls a custom image:

FROM        prefecthq/prefect:2.6.5-python3.9

RUN         pip install prefect-aws s3fs==2022.5.0

This installs everything I need to run a demo flow that is:

import prefect
from prefect import flow, task, get_run_logger
from utilities import AN_IMPORTED_MESSAGE

from prefect_aws.ecs import ECSTask

ecs_task_block = ECSTask.load("staging-test")


@task
def log_task():
    logger = get_run_logger()
    <http://logger.info|logger.info>("Hello %s!", "BEN")
    <http://logger.info|logger.info>("Prefect Version = %s 🚀", prefect.__version__)
    logger.debug(AN_IMPORTED_MESSAGE)


@flow
def my_test_flow(name: str):
    log_task(name)


if __name__ == "__main__":
    my_test_flow()

This flow is using S3 Block Storage, now when I trigger this flow from the cloud UI the task is generated in my cluster, it just fails with a

ModuleNotFoundError: No module named 'prefect_aws.ecs'

More details in 🧵

hi, my agent log always got httpx error, but seems like the flows or deployments are still work fine, how to deal with this?

Hey - say I have some CI process and it deploys a flow, its infra and storage on every deployment. Then I change the name of one of the deployments, or the flow name ( due to some new requirements ). Is there any recommended pattern besides manually cleaning everything up on the UI to delete the old resources that will continue to live in the workspace?

Hi, Is there Prefect Task that works with Gmail? Read email, download attachment?

Hi all, I can't find where the error occurred. I'm using prefect 1.0. As I looked for examples to set the dependency between flows, I saw an article written by anna geller. (Link) Some time has passed since she wrote this article, so I changed

FlowRunTask

to

StartFlowRun

(reference link) among the methods written in that article to create the same situation. Below attached scripts, it is the case that two flows(flow_waveforms and flow_pipeline) that are executed well have already been created, and parent_flow is created to sequentially execute those flows.

# parent_flow.py
import prefect
from prefect.tasks.prefect.flow_run import StartFlowRun

with Flow('mother_flow') as flow:
    first_job = StartFlowRun(flow_name='flow_waveforms',
                             project_name='InterFlow_Dependencies',
                             wait=True)
    second_job = StartFlowRun(flow_name='flow_pipeline',
                              project_name='InterFlow_Dependencies',
                              wait=True)
    first_job.set_downstream(second_job)

if __name__ == '__main__':
    flow.register("InterFlow_Dependencies")

# flow_waveforms.py
import prefect
from prefect import Flow, task

@task
def task1():
    something
    return one, two
@task
def task2():
    something

with Flow('flow_waveforms') as flow:
    var1, var2 = task1()
    finish = task2()

flow.run_config = UniversalRun(env={'PREFECT__CLOUD__HEARTBEAT_MODE':'thread'})
flow.register('InterFlow_Dependencies')

# flow_pipeline.py
import prefect
from prefect import Flow, task

@task
def task3():
    something

with Flow('flow_pipelines') as flow:
    task3()

flow.register('InterFlow_Dependencies')

If i excuted mother_flow.py in this case, prefect responds following error.(image) Any idea where I went wrong? Or could you tell me if there is an easy way to define dependencies between flows other than this way? Thx.

Hi, I have a pretty simple issue that I can't seem to find any documentation about. I set up an

orion

+

agent

container setup locally, and everything is working as expected when I try to run the flows manually. I was able to successfully create and apply a deployment as well (from within the

cli

container). However, when trying to run the deployment I am faced with a

File not found error

. Here's the traceback:

Flow could not be retrieved from deployment.
Traceback (most recent call last):
  File "<frozen importlib._bootstrap_external>", line 879, in exec_module
  File "<frozen importlib._bootstrap_external>", line 1016, in get_code
  File "<frozen importlib._bootstrap_external>", line 1073, in get_data
FileNotFoundError: [Errno 2] No such file or directory: '/tmp/tmp_zx6qpglprefect/flows/fetch_etfs.py'

My

docker-compose

is as follows:

version: "3.9"
services:
  db:
    image: timescale/timescaledb:latest-pg14
    volumes:
      - $PWD/data/db:/var/lib/postgresql/data
      - $PWD/config/db:/docker-entrypoint-initdb.d/
    healthcheck:
      test: [ "CMD-SHELL", "pg_isready" ]
      interval: 10s
      timeout: 5s
      retries: 5
    environment:
      - POSTGRES_DB=$POSTGRES_DB
      - POSTGRES_USER=$POSTGRES_USER
      - POSTGRES_PASSWORD=$POSTGRES_PASSWORD
    networks:
      - db_network
  orion:
    image: prefecthq/prefect:2.6.3-python3.10
    restart: always
    volumes:
      - $PWD/prefect:/root/.prefect
    entrypoint: [ "prefect", "orion", "start" ]
    environment:
      - PREFECT_ORION_API_HOST=0.0.0.0
      - PREFECT_ORION_DATABASE_CONNECTION_URL=$PREFECT_DB_URL
    ports:
      - 4200:4200
    depends_on:
      - db
    networks:
      - prefect_network
      - db_network

  agent:
    build: ./prefect
    restart: always
    entrypoint: [ "prefect", "agent", "start", "-q", "main_queue" ]
    environment:
      - PREFECT_API_URL=<http://orion:4200/api>
    networks:
      - prefect_network
      - db_network

  cli:
    build: ./prefect
    entrypoint: "bash"
    working_dir: "/root/flows"
    volumes:
      - "$PWD/prefect:/root"
    environment:
      - PREFECT_API_URL=<http://orion:4200/api>
    networks:
      - prefect_network
      - db_network

networks:
  db_network:
  web_network:
  prefect_network:

To create and apply the deployment(s):

> docker compose run cli
$ prefect deployment create ./flows/fetch_etf.py:flow_name --name flow_name
$ prefect deployment apply deployment.yaml

I've been re-deploying some flows this morning and I have been successful in re-deploying flows until about 15 minutes ago when I received this error. I have not changed anything prefect-wise in my code, is this a server error?

Hi, is there a Prefect integration with GitHub that could show the Status of my flow on my PR so that reviewers don’t need to login to Prefect to see that my flow succeeded?

Hello is there a way to send an email or slack alert if a flow fails to run?

One of my flows is struggling to process: agent is on AKS and the CPU usage, through lens, shows a value of 1 for about 6 hours before dropping to negligible amounts (0.002, etc) while still maintaining high levels of RAM usage (30-40gb). At first I assumed it was stalling, but looking further it seems that the flow is only using 1/8 cores at 100%. This explains the continued RAM use. Any suggestions on how to utilize the full compute power of the aks vm? I'm assuming DaskTaskRunner is the route to go, and I'm currently experimenting with it, but not having much luck with configurations.

i have a flow that i want to dynamically name, schedule, and pass parameters to, how do i do that?

I'm testing flows locally during development by simply typing "py .\flow.py" in powershell, and the results are inadvertently getting communicated up to my prefect cloud UI without an agent running. How is this possible?