I registered a Flow to the wrong Project. How do I delete it from that Project? The cli seems like it should but it does not seem to understand anything other than

delete project

.

$ prefect delete --help                                                                                                                                                     
Usage: prefect delete [OPTIONS] COMMAND [ARGS]...

  Delete commands that refer to mutations of Prefect API metadata.

  Usage:
      $ prefect delete [OBJECT]

  Arguments:
      project    Delete projects
...

Hey! How can I mount volumes on a specific flow without affecting the global configuration of docker agent as described here https://docs.prefect.io/orchestration/agents/docker.html#mounting-volumes

Hi, is there a way to set one

Parameter

as the default for another? Here's what I did

#task 
@task(target= {output_folder_path}/result)
def example_task(output_folder_path)
....

#in flow
step1_folder_path = Parameter("step1_folder_path")
output_folder_path = Parameter("output_folder_path", default = step1_folder_path)

When I do this, I got a warning

<ipython-input-226-3b472991f1ef>:8: UserWarning: A Task was passed as an argument to Parameter, you likely want to first initialize Parameter with any static (non-Task) arguments, then call the initialized task with any dynamic (Task) arguments instead. For example:

  my_task = Parameter(...)  # static (non-Task) args go here
  res = my_task(...)  # dynamic (Task) args go here

see <https://docs.prefect.io/core/concepts/flows.html#apis> for more info.
  output_folder_path = Parameter("output_folder_path", default = step1_folder_path)

When I execute the flow, it runs but the

target

doesn't work as intended, the task still runs every time even though the target files are there. When I provide a hardcoded string to the default of

output_folder_path

, everything works fine

Hi all! Just wondering if anyone has noticed memory leaking when using prefect with dask workers? I am already deleting the local flow variables, calling gc.collect(), and have completely disabled checkpointing with prefect. And yet, this is typical for my workers until the OOM and die over the course of a few days. All the little bumps and step ups lines (green, red, blues) at the top are the dask workers in the swarm that keep growing.

Hey, We use CircleCI to test and register our flows (Prefect v0.14.12) which we use Docker storage for. In our recent attempts to register a new flow we have been hit with this error and failing to build:

docker.errors.DockerException: Error while fetching server API version: ('Connection aborted.', FileNotFoundError(2, 'No such file or directory'))

(full trace in this thread) From reading around is looks like a potential docker issue, an inability to connect to a socket?! Has anyone else been hit with this?

Hi, while trying to execute a task that connects to a db(mongo) and extracts a collection I get this error :

Unexpected error: TypeError("cannot pickle '_thread.lock' object")

The task blueprint looks like this :

def mongo_connect():    
    db = get_db() // wrapper on top of pymongoclient to connect to the specified db
    collection = get_collection(db). // wrapper on top of pymongoclient to get a specified collection form the db
    return collection

I understand that prefect needs to pickle all the tasks , but this code works fine while being run as an independent script (without prefect decorators). How can I make the connection threadlock safe or rather pickle safe?

Can a flow be registered without storage configured? Let’s say I want to use docker storage, but I have a separate build and push workflow that I’m using and then just pass the image to my run configuration

Hello everyone 🙌 I'm having a problem with scheduling flows, all of my flows did not schedule automatically overnight and I can't activate them through the interactive API (When I try to

set_schedule_active

I always get

"success": false

). When I try to manually toggle the flow through the GUI the entire server hangs until I manually kill the GraphQL process on the server. Can anyone help or provide some insight? I'm acting as an interpreter for the tech team so I have to do some back and forth with my questions.

Is there a way to specify an alternative location for

~/.prefect/config.toml

? Maybe an environment variable, because I don't see any CLI flags about it?

Hello! I was wondering what are some of the patterns that people use to set up flows with configurable number of workers for Dask? Given that you need to pass the executor to a flow it seems like you can't use parameters (as far as I can tell)? Right now we're using KubeCluster to spin up ephermal Dask clusters in K8s but our n_workers is hardcoded, but we'd like to configure it per environment or possibly based on task inputs in the future. Thanks!

Does anyone have any experience with dynamically configuring a state handler for a flow or some subset of a flow's tasks? We have a flow that basically serves as a "flow template" (specifically, for running different

dbt

projects) that multiple teams at my org use. The individual teams will set up an orchestrator flow that has one task (the

StartFlowRun

task), and they'll pass in the various parameters that they need to pass in so that their specific

dbt

project will run. They also will directly attach state handlers that they configure to their

StartFlowRun

task within their orchestrator flow. Of course, in order for the status of the underlying template flow to bubble up to the orchestrator flow, we have to set

wait=True

on that flow. The issue with this setup is that it unnecessarily eats up our flow concurrency, since this setup always leads to at least 2 flows running at a given time. We don't really need the orchestrator flow to wait around; we're only keeping it waiting as a means to capture the template flow's status, which then gets passed along to the custom state handler that the respective team within our org has configured.

Hey Team! it seems like the interactive GraphQL API doesn't reflect my role permissions. I created a tenant using GraphQL, then tried to assign it my license with the assign_tenant_license mutation. It's forbidden; then i just created my tenant through the UI and it worked, automatically attached with my enterprise license. Now i have to delete the old zombie tenant but i don't have permissions to do that either 😞 . Any resolution to this?

