Hey y’all. I’m new to Prefect and I think it’s rad. The local version of Prefect is easy enough, but I’m a little confused by secrets in Prefect Cloud. I’m trying to write to a Postgres RDS instance on AWS. The script I’m using writes to an existing table with

<http://pandas.to|pandas.to>_sql

. Apologies if this is answered in docs somewhere, but how do I pass my secrets to Prefect Cloud? I have 2 cases. In the first case I can run the script locally and pass the secrets as env vars, but in the second case, I’m hoping to run the job on Prefect Cloud once a week. Any help would be great! Thanks.

Is there a way I can use

version_group_id

instead of

flow_id

in my GQL query? I tried to simply swap out

flow_id

for

version_group_id

, but that’s giving me the following error

Hi, For a `map`ed task, on the LocalExecutor -- is the map guaranteed to execute serially / in-order across the mapped object?

Is

flatten

expected to work on a

dict

?

from prefect import Flow, task, context, flatten

@task
def A():
    return {i: list(range(i)) for i in range(3)}

@task
def B(y):
    logger = context.get("logger")
    <http://logger.info|logger.info>(y)


with Flow('flat map') as f:
    a = A()  # {0: [], 1: [0], 2: [0,1]}
    b = B.map(flatten(a))

if __name__ == "__main__":
    f.run()

I'm getting an unexpected error:

ERROR - prefect.flat map | Unexpected error occured in FlowRunner: KeyError(3)

Hello everyone. I would like to know if is possible to dynamically change run name based on schedule parameter?

Alright, not sure what kind of monster I’ve created. I’m created a flow that two distinct paths that are unrelated and don’t share any dependencies other than a few secrets. For some reason one of the mapped tasks won’t start until an unrelated task finishes. I’m using

LocalDaskExecutor

with 32 threads. I’ve verified in logs and the schematic that the

Discover

task should be ready to go as all it’s upstream tasks are complete; however, it still waits around for the unrelated

Sync

task to complete before starting. Was hoping someone might have an idea how to figure out what’s going on. I’ve created another example flow that had a bunch of unrelated root/child tasks and verified it seems to behave as I would expected, but haven’t tried to reproduce this with a test flow that uses mapping yet. I’ll attach a screenshot that visually shows (hopefully) what I’m talking about.

Hello there! Is there a way of retrieving the current flow labels during execution? My guess would be that this is available on the

context

, but I can't seem to find it. I'm trying to use the

prefect.tasks.prefect.create_flow_run

using the very same labels as the parent flow, maybe there's a shortcut for it.

Do state handlers capture state changes induced by setting state via the UI?

Question about logging best practices. In the docs (see below), a logger is defined within a task with

prefect.context.get("logger")

. Does a logger have to be defined within each task? This gets a bit repetitive if I have 5-10 tasks within a flow. Is there a more concise way to define loggers across tasks?

import prefect

@task
def my_task():
    logger = prefect.context.get("logger")

    <http://logger.info|logger.info>("An info message.")
    logger.warning("A warning message.")

hi folks! I think I'm having a mental disconnect, coming from temporal.io I've created an ECS cluster with a long running agent that's happily registered. I'm not trying to register a task, and a flow. However, when I try register it using

flow.register("pulumi_test", set_schedule_active*=False*)

it sets the flow to only run locally. More in 🧵

Hi there! I have a question about schedules. How can I fetch schedules for a flow group and specific parameters. Example: I have very many schedules. If I fetch with this query it will be a lot of data for the client. I need to fetch with a filter based on “targetProjectId” from “parameter_defaults”, see the image. It is possible? Thanks :)

query GetSchedules {
  flow_group(where: {id: {_eq: "eb595947-xxxxxx"}}) {
    schedule 
  }
}

Is there anyway to remove a prefect agent and stop prefect cloud from using it to run more flows?

Is there an equivalent to

from prefect.engine.state import Skipped

in orion? Maybe

Cancelled

? https://orion-docs.prefect.io/concepts/states/#state-details How do I import it?

from prefect.states import Cancelled

doesn’t seem to work

Hi! can you share graphql example to get the task result of a flow run

Hello Prefect team, a question about Orion. I'm starting to build out an async ML pipeline. I see the "Getting Started" page's async example doesn't pass results between tasks. I'm looking into the documentation regarding PrefectFuture and the API seems largely similar to `asyncio`'s

Future

