Thank you very much in advance!

Hello I get 'Failed to submit notification' message when trying to create a notification on the orion server. I am able to create notifications on Prefect Cloud without any issues though. Any help would be highly appreciated. Thanks.

I am trying to use Prefect REST API to create flow run for deployment but it's giving 403 and an error message saying that "Not authenticated" in response. What should I add for authentication ? Could any one help me on this? Thanks in Advance.

Hi all, really enjoying Prefect so far. Much easier than airflow and much better suited to my requirements at work. I have a number of pipelines which load from Postgres into BigQuery, performing standard transformations along the way. What I've done is create a flow which loads a given JSON file from a bucket and then loads that table with the settings given in the JSON file. This flow can be run on its own with a parameter specifying the JSON filename. I have another flow which loops through the bucket and launches the single-table subflow for each file in the bucket, so I can schedule all the tables I need on a daily basis. It all works fine and I'm really happy with this. However, each subflow has a generic "active-pig", "tested-puma" etc name. Is there any way that I can set the names of these subflows in the python task decorator? It would give me improved visibility of which tables have been run.

Getting an error from AWS when trying to apply deployment with S3 storage block.

from flows.helloworld.helloworld_flow import helloworld
from prefect.deployments import Deployment
from prefect_aws.ecs import ECSTask
from prefect_aws.s3 import S3Bucket

s3_storage = S3Bucket.load("prod")

ecs_task = ECSTask.load("prod")

helloworld_deployment = Deployment.build_from_flow(
    flow=helloworld,
    name="Hello World",
    storage=s3_storage,
    path="helloworld",
    infrastructure=ecs_task,
)

helloworld_deployment.apply()

is giving me

$ python deployments.py 
Traceback (most recent call last):
  File "~/.pyenv/versions/prefect/lib/python3.10/site-packages/s3fs/core.py", line 112, in _error_wrapper
    return await func(*args, **kwargs)
  File "~/.pyenv/versions/prefect/lib/python3.10/site-packages/aiobotocore/client.py", line 358, in _make_api_call
    raise error_class(parsed_response, operation_name)
botocore.exceptions.ClientError: An error occurred (SignatureDoesNotMatch) when calling the PutObject operation: The request signature we calculated does not match the signature you provided. Check your key and signing method.

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

