I want to allow a task to fail with an exception but instead of the result being returned in a 'raise' the flow just exits and python ends

3 years and maybe I am missing something. If a task fails in a flow, should the whole flow exit? As far as my 100 other flows work, if a task fails, that seems to be OK, as I used allow_failure(failed_task_result) to continue on. This does not make any sense if a task failure exits the flow.

Hey wonder if anyone could help me with a v1 -> v2 migration Q - so in V1 I use the following pattern; • single flow with some param • multiple clocks combined into a single schedule for the flow • each clock has different dafault params How would something similar be achieved in v2?

still struggling in Prefect 2 to dynamically name mapped tasks like in Prefect 1:

# callable name generator for mapped task
def generate_task_run_name_foo(dataset: dict, **kwargs):
    return 'Foo:  ' + dataset['name']


# task that gets mapped in the flow
@task(... , task_run_name=generate_task_run_name_foo)
def foo(dataset: dict, ...):
	...

Prefect 2 doesn’t seem to pass the args of foo into the name generator like v1 does. Is there a different way of achieving this in v2? Prefect 2 transcription leads to

missing 1 required positional argument: 'dataset'

Hello everyone, I'm currently trying out Prefect and I was curious about the development life cycle in a remote environment. I set up an EC2 instance where I started a prefect server, and I can connect to the UI without issue. Now I deployed a flow and I'm trying to run an agent on the same machine to see whether it picks up runs, but the work queue remain unhealthy. I get that the agent doesn't know how to communicate with the api, but I don't really know how to solve this issue. I ran the agent with

--api http://<remote-ip>:4200

or

--api <http://localhost:4200>

or

--api <http://127.0.0.1:4200>

or

--api <http://0.0.0.0:4200>

; the result is the same. If I spin the agent locally while specifying the remote api url, I get the agent to pick up the runs though. Does that mean that, for some reason, it is not possible to run an agent on the same machine as the Prefect server? Is there something I'm missing here.

Hi team, I need some assistance as I am setting up new servers for our organization for prefect flows. Just to give you some context, we already have an Amazon EC2 server that has Prefect flows on it. But, the version of this prefect is 0.13.19 and we need to update this to the latest version. I have now newly setup a new server with prefect 2.9.0 and it is working fine. I am running the command - prefect server start, which starts the UI as expected and I do not see any error in the terminal. But, if I try to access that UI from any other machine/laptop, I do not see anything open on port 4200. We have checked that the inbound and outbound security of the EC2 is not an issue. Any reason why this is happening?

Thank you so much for bringing back the “red green” bars in the latest release. https://github.com/PrefectHQ/prefect/issues/9485 But as the issue is getting more attention, I think there is another fine tune to make for the final form of the “red green” bars. Currently the “red green” bars are shown per FLOW. But this is problematic since I might have several DEPLOYMENTS per single flow, and the “red green” bars fails to reflects this, For example, let’s say I have a FLOW called send_push_notification, which accepts a parameter notification_type. This FLOW has multiple deployments, each has a different CRON setting and different notification_type. So you would have: DEPLOYMENT called send_push_notification_daily_water_remind to run twice a day at 10:00 and 16:00. DEPLOYMENT called send_push_notification_daily_sleep_remind to run each day at 20:00. So I want to be able to look at the “red green” bars for each of those deployments to know if they are running as intended. Currently the “red green” bars are shown for the entire FLOW, and you couldn’t really know how each DEPLOYMENT is doing. I think it’s really small change: Just to move the “red green” bars to the deployments page, and show them per deployment. Please contact me to provide more help and context. Also, please correct me if I am missing something here in my analysis. Thank you!

Hey team, , are there any caveats to watch out for if process type worker is deployed in production ? Coz the docs page says this it typically used for getting started / testing purposes. Thanks!

hello everyone i am new to prefect and i would like your help with this error. i am training a model and tracking the experiments in ml flow run this for deployment prefect deployment build orchestration.py:main_flow --name first deployment --crone 0 9 3 * * got erorr Got unexpected extra arguments (deployment 9 3 data mlflow.db mlruns models orchestrate.py data │ │ mlflow.db mlruns models orchestrate.py)

Hi all another v1 -> v2 question - I am currently installing a module with all my flows in inside a docker container - I would like to accomplish the same thing in v2 - however it seems that this is only possible by copying the source files into /opt/flows - anybody know how I can make the flows available by installing as a module?

Hello, how can I install the extra prefect

