What is the proper way to modify one Parameter() based on another Parameter()? This does not work:

from prefect import Flow, Parameter, case, task

@task
def mytask(p2):
    print(p2)

with Flow("case-demo") as flow:
    p1 = Parameter("p1", default=False)
    p2 = Parameter("p2", default="good")
    with case(p1, True):
        p2 = "bad"
    mytask(p2)

print(flow.tasks)

$ prefect build -p flow.py
UserWarning: Tasks were created but not added to the flow: {<Parameter: p1>, <Parameter: p2>}.
$ prefect run -p flow.py
...bad...

hi I'm trying to run prefect2.0 in docker containers that I build manually when I run flows in a container, they seem to work I mount orion.db from host in docker run

-v /home/marcin/.prefect/orion.db:/root/.prefect/orion.db

when I run orion dashboards from host:

prefect orion start

it works as intended - shows flows run from container. But when I try to host dashboards from container I get empty dashboards (also mounting oriond.db, so both containers see the same database

-v /home/marcin/.prefect/orion.db:/root/.prefect/orion.db

added -p 4200:4200 to map port fdrom host to container, and exported PREFECT_ORION_API_HOST=0.0.0.0 so I can access dashboards in container from my host. orion.db is properly mounted in both containers, because I see that it changes in them when i run a flow. Dashboards are accessible, but empty...

Hi everyone, is there a sort of caching process with prefect secret ? I just changed one and my workflows are still getting the older value of the secret.

Hey, I'm trying to configure an ECSRun using a task definition. However when I try running it, it throws

An error occurred (ClientException) when calling the RegisterTaskDefinition operation: Container.image should not be null or empty.

even though I know that the image URI that I'm using within

containerDefinitions

has content.

Hey, I have a flow that generates a big pandas DataFrame (about 19M rows and 13 cols) and then saves it to Google BigQuery/Google storage.. etc (There are multiple tasks - but for this question I’m gonna simplify the flow). Task_a returns a df - I log the shape of it (at the last line of task_a) and get the 19M and 13 cols df - then this output gets sent to the next task - Task_b - I log the type and shape in the first row of this task - and its a Nonetype. Anyone ever had such an issue? (I debugged it locally and I cant reproduce the issue - I only get it once processing a lot of data - maybe its the issue here?) Example -

@task()
def task_a():
   df = pd.Dataframe()
   ......
   <http://logger.info|logger.info>(df.shape) # 19M and 13 cols
   return df

@task()
def task_b(df):
   <http://logger.info|logger.info>(type(df)) # Nonetype

with Flow('****',
          storage=Docker(registry_url="us-central1-docker.pkg.dev****/",
                         dockerfile="./Dockerfile"), executor=LocalDaskExecutor(scheduler="processes")) as flow: #
      df = task_a()
      task_b(df)

Hi! Is there any news on orion beta 6 release?

Can I bucket tasks together for the purposes of parallel execution with a Dask Executor? For example, Tasks 1-9 are dependent (3 depends on 2, 2 depends on 1) while Task 10 can be run simultaneous with the entire group of tasks 1-9.

Hi! I have a question about pulling secrets when a flow is executing. I've followed this guide for setting up a Prefect/DBT flow and want to be able to grab fresh database credentials at runtime without having to reregister the flow. Following the guide, I've got basically this:

@task
def get_dbt_kwargs(**kwargs):
    return kwargs


with Flow(name="dbt_command_flow", run_config=docker_run, storage=STORAGE) as flow:
    dbt_repo = Parameter(dbt_repo_url", ...)

    dbt_env = get_dbt_kwargs(
        DBT_ENV_SECRET__SQL_SERVER=PrefectSecret("DBT_ENV_SECRET__SQL_SERVER"),
        DBT_ENV_SECRET__SQL_DATABASE=PrefectSecret("DBT_ENV_SECRET__SQL_DATABASE"),
        DBT_ENV_SECRET__SQL_USER=PrefectSecret("DBT_ENV_SECRET__SQL_USER"),
        DBT_ENV_SECRET__SQL_PASSWORD=PrefectSecret("DBT_ENV_SECRET__SQL_PASSWORD"),
    )
   ... # a couple of tasks omitted for brevity
   dbt_run = dbt(
        env=dbt_env,
        command=dbt_command,
        task_args={"name": "Run DBT Shell command"},
        upstream_tasks=[pull_task, dbt_env],
    )

My question is - is there a better/preferred way to do the above? Basically, I'm trying to ensure that if credentials/secrets are changed in Prefect Cloud, then the next run picks them up correctly.

I'm having a hard time recalling this, and I know I've fixed it before. I've got prefect installed in an environment in anaconda, and I'm running the env, but I can't call the Prefect CLI. I keep getting this "prefect: command not found" error, but I don't know where to go find the executable to go make it executable. Can someone point me to it?

Hey, maybe a stupid question, but how do you re-apply a deployment withour re-creating it? I'm refering to the update path for the latest prefect release.

◦ If you are using an agent running on Kubernetes, update the Prefect image version to

2.0b6

in your Kubernetes manifest and re-apply the deployment.

Thanks for the update by the way - up until now, working with prefect is just awesome - I enjoy every minute of it.

hi, i managed to run local agent in docker container + my app and services in docker container two..i was trying to put the creation of the tenant

prefect server create-tenant --name default --slug default

in the dockerfile of my app, it doesn't work. why do other users have to install prefect locally outside the containers and run the command there. no way to have the command somewhere in the docker images?

HI! I would like to ask if parameters passed to a task in Prefect 2.0 are also validated with pydantic or is this the case only in the flow level. (For example if I have provided pydantic annotations/field types or a pydantic model to a downstream task will prefect still check the data the upstream task passes to the downstream task in runtime?)

Hi Prefect community. I am playing around with the GraphQL API and I do not understand the output. I want to check if an agent has some task "pending" or "running". I managed to do it but I do not understand the answer I get back from the API

Hi all< We're attempting to use the Docker agent but have run into a problem: the test flow that we created is registered to a local agent. Could you please assist. Thanks!

I'm trying to run

prefect cloud login --key <API Key I Just Created>

but I keep getting an error that says:

Unable to authenticate. Please ensure your credentials are correct.

Am I missing something?

Is there a plan to add the ability to run flows on GCP Cloud Run with Prefect 2.0 anytime soon?

Hello, we’re running Prefect 1 and recently upgraded from LocalDaskExecutor to DaskExecutors run on KubeClusters with Azure Kubernetes Service running the agent. We have a number of flows which are then run by a master flow. The master flow runs and can complete but doesn’t seem to be parallelising as expected - for the underlying tasks, the mapped children are running 1 by 1 even though the adapt_kwargs are set to minimum of 1 and maximum of 10 in each DaskExecutor. Is this a problem with running the flows from a master flow which also has a DaskExecutor defined?

Hello again, Thanks for the help so far, i've been able to build the foundations for orchestrating some pipelines. However, i'm coming back with a question. I have a flow containing two tasks: the first one is creating a bigquery table and the second one is calling an API at seconds interval (while loop with sleep) and inserting the payload into the bigquery table. The problem i encounter is when i try to register the flow using

flow.register(project_name='my-project')

inside the python code. The while loop from the second task is preventing the register and it is unreachable. I want to register and run the flow using

python file.py

More code inside the thread. Is there a way to reach the register command of the flow inside the python code? Thank you!

Hi all! I am trying to retreive time taken for @task and Flow( ) How to retreive/monitor start and end times of flow in the code. Is there any built in functions?

In the REST API docs for version 2, is there a chapter about authentication I am missing? https://orion-docs.prefect.io/api-ref/rest-api/

Looking back at Prefect 2.0 and trying to get my old testing (from alpha days) spun back up. I am getting this error I think with regards to registering my flow?

sqlite3.OperationalError: table flow already exists

Hi everyone, having a bit of a configuration problem. I have a script that runs fine when I run it locally but when I register it with Cloud, it fails to execute on there with the following error:

Failed to load and execute Flow's environment: ValueError('No flows found in file.')

This is the flow definition:

if __name__ == '__main__':

with Flow('Scouter-Solr-Script') as flow:

snowflake_user = PrefectSecret('snowflake_usr')

snowflake_pwd = PrefectSecret('snowflake_pwd')

snowflake_to_solr(snowflake_user, snowflake_pwd)

flow.run_config = LocalRun(labels=['SVRNAME1'])

flow.storage = Git(repo="Prefect-Flows", flow_path="Python/Scouter/snowflake_to_solr.py", git_clone_url_secret_name="azure_devops_clone_url")

flow.register(project_name='Scouter')

I'm assuming it has to do with the

if __name__ == '__main__'

part on top of the Flow class definition, but I don't know how to go about fixing this. I need that for multiprocessing that happens within the main task of the flow.

hi everyone, is it possible to add annotations to k8s jobs created by 2.0 agents? currently we inject Vault secrets via annotations and it works fine in 1.0 (we have a custom job template deployed with the agent) but I’m trying to scope the migration process and I haven’t seen any equivalent in 2.0 (I also looked at the agent code for 2.0b6)

Hi all, relatively new to prefect 2.0 and using the cloud to get going. Had two questions that somewhat baffle me so far and I couldn't find in the docs: 1. regarding the storage. Configured remote S3 storage... what I'm not sure about is if this storage is used somehow in prefect? I had to give aws keys and I wasn't sure if this is somehow kept on prefect side (cloud). 2. i made a scheduled deployment, every 300 secs. when i look in the UI it seems to have put there about 100 flow runs into the future, but I can't figure out why approx 100... why not 20.. or 50.... I don't remember seeing any configuration for this.

Hello! where can I find out more info about what changed in the block api for beta6? I have a script that configures minio as default storage and I cant get it to work with the new apis. Before beta6 there were 2 abstractions bloc_spec and block and now there are 3? block_type, block_document and block_schema? thanks! https://github.com/fraibacas/prefect-orion/blob/main/flows/init_orion.py#L15

Custom Roles Is it possible to limit access of users to specific projects in their "workspace" using "Custom Roles" in the enterprise edition ? If yes, is there an example? I do not see it in https://docs.prefect.io/orchestration/rbac/overview.html

hey all — is there a straightforward way to run a flow continuously, sequentially and in perpetuity? like an infinite loop.

Is it just me, or is the UI broken for flow runs at the moment? https://cloud.prefect.io/zesty-ai/flow-run/fc3a37ca-bcce-44b9-ae05-572ecf91328d

Where can I find the signals FAIL, SKIP, etc. in Prefect 2.0? I am getting an error when importing them from the old place.

I'm attempting to use map to reuse tasks and am running into the following error. Any insight into what I might be doing wrong?

Unexpected error: KeyError(0)
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/prefect/engine/flow_runner.py", line 569, in get_flow_run_state
    executors.prepare_upstream_states_for_mapping(
  File "/usr/local/lib/python3.8/site-packages/prefect/utilities/executors.py", line 681, in prepare_upstream_states_for_mapping
    value = upstream_state.result[i]
KeyError: 0