Traceback (most recent call last):
  File "~/code/wfp/wfp-prefect/deployments.py", line 29, in <module>
    helloworld_deployment = WFPDeployment(
  File "~/.pyenv/versions/prefect/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 230, in coroutine_wrapper
    return run_async_in_new_loop(async_fn, *args, **kwargs)
  File "~/.pyenv/versions/prefect/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 181, in run_async_in_new_loop
    return anyio.run(partial(__fn, *args, **kwargs))
  File "~/.pyenv/versions/prefect/lib/python3.10/site-packages/anyio/_core/_eventloop.py", line 70, in run
    return asynclib.run(func, *args, **backend_options)
  File "~/.pyenv/versions/prefect/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 292, in run
    return native_run(wrapper(), debug=debug)
  File "~/.pyenv/versions/3.10.9/lib/python3.10/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "~/.pyenv/versions/3.10.9/lib/python3.10/asyncio/base_events.py", line 649, in run_until_complete
    return future.result()
  File "~/.pyenv/versions/prefect/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 287, in wrapper
    return await func(*args)
  File "~/.pyenv/versions/prefect/lib/python3.10/site-packages/prefect/deployments.py", line 755, in build_from_flow
    await deployment.upload_to_storage(ignore_file=ignore_file)
  File "~/.pyenv/versions/prefect/lib/python3.10/site-packages/prefect/deployments.py", line 600, in upload_to_storage
    file_count = await self.storage.put_directory(
  File "~/.pyenv/versions/prefect/lib/python3.10/site-packages/prefect/filesystems.py", line 492, in put_directory
    return await self.filesystem.put_directory(
  File "~/.pyenv/versions/prefect/lib/python3.10/site-packages/prefect/filesystems.py", line 368, in put_directory
    self.filesystem.put_file(f, fpath, overwrite=True)
  File "~/.pyenv/versions/prefect/lib/python3.10/site-packages/fsspec/asyn.py", line 114, in wrapper
    return sync(self.loop, func, *args, **kwargs)
  File "~/.pyenv/versions/prefect/lib/python3.10/site-packages/fsspec/asyn.py", line 99, in sync
    raise return_result
  File "~/.pyenv/versions/prefect/lib/python3.10/site-packages/fsspec/asyn.py", line 54, in _runner
    result[0] = await coro
  File "~/.pyenv/versions/prefect/lib/python3.10/site-packages/s3fs/core.py", line 1101, in _put_file
    await self._call_s3(
  File "~/.pyenv/versions/prefect/lib/python3.10/site-packages/s3fs/core.py", line 339, in _call_s3
    return await _error_wrapper(
  File "~/.pyenv/versions/prefect/lib/python3.10/site-packages/s3fs/core.py", line 139, in _error_wrapper
    raise err
PermissionError: The request signature we calculated does not match the signature you provided. Check your key and signing method.

Is there a way to check what the python API client expects for deployment name and flow name for deployments that have already been deployed to the cloud? I have found this behavior confusing. See examples in comment thread

Hi! I've followed this tutorial to set up a scheduled deployment in Prefect Cloud. I edited the yaml with

schedule:
  cron: '*/5 * * * *'
  timezone: null
  day_or: true
is_schedule_active: true

To have it run every 5 minutes, and then applied it. When I run

prefect agent start -q 'test'

I see output from my deployment every 5 minutes, so its working, but as soon as I

ctrc + C

to exit the output of the agent, it stops, and nothing is scheduled. How do I get the agent to run without my terminal? 🙂

I've followed this tutorial trying to set up Prefect with Google Cloud Run. However after following all the steps, and running the deployment, the Flow run is still Late after 5 minutes. How can I debug what have gone wrong?

Hi everyone. I have just started looking into prefect and coiled and planning to use these 2 with ECS Cluster. Does anyone have any recommendations about good tutorials, examples, scripts on those? I have watched most from YouTube and looking into documentations on Coiled and Prefect pages now, but didn't find clear workaround on these.

Hey all, I need help so if I use this combination [ cache_key_fn=task_input_hash, cache_expiration=timedelta(seconds=20), retries=1, retry_delay_seconds=20 ] for a task then what will priortize i.e. if task crashes or fails and retries with same input will cache ignore that and run the task again or it will respond the same response from failed state without running the task again ? Also as I have kept cache_expiration _time and retry delay time equal so would it allow to run task again without any cached response ?

Think I have a bug but want to check that I'm using these methods correctly. The use of

prefect.get_run_logger()

is incompatible with testing task functions. A minimal example based on the documentation for testing task functions. I only added the two lines defining and using the logger.

import pytest
from prefect import flow, get_run_logger, task
from prefect.testing.utilities import prefect_test_harness

@pytest.fixture(autouse=True, scope="session")
def prefect_test_fixture():
    with prefect_test_harness():
        yield

@task
def my_favorite_task():
    logger = get_run_logger()
    <http://logger.info|logger.info>("running task")
    return 42

@flow
def my_favorite_flow():
    val = my_favorite_task()
    return val

def test_my_favorite_task():
    assert my_favorite_task.fn() == 42

Raises

MissingContextError("There is no active flow or task run context.")

on the call to

prefect.get_run_logger()

Hello Everyone!! I wanted to know what is a recommended configuration to run Prefect Server? Is it single threaded ( since it runs on python? ) will it make use of multiple cores? ( ex: 2, 4, 8, 16 ). What is the recommended RAM for server? ( 4GB, 8GB, 16GB )? etc. Are there any benchmarks from prefect that can be viewed? I tried to find the recommended system requirements in the docs, issues, Q&A, discourse etc but couldn’t find anything. Hoping to find some help here. Thanks.

Sorry if this is incredibly basic of a question or if Prefect is not a good solution for this. Is it possible for me to setup a prefect server on an old gaming desktop with Cuda software? I want to run jobs from my laptop onto the desktop. Use case is training CNN models overnight. If this is possible, I'd love any helpful advice to get me going in the right direction.

Hi everyone, I try to publish notification to mattermost channel, The channel is within a specific team group. How can i publish a notification to different channel on different team group. example to publish webhook from jira: "mattermost.com/webhook?secret="TOKEN"&team="TEAMGROUP"&channel="prefectchannel" " Posted in #prefect-community

📺 Join @Bianca Hoch in 5 minutes for a Getting Started with Prefect Demo livestream

Hi 😄 I was wondering why we can only write or read bytes on prefect filesystems. Why aren't there any method to write more conventional files like images or videos ?

Is there any way to make task retry logic specific to particular exceptions? It seems like an anti-pattern to handle any/every exception the same way. I don't see anything in the documentation for the

@task

decorator that would allow this.

Hi everyone, I want to run prefect locally using the official docker image (https://hub.docker.com/r/prefecthq/prefect/tags?page=2). I made a simple docker-compose.yml to get started and I've created 2 services (agent and ui), but I can't access the prefect ui at

localhost:4200

Here is my code repository https://github.com/dendihandian/transfermarkt-prefect, it requires docker-compose and anyone can try it by executing

docker-compose up -d

Hello everyone, Any advice on how to have a prefect flow triggered by rabbitmq events ? My current implem is to have a flow that is always running and listens on messages, then triggers the corresponding deployment when an event is received, but it seems very flaky and would prefer if prefect server is able to do it instead of running it in an agent 🤔

Hi I have this flow and i want to run some flows based on condition, but even if the condition is false, rest of the task also execute

ef start_inferencing(my_param, file_path):
    job_id = my_param
    algo_id = "402"
    res_1 = generate_uuid.submit(algo_id, job_id)
    call_on_failure_if_failed(res_1)
    res_2 = get_file.submit(file_path)
    call_on_failure_if_failed(res_2)
    res_4 = choose_valid_file.submit(prev_task=res_2)
    call_on_failure_if_failed(res_4)
    res_5 = prepare_request_upload.submit(prev_task=res_4)
    call_on_failure_if_failed(res_5)
    res_6 = send_data_request.submit(prev_task=res_5)
    call_on_failure_if_failed(res_6)
    res_7 = prepare_predict_request.submit(prev_task=res_6)
    call_on_failure_if_failed(res_7)
    res_8 = send_predict_request.submit(prev_task=res_7)
    call_on_failure_if_failed(res_8)
    res_9 = extract_lunit_jobid.submit(prev_task=res_8)
    call_on_failure_if_failed(res_9)
    res_10 = prepare_fetch.submit(prev_task=res_9)
    call_on_failure_if_failed(res_10)
    res_11 = fetch_request.submit(prev_task=res_10)
    call_on_failure_if_failed(res_11)
    if res_11:
        res_12 = convert_output.submit(prev_task=res_11)
        call_on_failure_if_failed(res_12)
        res_13 = zip_file.submit(prev_task=res_12)
        call_on_failure_if_failed(res_13)
        res_14 = send_to_HGW.submit(prev_task=res_13)
        call_on_failure_if_failed(res_14)
    else:
        pass
        

if __name__ == "main":
    start_inferencing(parameters=dict_)

i want to reun convert_output, zip_file, send to hgw, only when condition is true, else exit.

is there any prefect native function in python which is equivalent to the command:

prefect deployment ls --flow-name <flow-name>

?

hi there beautiful community! I'm getting started with Prefect and trying out CI/CD im watching this from the Prefect team about setting up CI/CD with Prefect here's roughly the code for the github workflow, that is supposed to apply a deployment on main branch push:

name: Ad Hoc Deployment

env:
  PREFECT_API_KEY: ${{ secrets.PREFECT_API_KEY }}
  PREFECT_API_URL: ${{ secrets.PREFECT_API_URL }}

on:
  push:
    branches:
      - main

jobs:
  ad-hoc-deployment:
    name: Build and apply deployment

    runs-on: ubuntu-latest
    timeout-minutes: 45

    steps:
      - uses: actions/checkout@v3
        with:
          persist-credentials: false
          fetch-depth: 0

      - name: Set up Python 3.11.1
        uses: actions/setup-python@v4
        with:
          python-version: "3.11.1"
          cache: "pip"
          cache-dependency-path: "requirements*.txt"

      - name: Install packages
        run: |
          python -m pip install --upgrade pip
          pip install --upgrade --upgrade-strategy eager -r requirements.txt

      - name: Build and apply deployment
        run: |
          prefect deployment build ./flows/pipeline.py:hello \
          --name "Ad Hoc Deployment" \
          --params '{"user_input": "github action"}' \
          --version $GITHUB_SHA \
          --tag ad-hoc-gh \
          --work-queue github \
          --apply

In the video they set up a

--storage-block

and use github as a storage block, and use that on the prefect deployment build command, but I'm wondering why that'd be needed. In the step above, we're checking out the repo so for me that would mean that we have all of the code already in this job, so no github storage is needed? I've had one build that succeeded with this setup, but I didn't see any deployments in the UI, double checked my env vars, and since then getting this error message that isn't very helpful, here's the repo where you can see it:

prefect.exceptions.PrefectHTTPStatusError: Client error '404 Not Found' for url '***/flows/'
Response: {'detail': 'Not Found'}

I'm just trying to get a minimal setup going, and understand what's needed and not needed for that. TIA for any pointers!

Hi there! I'm running into an issue with api authentication. I've sucessfully logged in through my terminal, pasting my API key when prompted but when I try to run a flow locally I get the following error:

Cannot create flow run. Failed to reach API at <https://api.prefect.cloud/api/accounts/a9c0f124-ca06-4646-a501-57a405ebf3c7/workspaces/43f1b8a7-ed9c-46d2-a88a-55eef95b8ef7/>.

-- does anyone have any recommendations?

task.map()

is useful for setting up a set of tasks to run in parallel. Is it possible to set up a lot of tasks to run in parallel, with some substructure? E.G. a task structure that looks like this image

prefect.get_run_logger()

raises

MissingContextError

when called in a method called by

concurrent.futures.ThreadPoolExecutor

. Is this a bug or is it a prefect anti-pattern to use ThreadPoolExecutor within a prefect task?

from concurrent.futures import ThreadPoolExecutor
From prefect import task, flow, get_run_logger

def concurrent_subtask() -> None:
    # This raises MissingContextError
    get_run_logger()

@task
def basic_task():
    get_run_logger().info("This works.")
    with ThreadPoolExecutor(max_workers=2) as executor:
        futures = []
        for _ in range(2):
            future = executor.submit(concurrent_subtask)
            futures.append(future)

        for future in futures:
            future.result()

@flow(name="Hello World")
def helloworld() -> None:
    basic_task()

I have 3 flows for a data workflow (flow1=extracts the data, flow2= makes data ready, flow 3= creates training set) For the next step of work, we want to run this workflow (3 flows) in parallel for different parameters +1000 times . Our concerns are: 1- memory 2- how to set up the pipeline to run +1000 flows in parallel + run the subflows sequentially Any thoughts?

Hello When I stop agent service, the flow runs schedule still written to database. When I started back the agent, the missed flow runs schedule running all at once. How to prevent this from happening and make it only 1 flow run at time for the same flow?

Hi i have configued aws S3 as my storage block, But i am getting this below error while pulling the flow

An error occurred (AccessDenied) when calling the ListObjectsV2 operation: Access Denied

Can anyone please help me on this, is there any permission i am missing here

What if I want to import

pandas

or

numpy

or any other 3rd party dependency from my workflow, what is the best practice to do that? The agent running my deployment/workflow might not have these packages in place, right? Should I build a custom agent Dockerfile with all of the dependencies, or is there a better approach to it? what are the tradeoffs between the different solutions? (edited)

How should web applications (Node.js) listen/subscribe to Prefect workflows completion? I have a task that trains a deep learning model through a Web Application. The “*Start training*” button on the web app sends a

POST /model

REST API call to the Node.js server which then in turn calls Prefect 2.0 using the REST API to start a new run (of an existing deployment). The workflow will train the model, run inference on some data and then persist the output to a database. This is obviously an asynchronous operation that may take a few minutes (or more) to complete. Assuming the workflow succeeded, I would like to notify the users (those who have Chrome opened to my web app) that something has happened, i.e. training completed. How should the Node.js be notified when the flow has finished (either successfully / failed)? Is there like a middleware / RabbitMQ / other message queue that the Node.js app can subscribe to, onto which Prefect publishes event? If not, does Prefect expose other broadcast events? And if not, should I poll periodically from my app and maintain state diff? Thanks!