Hello! Is there a way to install only the prefect python client? I am writing some tooling that calls the prefect api and I prefer using the client instead of manually making http calls to the api. Thanks!

Hi. In Prefect V1 we used the following variable -

prefect.context.scheduled_start_time

. Is it any alternative in Prefect 2? Thanks.

hi all! Any way to configure the slack notifier to give a notification only when the FLOW fails? The tasks in the flow have retry attempts, so would like only a notification when the whole flow fails (if all retry attempts have been exhausted for example or something else failed)

Hello. I'm getting

ImportError: dbt-core needs to be installed to use this task; run `pip install "prefect-dbt[cli]"

when I definitely have

prefect-dbt[cli]

installed both for the system and for the project. The error happens when running

prefect_dbt.cli.commands.trigger_dbt_cli_command("dbt debug")

-- does this have to do with missing

trigger_kwargs

-- not the package missing? SOLVED: INSTALL PREFECT-DBT[CLI] AS ROOT (the subprocesses don't have the same PATH variable so the

which

command doesn't find them)

Hi folks! Would like to pick your brains regarding the CICD side of Prefect2. We currently are running flows via ECS with S3 storage. We have multiple different flows that don't share the same dependencies. It seems inefficient and error prone to build one large image for all of these flows to run in. So my question is: • Is anyone using a mono repo for prefect2 with multiple flows each having their own image and requirements files getting deployed via github actions or any other automated deployment tool? • If so, any insight into how you are building images with different requirements files in an automated fashion would be really helpful. Thank you!

does the v1 Git storage support submodules? docs https://docs-v1.prefect.io/orchestration/flow_config/storage.html#git

Hey, I’m finding

logger

doesn’t seem to output logs either to terminal or to the cloud UI when using the

DaskTaskRunner()

- I’m calling

get_run_logger()

within the task - code attached in 🧵 . Any ideas / am I missing something? (works fine with sequential / concurrent task runners, just not the Dask one)

I'm looking to get mypy working with 2.0 and it looks like every task's return value is changed to

None

instead of what is hinted at in the method. Does anyone know of a work-around besides ignoring the errors?

Hi folks, for a

state_handler(task, old_state, new_state)

Is there a way to access the original input parameters for the

task

? For example, my original task is as below, and I would like to access

config

in the

state_handler

def state_handler(task, old_state, new_state):
   # Question: how can I access the `config` variable I originally passed into `task`?



@task(state_handlers=[state_handler])
def my_task(config: Any):
    if config.is_prod:
        # do something

Is there a way to provide a dropdown in the Prefect Cloud UI for parameter values? We would specify the list of parameters in the flow code and then users would have a dropdown for them in the UI.

Does anyone know if it is possible to figure out in which context a flow is executing, local or remote - and accordingly modifying a task runner? I have a Ray cluster in k8s that our remote deployments will use, but have a requirement to allow local exec too. Problem is, users won't have access to our Ray cluster, which wouldn't be a problem, but the flow def has the address hardcoded. Looking to do something like (local exec has a temp Ray cluster, k8s exec has remote cluster):

@flow(
    task_runner=RayTaskRunner(
        address="<ray://ray-cluster-kuberay-head-svc:10001>"
    ),
)
def main(...):
    ...

if __name__ == "__main__": # If this flow is called directly, don't use the address
    main(..., task_runner=RayTaskRunner)

Hello everyone, I’m trying to process tables in an execution order defined in the dependency list, made some progress, however I’m not sure how I can refer to an existing task without creating one each time, using Prefect V1, sample code in the 🧵

Not sure if this is the right place, but I am running a prefect flow in a VScode python file and it will not stop running. Is there any sort of functionality like a flow.cancel() to stop this from running forever?

Has anybody got a working example of using EXDATE with the rrule schedule. Basically I want a daily schedule but to exclude public holidays. Here's an example rule I have that's meant to run daily and exclude Christmas & Boxing day.

RRuleSchedule(
    rrule="DTSTART:20221007T120000Z\nRRULE:FREQ=WEEKLY;UNTIL=20240330T120000Z\nEXDATE:20221225T120000Z\nEXDATE:20221226T120000Z"
)

Hello prefect community 🙂 We're getting some strange error when running flows:

Failed to load and execute flow run: FlowStorageError('An error occurred while unpickling the flow:\n  JSONDecodeError("Expecting \',\' delimiter: line 1141 column 82 (char 62124)")')

(We're using GCS for storage)

Hi there. Is there any way to have Redis act as my cache-store, instead of a filesystem? Assuming I were to create a custom block for this, how do I have the cache use Redis as a backend?

Hi guys ! Quick question here, I see that the fresh new version Prefect 2.5 was available with new blocks like

GCloudRun

. Great 🎉 ! However, after connecting to my prefect cloud environment no new block available and I could not find the way to find the Cloud UI current version. So my question was how can I retrieve the version from the cloud , and is there a way to update / rollback it manually ?

Hey everyone! I just started using Prefect earlier this week and it has been amazing so far! I want to clarify a doubt I had regarding work queues. When information is sent to a work queue (eg: when I manually trigger a flow from the UI) where does it go? Is the content of the work queue stored on Prefect Cloud?

I've gone through the

dataflow-ops

repo and I am planning to create an ECS service for a Prefect Agent that submits ECS Tasks that run the Prefect Deployments. In the repo, as per my understanding, the agent runs as an ECS based on an ECS Task Definition which requires a Docker Image. Does this Docker image need the different Python packages required by the

flows

? Or can it be lightweight and only include packages necessary for

prefect

?

Hi! I have just started testing Prefect and are looking into using Prefect Cloud. We want to configure the Prefect database (which can contain sensitive logs) to be hosted in our own cloud and not in Prefect Cloud. Is this possible, and if so how can this be done?

Anyway to create a Microsoft Teams Webhook block from the UI? I see slack, but no teams. It's in the Docs, but not in the UI.

Hi everyone! We're in the process of migrating to 2.0 and experimenting with the ECSTask block. We define our task definitions within terraform and deploying our agent ECS task works fine and it can successfully retrieve a task definition for a flow - the type, cluster and task_definition_arn are defined on the ECSTask block for the flow we're testing. However, when the agent tries to register the task definition we receive AWS errors of the form:

Unknown parameter in input: "registeredBy", must be one of: family, taskRoleArn, executionRoleArn, networkMode, containerDefinitions, volumes, placementConstraints, requiresCompatibilities, cpu, memory, tags, pidMode, ipcMode, proxyConfiguration, inferenceAccelerators, ephemeralStorage, runtimePlatform

This appears to be an api incompatibility on the aws-sdk side where the json returned from the describe call contains fields that are incompatible with the register command. There is a workaround described in the issue, but I'm wondering if anyone has come across this before when using prefect-aws or whether we need to do some additional configuration on our end to account for this? Thanks in advance!

I've been looking at trying to integrate Prefect (2.x) with Object Oriented Programming and design patterns. As discussed in an earlier thread https://app.slack.com/client/TL09B008Y/CL09KU1K7/thread/CL09KU1K7-1658744762.332369 it's not straight forward to integrate Tasks and Flows with classes. I've got something that works but I was wondering if there's a more elegant way In this example I'm making a process that gets a number and adds another number to it for simplicity

class Maker():
    def __init__(self, val=5):
        self._val=val
        
    def make_number(self):
        return self._val
    
    @property
    def make_number_task(self):
        return prefect.Task(self.make_number)
    
class Adder():
    def __init__(self, val=3):
        self._val=val

    def add_number(self, x):
        return x+self._val
    
    @property
    def add_number_task(self):
        return prefect.Task(self.add_number)
    
class Calculator():
    def __init__(self):
        self.maker = Maker()
        self.adder = Adder()

    def _flow(self):
        a = self.maker.make_number_task()
        b = self.adder.add_number_task(a)
        return b
    
    @property
    def flow(self):
        return prefect.Flow(self._flow)

This seems to work.

c = Calculator()
c.flow()
<Prefect messages>
8
c.maker._val=10
c.flow()
<Prefect messages>
13

I'm wondering if there's a better way of doing it. Wrapping class methods in prefect.flow or prefect.task doesn't work because it treats "self" like a required argument but the above seems to work.

What happens when a

flow

(

foo

) is running and a new deployment is created that changes the flow code that

foo

was executing. QUESTION 1: Would

foo

continue to execute and ignore the new deployment? Also, what happens if

foo

intends to trigger a

sub flow

(

bar

), but a new deployment is created that changes the flow code that

bar

would execute? So the timeline is a.

flow

foo

is triggered b.

sub flow

bar

is updated (sub flow id changes from

123

to

456

) c.

foo

gets to the point in execution where

bar

is called. QUESTION 2: Would

foo

trigger the newly updated

bar

code

456

? Or would

foo

ignore the change to

sub flow

bar

and trigger

123

as

foo

was triggered before

bar

was updated?

Scenario: I have a Docker Image built in GHCR, next I want to schedule this docker image to run every 2 hours. Is Prefect correct option for doing the schedule. If yes how. Unable to find documentation on same.

I would like some advice about production deployment of agents. I’m using Prefect 2 with Prefect Cloud 2… So I’m working to get my agent(s) to work within my Django environment…once that is finished, I am planning on deploying a new production machine behind a Load Balancer with a minimal percentage of production load as it’s primary job is to run the agent(s) and their tasks. Is there anything wrong technically or theoretically with that thought? I have to have my full production code available to Prefect as the tasks are going to be triggered by actions people can take within our system. (We are using this as a replacement for Celery, hopefully).

Hi, I have a prefect-orion 2.50. prefect-agent 2.5.0 and MinIO running on openshift. I have written a simple .py to create a deployment. The files are uploaded to minio and the script returns me a deployment ID but it doesnt seem to be registered on the server: the UI deployment panel is still blank. Any idea how to debug ? I’ve been trying for hours but couldnt solve it. thanks

Hello! Can tags be added to flows? I see a 'tags' column in the database, but i don't see any way to set it in the

flow

decorator. thanks!

I’ve asked a similar question before, but I have a bit better summary of what I am trying to accomplish. Original thread is here: https://prefect-community.slack.com/archives/CL09KU1K7/p1659389359820829 New “angle” to the question is basically: “I have a flow state handler function and I need it to behave differently based on something that happens during the flow run” The specific example is basically • I have a flow state handler that updates an external api with the status/running of a flow. This functionality is optional and controlled via an

external_job_id

parameter. So if a flow is running and it is to be associated with some external job record, the id is passed. If the id is passed, updates are made as part of the flow state handler (marking the job done when the flow completes, recording error messages, etc) The above is working great. The flow transition handler is generic, and simply checks for the existence of the parameter in the context. The “new” use case is basically a flow that is triggered by some outside automation. Part of the flows responsibility is to create the external job record, get its ID, and update it when the flow completes. Meaning I dont have the parameter at the start. Is there anyway to save some kind of information onto the flow, within a task, so that I can check for its existence within the flow state transition function? I’ve tried doing

flow.add_task(Parameter('external_job_id', default=$res)

where

$res

is the result of a task that creates the job and returns the id (half way though the flow). I also tried setting it in the context directly within the task. No luck