For a Cloud Automation to a webhook, it mentions that you can use a Prefect Secret for the Webhook you want to send. But I cannot figure out how to get the secret that I made to be filled in here. Anyone have any ideas?

Should

prefect register -p my_flow.py

be updating the version of the Flow? I have changed the code and ran

register

again, multiple changes and multiple registers, but the VERSION is still

1

in

prefect get flows

Hi Guys, We're looking to use Webhook Storage to store the flows in our local environment. The Prefect server will be hosted in a K8 Cluster, any recommendations for a storage service that would fit that requirement?

Hello! when I see the calendar on the board it appears: Reminder! The Prefect Scheduler will only schedule 10 runs in advance. How can I program more scheduling??

Hey, I have been trying to delete a project in the UI for half of the day now. Keep getting this error

Hi everyone, I am new to prefect, so maybe my question seems naive! is it possible to use @task for methods inside a class? I have a class with several methods that can be set as task.

Hello folks, i am trying to find a way to give a flow runs a name (currently its generated dynamically in prefect, see screenshot) Is there a way i can configure a flow run name in a code? I don’t see any parameter in

flow.register()

I looked into the documentation but couldn’t find it.

Hey there! In the Prefect settings, you can use Parameters with the data type 'list', 'dict'. Tell me, what is the maximum number of elements a list and a dictionary in a parameter can contain?

Hello there! If there a way to perform a mapping with a specific batch size ? for example lets say I have two tasks

f

and `g`:

ys = f.map(xs)
zs = g.map(ys)

Is there a generic way to give

g

batches of y values, for example 10, rather than one at a time ? (And obviously, without having to wait for all the ys to be computed.) Or is this something that is specific to the executor used ?

Hello Prefect. I have a question about dealing with Flow Parameters. Whats the preferred way to concatenate flow parameters that will be passed into a number of task? Let me type up a simplified example below

Did the "Mapped Tasks" panel go away/move recently?

Hi, does anyone know if the env vars set on the prefect agent (deployed in k8s) get injected into a dask cluster automatically? Or do we also have to duplicate them when making the pod spec for dask? Our executor is defined like:

flow.executor = DaskExecutor(
    cluster_class=lambda: KubeCluster(
        name=f"dask-{flow_image_name}",
        pod_template=make_pod_spec(
            image=prefect.context.image,
            cpu_request=worker_cpu,
            memory_request=worker_mem,
            threads_per_worker=threads_per_worker
        )
    ),
    adapt_kwargs={
        "minimum": min_workers,
        "maximum": max_workers
    },
)

We have a CI/CD pipeline that automatically registers flows for us (e.g. when a PR is opened); in my registration script I do something like:

if not args.dry_run:
        logger.log(SUCCESS, f"Registering flow {flow.name}")
        flow.register(project_name=args.project, idempotency_key=flow.serialized_hash())

Everything is working as expected, except I'm getting more versions of the flow (after multiple registrations) than I was previously expecting. It makes me wonder what I don't understand about the

idempotency_key

- could someone explain to me what could cause the serialized_hash to change unexpectedly? for example, if a different VM registers flows, would the hash change?

Hey there. I’ve been playing around with Docker storage today, trying to get all source code packaged together with the flows each time they are registered, and am using the

files

and

env_vars

attributes as outlined in the Docs. But it seems that my

.dockerignore

file (in the directory from which I am registering the flow) is ignored by this build process. I have massive data directories nested at different levels inside the codebase so it’s essential there be a way to include this somehow. Anyone come across this / have any ideas? I’m curious about playing around with a custom Dockerfile, but the code that copies flows to a tmp Dockerfile in

create_docker_file

is dynamic so it seems tough that I could write a standalone Dockerfile to handle this case.

Hi there! I have some (not all) tasks that need need PandasSerializer, and also need to use different result types depending on if dev or production. I have a solution for this (below), but wondered if anyone had a cleaner solution?

condition = True
resultType = LocalResult if condition else S3Result

@task()
def create_list():
    return [1,2,3]

@task(result=resultType(serializer=PandasSerializer(file_type='parquet')))
def to_df(a_list):
    return pd.DataFrame({'col1': a_list})

with Flow(
        name="my-flow",
        result=resultType(serializer=PickleSerializer())
) as my_flow:
    my_list = create_list()
    my_df = to_df(my_list)

Hey prefect team! I'm trying to register multiple flows generated in a loop in my file. However, it seems to only register the latest flow in the loop. Whats the right way to do this?

Is there a way to pass indexed items as arguments to a mapped tasks? E.g. If I have some inputs that are a list of tuples that i'm mapping over can I specify specific elements to pass as args when mapping over them? E.g. Something like this:

@task
def gen_records():
    return [('item', 1), ('item', 2), ('item', 3)]


@task
def subscript_input(x, y):
    return x+str(y)


with Flow(
    name="Test Flow",
) as test_flow:
    records = gen_records()
    total = subscript_input.map(x=records[0], y=records[1])