Is there a way to restrict access for an API Key to only one workspace?

Starting a new thread for a new topic - Are we able to tag infrastructure blocks? I would like to tag my EcsTask but can not see the arg available for it in the code?

In Prefect 2.0, is there a pattern to replicate the subflow orchestration in the same way as 1.0? Instead of the subflows running in the same process asynchronously using asyncio.gather, I need the flow runs to be created with create_flow_run() (this part works) but then I need to wait for the subflow runs to complete in the parent flow. How can I make the parent flow wait for subflows running in different processes?

Hey all, I need some urgent assistance dealing with corrupted flow metadata. We deleted a flow yesterday after we had some performance issues. Every since then, our prefect agent has been stalling and repeatedly producing this error:

prefect.exceptions.ClientError: [{'path': ['flow_run', 0, 'id'], 'message': 'Cannot return null for non-nullable field flow_run.id.', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}]"

As far as I understand, this is a flow_run record without a flow. I need some assistance with removing this corrupted metadata as our data pipelines are all majorly delayed.

In the prefect 1.0 cloud UI, how can I see a breakdown of where runs are coming from?

Deployment Environment Variables There's something confusing about executing flows from the command line vs. deployed flow runs in a local execution environment. Deployed flow runs are writing to files in system temp directory, rather than to the local file system relative to the flow invocation. • Running prefect orion and flow code from a poetry managed environment, locally • CLI invocation of flow code runs in local $PWD, ie the local python project folder • Deployment runs flow code in a system temp folder, like

tmp_path

from

pytest

The confusing part was the flow code was found via relative path, but when writing the output to text file on a relative path the output files landed at an ephemeral system temporary folder. Naturally this wouldn't be an issue writing to s3, but for development I like to work in a local environment and write files relative to my project root. I don't think this is caused by the poetry environment, because the agent is running in the same environment but the distinction in behavior is between deployed flows and flows called from the shell:

# flow.py
import
@task 
def write_output(dataframe, filepath):
    dataframe.to_csv(filepath)
    logger.debug(f"current working directory: {Path().cwd()}")

@flow
def write_dataframe():
    write_output(df, 'data/csv/datafile.csv')

# shell invocation
poetry run python src/flow.py
    # prefect logs:
    | DEBUG   | Task run...  - current working directory: /Users/merlin/prefect-repo

# deployed flow invocation
poetry shell
prefect deployment run write_dataframe/depl-write
    # prefect logs:
    | DEBUG   | Task run...  - current working directory: /private/var/folders/gh/...4r3zq.../T/tmpXookj0s08prefect