. However, before I continue down this async rabbit hole, I'm wondering if there are any gotchas (or docs or source code) I should look into? In particular, regarding how to use PrefectFutures in mixed async / sync flows and tasks. Moreover, will I be able to use `asnycio`'s synchronisation primitives (https://docs.python.org/3/library/asyncio-sync.html) with PrefectFutures? Thanks in advance! 🙌

Hi Prefect team, I have currently implemented a task to load from disk a file and perform some action serially as mapping did not work due to memory constraints. Is there a way of mapping this task and putting some memory constraints on the task or mapping ? like for example telling prefect to map only if there is sufficient memory ?

Hi. I have a general question about architecture. I know there's repo with TF scripts for Agent on AWS, but are there any scripts for Prefect Server that could be setup on AWS so that we could use both Agent and Server on AWS without contacting Prefect Cloud?

Orion question: is there an equivalent of `Parameter`’s from Prefect 1.0?

There is a poll on Twitter with a question: "what orchestrator would you use next if you could switch today?" I would definitely choose Prefect! P Thanks to everyone who participates in the poll! 🙏

👋 Hi all! Is there a good place to ask about logging configuration options using Prefect Cloud? 😄

Hi all, how do I upgrade to the latest version of dbt within my Prefect flow? I am running it on a Docker container

I’m trying to follow the Scheduling a Flow-of-Flows example in the docs, but I keep running into the following error:

Error during execution of task: ValueError('Received both `flow_id` and `flow_name`. Only one flow identifier can be passed.')

More details in thread.

Hi there, we are running the prefect agents in kubernetes as per https://docs.prefect.io/orchestration/agents/kubernetes.html#running-in-cluster. We notice if we don’t specify the agent name with

--name

, it will assume the default name

Kubernetes

. This is okay when you only run one replica, but when you increase the replica number to 2, both will take the default name and in the Prefect UI, the two agents are treated as the same agent. So what is the recommended way to deploy multiple agents for HA purposes in Kubernetes?

Hilo, hopefuly simply question. When I write my code password=Secret(“PREFECT_PW”).get(), how do I test this when i run it locally?

Hello. We upgrade from 0.15.14 to 1.1.0 and it seems like our flows are still running but nothign shows up on the dashboard unless we pick a project

Hi, regarding prefect logger, is there any way to store them to external storage like my own database, in additions to Prefect Cloud? Eg [2022-03-25 151344-0400] INFO - prefect.TaskRunner | Task 'get_config['textlist_id']': Finished task run for task with final state: 'Success' [2022-03-25 151344-0400] INFO - prefect.TaskRunner | Task 'query_from_opensearch': Starting task run... [2022-03-25 151344-0400] INFO - prefect.TaskRunner | Task 'query_from_opensearch': Finished task run for task with final state: 'Success' [2022-03-25 151344-0400] INFO - prefect.TaskRunner | Task 'from_s3': Starting task run... [2022-03-25 151345-0400] INFO - prefect | Read from S3: 3 [2022-03-25 151345-0400] INFO - prefect.TaskRunner | Task 'from_s3': Finished task run for task with final state: 'Success' [2022-03-25 151345-0400] INFO - prefect.TaskRunner | Task 'validate': Starting task run...

Hey all! Simple question: if I raise an error within a task (i.e.,

ValueError

) is it possible to recover that traceback/error from a state handler? Right now I can say “this task failed and will be retrying in X minutes”, but not “this task failed for reason Y and will retry in X minutes”. Or can we pass keyword arguments to a state handler with this signature:

# What I have now.
def notify_on_retry(task: Task, old_state: State, new_state: Retrying) -> State: ...

# Maybe what I want?
def notify_on_retry(task: Task, old_state: State, new_state: Retrying, message: str) -> State: ...

# Alternative?
def notify_on_retry(task: Task, old_state: State, new_state: Retrying, **kwargs) -> State: ...

is it possible to create a helper function/task that implements both create_flow_run and wait_for_flow_run? I tried:

@task()
def flow_run_helper(flow_name, project_name):
    cfr = create_flow_run(flow_name=flow_name, project_name=project_name)
    wfr = wait_for_flow_run(cfr, stream_logs=True, raise_final_state=True)
    return wfr

but its yelling at me about "ValueError: Could not infer an active Flow context while creating edge". my app does a lot of chaining of flows, so i wanted to streamline this

still wrapping my head around some concepts here, so bear with me. I've got a a working prototype up and running which uses a flask webapp to dispatch flows to prefect cloud. Currently, I'm seeding the data (ie the tasks and flows to run) with a "bootstrap" script that needs to run before the webapp is launched, but I'd like it really to be a "batteries included" experience. How are people generally solving this? I could have a "setup" page in my flask app, but I don't want to get super involved for now

Hi all, when I create a connection to a db using prefect.task.sql_server and I create a global temp table, will the temp table persist even when the flow is done? Or will I have to explicitly close the connection as a final task in the flow?