I’m struggling to wrap my head around how state signals interact with the

retry_delay

attribute of

Task

objects. Based on the description of state signals here, I’d expect to be able to raise

signals.RETRY()

and have the task wait

retry_delay

before the task is run again. However, the retries appear to be being carried out immediately. Can someone explain where I’m going wrong in the

wait_until_success

task in the trailing code snippet?

Hi, I want to change the schedule of a Flow in my code. As an example I set

flow.schedule = Schedule(clocks=[CronClock("* * * * *")])

which works fine and later want to add another schedule or delete the current one. My idea is to use GraphQL with the

set_flow_group_schedule

and

delete_flow_group_schedule

methods.

@Chris White and I are going on PrefectLive starting in 10 minutes! We'll be playing with Prefect 2.0 and giving away Prefect branded socks in honor of our SOC 2 announcement - hope to see you all there.

What's the best way to retry a task, only in certain scenarios? For example, I'm pulling from an API. I only want to retry (after 30s) if I get a 429 code (rate limited)

Flow run from TwitchStream completed 🎉

Flow run from TwitchStream completed 🎉

crontab * * * * *

😆

For P 2.0 - do you have some general guidelines to use to decide if I should decorate a function(s) with

task

or

flow

(i.e. subflows)?

Hello, I have a flow on prefect cloud that is stuck in a pending state, even though I have an agent ready that was previously working. It says

x slots

used in the late runs section, even though the flow does not have any tags associated with the concurrency limit (I have some other flows that do). Any ideas on how I can resolve this issue?

5000 followers 🥳

I'd like to excute a graphQL with this kind of predicate:

{parameters: {_contains: {"foo": "bar"}},

but the syntax isn't quite right, I get the error

Syntax Error: Expected Name, found String "date".

can someone help?

Hi, what's the typical way you would run a ShellTask with an argument that's used in the

command

which is a return from a prev Task? eg

with Flow(name='t') as f:
   a = task_a()
   b = shell_task(a)

hi, seems I've found a bug: after upgrade to 2.0b2, I'm getting this when running even the simplest flow:

$ python flow1.py 
Traceback (most recent call last):
  File "/home/tarkovskyi/miniconda3/envs/prefect_exp/lib/python3.10/site-packages/prefect/client.py", line 257, in api_healthcheck
    await self._client.get("/health")
  File "/home/tarkovskyi/miniconda3/envs/prefect_exp/lib/python3.10/site-packages/prefect/utilities/httpx.py", line 102, in get
    return await self.request(
  File "/home/tarkovskyi/miniconda3/envs/prefect_exp/lib/python3.10/site-packages/prefect/utilities/httpx.py", line 47, in request
    request = self.build_request(
TypeError: BaseClient.build_request() got an unexpected keyword argument 'extensions'

for prefect 2.0 is there some sort of feature like a hot reload that will automatically create a new deployment when I change the file in the deployment location?

What's the meta way to run one task if another task fails in 2.0? pass the task as a parameter to the dependent task and check if the state is failed there? From what I can tell, the wait_for keyword will only check if it is Finished, not whether it is completed or failed

Hello- I'm wondering if there is any reason behind delaying the conda-forge releases of 1.x? https://github.com/conda-forge/prefect-feedstock

in 2.0, is there a way to see inputs and outputs of a task?

Hello, in Orion - how to handle logging from python files which are not part of flow code? I usually use loguru for my logging but not able to make it work with Prefect logging seamlessly. I would like to have both prefect logging and app logging to sink to same target (file/console) and with consistent format. Are there any good examples available on this ?

Hey team - I’m trying to have a play with some work queues on a remote host and I think something might be not right with the api endpoints

Hello, signed up for cloud beta but now the task always pending

when registering flows from python modules with the cli is it possible to override the

executor

and `run_config`s of the flows somehow?

I have a bunch of kubernetes flows for prod/staging and i'm working on a local docker-compose environment for dev purposes, currently the local flows fail when they try to spin up the

KubernetesRun()

environments

would a better way to do this be to write a simple python script using the client api and override the executor and run_configs there?

Hi! I try to implement custom retrying policy in Prefect Orion (2.0b2). But, as i see at this moment, policies in Orion are hardcoded (CoreFlowPolicy/CoreTaskPolicy, MinimalFlowPolicy/MinimalTaskPolicy) and server API handlers for setting states do not have parameters in requests to receive data about custom policies (task policy and flow policy not a parts of requests, but just a dependency injection parameters initialised with

provide_task_policy

/

provide_flow_policy

). So my question is about orchestration settings that will be released in the future. Is custom orchestration policies will be parameters of API requests to set state of task/flow? Or orchestration concepts and APIs will be seriously changed in the near future? How orchestration policies will be serialized/deserialized between server and agent’s engines?

Hi everyone 👋 ! I'm using prefect since a few months now but I keep encountering the same issue with my Airbyte tasks: From some reasons, my Airbyte connection failed at some point and throw a

Failed

status but, I don't know why, prefect considers the task as successful (cf. my screenshot in attachment). It's a blocking point from my point of view as it prevents me from relying on prefect and therefore forces me to double-check each time with Airbyte. Maybe this issue comes from my code but I don't see why ?

sync_airbyte_connection = AirbyteConnectionTask(
    max_retries=3, retry_delay=timedelta(seconds=10), timeout=timedelta(minutes=30),
)

with Flow("my flow", run_config=UniversalRun()) as flow:
    airbyte_sync = sync_airbyte_connection(
        <connection_infos>,
    )
    [...]

flow.set_reference_tasks([
   airbyte_sync
])

Hi, if anyone is experiencing issues with EKS and/or Kubernetes on dask using prefect, I found that the latest release (14h ago) of kubernetes-asyncio (22.6) has been breaking all my flows that worked yesterday (it fails to pickup serviceaccount for rbac, sends all requests as 'system/anonymous'). Pinning version 'kubernetes-asyncio<22.6' in my requirements fixed it for me. Issue: https://github.com/PrefectHQ/prefect/issues/5573

Hey. In test I would like to assert that create_flow_run was called with expected parameters. Is it possible somehow to get task result?

We've started seeing an error when accessing secrets from Prefect Cloud. Any ideas how I can troubleshoot this?

prefect.exceptions.ClientError: [{'path': ['secret_value'], 'message': 'An unknown error occurred.', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}]

We've laughed, we've cried, but most of all we've 🚀 launched! Come join @Jeremiah, @Chris White and myself as we recap Launch Week 2022 at 3p Eastern on . You might find yourself hearing a few more exciting announcements 😏 https://prefect-community.slack.com/archives/C036FRC4KMW/p1647621014192929

I'd like to set up a testing framework for testing Prefect flows. The idea is that when someone modifies a Prefect flow in a pull request, an automated DevOps process can launch that flow with certain parameters and somehow test that the flow did what it was supposed to do. I couldn't find anything in the documentation or in Slack. Is there an official testing framework for Prefect flows? If not, what do you all use?