collections

for integration? I couldn't find it in docs but it was mentioned in the . Or did I misunderstand something?

Is there value in using the Prefect Great Expectations integrations (https://prefecthq.github.io/prefect-great-expectations/) over just using Great Expectations directly?

Hi. Is it possible to somehow specify the work queue when you start a deployment via

run_deployment

?

I am getting this error only when I run Prefect-Server in the cloud, I am running the server in GCP as a container in a Container Optimized OS

asyncio.exceptions.TimeoutError
04:20:16.970 | WARNING | prefect.server.services.marklateruns - MarkLateRuns took 5.009215 seconds to run, which is longer than its loop interval of 5.0 seconds.
04:20:17.195 | ERROR   | prefect.server.services.failexpiredpauses - Unexpected error in: TimeoutError()
Traceback (most recent call last):

asyncio.exceptions.TimeoutError
07:48:18.910 | WARNING | prefect.server.services.flowrunnotifications - FlowRunNotifications took 5.011953 seconds to run, which is longer than its loop interval of 4 seconds.
07:48:18.917 | ERROR   | prefect.server.services.recentdeploymentsscheduler - Unexpected error in: TimeoutError()

Hi! If want to run the same task for different inputs, and if they all succeed (no exception raised), I want to run a different task later in the flow. What is the appropriate pattern for this? Pseudo-code below of what I want to acheive. I tried something using the return_state=True option but it didnt seem to support concurrency

@flow
def test_flow():
    for i in range(20):
        task_res = test_task.submit(i)
    if all_test_task_succeeded:
       final_task()

How ca I specify a schema in a SQLAlchemy block using asycmpg?

Hello! I have a problem deploying a flow via GitHub Actions. I am using this example https://github.com/PrefectHQ/prefect-recipes/blob/main/devops/github-actions/minimal-prefect-deployment-build.yaml. Everything is working. But when I add the --params argument, the parameters seem to be ignored. At the same time, if I copy the prefect build command and run it on my computer, the --params argument works as it should. What can you advise me?

I am executing a task asynchronously (with the

.submit()

method), the task executes a deployment using the

run_deployment()

method, then the deployment creates a

flow_run

. I need to obtain the final status of the execution of the

flow_run

and that the final status of the task that executes the deployment is the same final status of the execution of the flow_run (subflow). Currently, my task that executes the deployment remains

COMPLETED

regardless of the final state of the execution of the flow_run that was triggered by the deployment.

from prefect.deployments import run_deployment

run_deploy = run_deployment_task.with_options(name=<NAME>).submit(
        deployment_name=,
        environment=environment)
run_deploy.wait()

So if workers do not support storage currently, how can I load my code on them?

I'm getting confused

and nowhere this is documented as far as I know.

Hello, I'm following this guide to connect prefect to Google Cloud Run using github actions. When i follow the guide everything works well, but when copy-paste the actions to my repository i need to update the dockerfile to build it

FROM prefecthq/prefect:2-python3.10-conda

(I need a conda env to install all the deps my code need). My Dockerfile is:

FROM prefecthq/prefect:2-python3.10-conda

COPY environment.yml .
COPY requirements.txt .
COPY setup.cfg .
COPY setup.py .
COPY my-logic .

RUN apt-get update && apt-get install -y libarchive13
RUN conda install -c conda-forge mamba
RUN mamba env update --prefix /opt/conda/envs/prefect -f environment.yml
RUN pip install --upgrade pip setuptools --no-cache-dir
RUN pip install --trusted-host <http://pypi.python.org|pypi.python.org> --no-cache-dir .

ARG PREFECT_API_KEY
ENV PREFECT_API_KEY=$PREFECT_API_KEY

ARG PREFECT_API_URL
ENV PREFECT_API_URL=$PREFECT_API_URL

ENV PYTHONUNBUFFERED True

COPY flows/ /opt/prefect/flows/

ENTRYPOINT ["/bin/bash", "--login", "-c", "prefect agent start -q default"]

The action build goes well, but when i run the deploy from prefect cloud, the VM on GCP creates the job in CloudRun but it returns

/opt/conda/bin/python: Error while finding module specification for 'prefect.engine' (ModuleNotFoundError: No module named 'prefect')

What I'm doing wrong? I still do not understand if I have to change also the

action/deploy-flows/action.yml

to make it create a conda env; in any case I tried both and nothing changed. The modified action.yml is

runs:
  using: "composite"
  steps:
    - uses: actions/checkout@v2
    - name: Set up Python 3.10
    - name: Set up Python
      uses: actions/setup-python@v4
      with:
        python-version: '3.10'
    - name: Install dependencies
      shell: bash
        python-version: "${{ inputs.python_version }}"
        cache: 'pip'

    - id: install-myenv
      run: |
        conda install -c conda-forge mamba
        mamba env update --file environment.yml
        ${{ inputs.install_command }}
      shell: bash

    - id: login-prefect
      run: |

        prefect config set PREFECT_API_KEY=${{ inputs.prefect_api_key }}
        prefect config set PREFECT_API_URL=${{ inputs.prefect_api_url }}
      shell: bash

If i switch my dockerfile back to a non conda image (

FROM prefecthq/prefect:2-python3.10

), everything goes fine, but my deps are not fully covered.

How can I resolve a 403 Error on the Prefect Cloud API? I'm trying to start the prefect agent on the Cloud from our Prefect Runner, but I get the following response. By the way, I upgraded my Prefect package to the latest version. raise PrefectHTTPStatusError.from_httpx_error(exc) from exc.cause prefect.exceptions.PrefectHTTPStatusError: Client error '403 Forbidden' for url 'https://api.prefect.cloud/api/accounts/

Hey team - in prefect 1 we were able to reference codeCommit for the code storage. I'm not seeing the documentation for the same in prefect 2. Is there an equivalent or a workaround for this case?

Hi all, I have an internal package that I would like to have installed alongside my other packages when the infrastructure is set up (I am using the ECS Task from prefect-aws). Typically I just set the environment's

EXTRA_PIP_PACKAGES="package-a package-b"

. My current strategy is submodule the internal dependency and install as editable, but that depends if this variable just being passed to pip as the arguments like

pip install $EXTRA_PIP_PACKAGES

? Then I could set it like

EXTRA_PIP_PACKAGES="package-a package-b -e ./package-c"

. Anybody have insight here?

Hey Team, I'm encountering an issue related to task loops and writing data to a file using the new result handling in Prefect 2.0. Specifically, I'm struggling to write the data to a file in a binary format and would like to retrieve and return the file path associated with the written file. Prefect 1.0 code that works.

hey all? what's the best way to create flow dependencies in prefect 2.0? We used a state handler to do it in prefect 1.0 and we're looking to migrate to 2.0 but scoping out a few different things first. our particular use case is we have a few different flows that extract data from sources and push the data to our data lake, then our redshift cluster. something like elt_1 elt_2 elt_3 once all the different elts are done, we want to trigger our higher level transform flow that preps data for looker. we want to wait till all 3 elts are done and not trigger the final transform until all parents are completed

Can Prefect2 work with Github Storage for the tasks like Prefect1? We've been struggling getting it working - has anyone managed to do this, or should we just abandon it as storage?

Hi all, I'm trying to deploy a flow that uses

prefect-snowflake

, however I received this error message indicating one of the subpackages is missing something:

AttributeError: cffi library '_openssl' has no function, constant or global variable named 'Cryptography_HAS_OPENSSL_CLEANUP'

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/deploy.py", line 80, in <module>
    flow = load_flow_from_script(path=f"{working_dir}/{config['flow_path']}", flow_name=config['flow_name'].replace('_', '-'))
  File "/usr/local/lib/python3.8/site-packages/prefect/flows.py", line 794, in load_flow_from_script
    load_flows_from_script(path),
  File "/usr/local/lib/python3.8/site-packages/prefect/flows.py", line 770, in load_flows_from_script
    return registry_from_script(path).get_instances(Flow)
  File "/usr/local/lib/python3.8/site-packages/prefect/context.py", line 449, in registry_from_script
    load_script_as_module(path)
  File "/usr/local/lib/python3.8/site-packages/prefect/utilities/importtools.py", line 164, in load_script_as_module
    raise ScriptError(user_exc=exc, path=path) from exc
prefect.exceptions.ScriptError: Script at '/github/workspace/etl_trigger/trigger_candidate.py' encountered an exception: AttributeError("cffi library '_openssl' has no function, constant or global variable named 'Cryptography_HAS_OPENSSL_CLEANUP'")

Is anyone else seeing this issue?

What would be the best way to access variables from within a Flow in a State Handler? I'm making a Flow that runs Jupyter notebooks, and on failure I want to post a link to the failed notebook file to slack. There's a task earlier in the flow that generates the filepath.