My goal here is for my task to write to the same local file path where my project folder is stored. Is there a standard way to specify the working directory for deployed flow code on a local environment? Maybe this is managed by the config defaults (I couldn't find a likely ENV variable). My next attempt will be to specify an absolute path for the write location, but this is not ideal.

hi, anyway to set a custom default time interval here?

Hi, I would like to save task-results in a custom way. Say a task produces an image and I would like to cache this result. Does anyone know how I could save it as

.tif

file? I tried to implement a custom

Serializer

which just passes the image through and then wanted to use a custom

WritableFileSystem

as storage option that takes care of writing/reading the image as

.tif

. So far I did not manage and would be grateful for some advice. I am trying this in Prefect 2.x. In Prefect 1.x I managed to do something like this by implementing a custom

LocalResult

.

Hey all, in Prefect 2, is there a way to figure out if I’m running locally or with a remote? From my understanding, in Prefect 1 there was

prefect.context.get("running_with_backend")

Btw, docs for

ContextModel.copy

are broken https://docs.prefect.io/api-ref/prefect/context/?h=context#prefect.context.ContextModel.copy

I think there is a bug in the cloud UI for the infrastructure block process. Once a environment variable is set, the field can not be removed completely. It can be changed, but setting no variable at all is impossible.

How can I set a state with no data? Prefect throws an error if there’s no data set, but I have nothing to return

Hey folks, bit of a niche one - what’s the intention / purpose of job_watch_timeout_seconds in the

KubernetesJob

infra? I’m finding that the agent creates the job + pod just fine (and it the flow + pod run through to completion) but after X seconds (per that timeout parameter) the agent logs

Job 'xxxxx': Job did not complete.

per this even though the job is mid-way through running? ie it doesn’t seem to have any negative effect on the flow, it’s just telling me the job didn’t complete even when the job is very much still running..? Feels like something’s not quite right, just wanting to understand what the intention is….

Hey, I would like to use a string-block value during the flow configuration. How would I do that?

👋 are there guidelines around long-running tasks (e.g. max timeout, should they be marked

async

, etc.)?

Hello, question. Is there a way to change the backend to "cloud" or "server" directly from the prefect python api?

Hi, (1.0) we're seeing some strange behavior with child flows. When a new version of a long running parent flow (e.g. a backfiller) is registered with prefect cloud, some child flows kicked off with

prefect.tasks.prefect.create_flow_run

are being canceled without any additional info in the logs. Is this expected? Has anyone else encountered similar behavior?

For the Fivetran connectors in 2.0, what does

schedule_type

signify? Do you only use this if you want the sync to be scheduled outside of a scheduled flow?

Another question- will my secrets created in 1.0 transfer to the 2.0 UI? Or do I need to recreate them?

Hi all, I’m experiencing an issue with passing a large parameter in a flow that leads to the UI becoming unresponsive and crashing. There is discussion here specific to dataframes, https://github.com/PrefectHQ/prefect/issues/6962, and a fix in PR that will use orjson, but the issue with the UI crashing is not specific to dataframes, and any large parameter (e.g. array with 1MM elements) would cause the same issue. How Prefect is designed, the parameters passed to a flow are retrieved as part of the flow runs api. If a parameter produces a large string when serialized, this will led to a massive response coming back and either causing a timeout, or crashing the server dependent on the size. A workaround here is to avoid passing large parameters, write to disk/s3 and then pass the URL as the parameter. This has the downside increasing runtime and creating an IO bottleneck. Curious on the communities / Prefect teams thoughts about this paradigm? Wanted to start a discussion here as it is technically not a bug and working as intended, but has side-effects in standard use-cases. Additional information added in reply.

Hey Prefect, I have a task in 2,0 that I want to test, in that task I use the

get_run_logger()

. In order to test that function I am using

.fn()

but I can not test it because

get_run_logger()

failed every time because

E           prefect.exceptions.MissingContextError: There is no active flow or task run context.

Besides possibly using an optional kwarg in every single task that I have, how can I get around this to test my tasks logic properly?

This message was deleted.

Hi, I have a question .. I got a some flow id on cli (i.e., prefect get flows), and then i try to get flow’s status like is_running. So, I use FlowRunView.from_flow_run_id(‘my flow id’) but I got a ValueError: No flow runs found while querying for flow runs where ~~. Could you any one help me?

hi! i was trying to use the default dask task runner on our self hosted agent and orion. im running into

AttributeError: 'coroutine' object has no attribute 'type'

and the flow run crashes. any idea why this happened. adding detailed logs in the comment

Hey Prefect, I am completely lost on this one, I am trying to deploy a flow from a docker container and getting this error 🧵

Hi all, I'm currently running Prefect 2.6.5 on Kubernetes, running all flows in Kubernetes Jobs with flows baked in docker images. I seem to run into task run crashes which cause my flow run to crash, while I have retrying on these specific tasks. The task that reports the crash is a mapped task, over 12 inputs. The task is executed in a loop, with sleeping in between loop iterations. It mostly manages to execute a certain amount of iterations before crashing (like 30ish). Does anyone have an idea on how I could solve this, or what exactly is going wrong? Stack trace in thread

Getting this error while execution of flow sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) unable to open database file Also when i changed backend database to postgres this error comes up: sqlalchemy.exc.ProgrammingError: (sqlalchemy.dialects.postgresql.asyncpg.ProgrammingError) <class 'asyncpg.exceptions.Unde finedFunctionError'>: function gen_random_uuid() does not exist HINT: No function matches the given name and argument types. You might need to add explicit type casts.

Can anyone supply me with the correct format for environment variables in the docker container block?

When we have to loop a task to run multiple times, how can we avoid memory issue? I found that each time a task run, the memory steadily going up but not going down even if the variable is overwritten every time. (i’m using Prefect 2)

Hello. Have few questions about the flow runs in Prefect 2: 1. How to restart flow run from the start(as I understand retry will restart only failed tasks)? 2. How to cancel the flow run(Currently see only 'Delete' option)? Thanks