Is there a way to start a prefect agent via code instead of manually using the CLI?

Is it possible for an enterprise account to have multiple versions of Prefect cloud running along side each other? Like, ideally I’d be able to upgrade at the tenant level so I can upgrade in dev, then do production etc.

I’d also like to have multiple agent versions running along side each other (separated by tags and tenant)

I perused a lot of the examples and youtube vids, and the examples were for local deployments.

Hey folks! I’m currently trying to solve an issue with Prefect 2.0 when building a deployment via CI - currently, the deployment uses local storage and the path to the code within the CI environment doesn’t match the path in the runtime environment. I’ve been trying to figure out how to override the path in the

prefect deployment build

command, but this hasn’t worked:

prefect deployment build \
          ...
          --override path=/app/

Is there a way to override the path setting, or is this something I’d need to modify in the YAML files after generation?

I'm wondering if anyone had success creating data streaming workflows with Prefect 2 using AWS API Gateway + Lambda. I noticed @Zheng Cheng posted about it a couple months ago. I just wanted to check and see if anyone had success with this before I try to tackle it blindly.

I am running two different agents in the same machine. It seems by default besides the labels you might add to the agents, it adds the label from the flow's storage (Flow.storage.labels) I do not want to have the same label present in those two agents. I was able to overcome this by simply defining

agent = LocalAgent()

agent.labels = ['custom_label'}

instead of passing the labels to the LocalAgent params upon instantiation. This way i have complete separation from agents even if they are running locally on the same machine Now the issue with registering flows is the same, the labels assigned to the flow are the ones you provide + that default local label. I can't seem to remove the default label before registration and the agents don't work, because their label must be a `superset`of the flow's labels.

Hi everyone, i'm currently transforming some previous Prefect flow to the Prefect 2.0.....i encountered some issues during migration • How can I write log into the local file? In 1.0, i use python's file handler and add that handler to the Prefect logger. However, in 2.0, it seems there's no addHandler method in Prefect logger.

Is there a way to "package" up a set of tasks into a super task that can be re-used easily? For example, we have a few tasks that usually get used in sequence: validate some pandas dataframe, grab some secrets from AWS, use those secrets/credentials to insert the dataframe into a database

Hi. Looking for some tips on self-hosted authentication / authorization (user logins, roles) when running our own Prefect Orion (2.0) server. We can't use Prefect Cloud due to compliance issues with using the SaaS offering (despite its hybrid security model), and this seems to be about the only box not checked for what we want the tool to do in standalone form. I don't want to have to open it up and bolt on FastAPI OAuth2 support to the web api. Any other solutions or workarounds to make it so that anyone on the network can't arbitrarily start, stop, modify flows? Maybe there's a way to allow all GET requests and block any POST/PUT/DELETE etc. to effectively make it read-only by default, and writable for privileged users?

I build deployment object using: prefect.deployments.Deployment.build_from_flow(), but this object does not have apply() method? how should I deploy a flow from Python instead of CLI?

Hi All,

er, oops

Hi All, We've recently upgraded to prefect orion and moved from dask to ray. (Everything running on kubernetes in AWS EKS). Most things are working smoothly but we're hitting intermittent errors

ValueError: Path /root/.prefect/storage/<GUID> does not exist

sometimes. What I suspect is happening is a task is failing and a different ray worker node is attempting to retry it, seeing that the result was already serialized to disk and trying to read it from the cache. Except that the cache was on a separate node and doesn't exist locally. With Prefect 1.0 we were using S3Result() for all of our tasks which worked really nicely. It looks like there's currently no equivalent for orion. Are there any plans to add that functionality back in? Is there a good work around we can use in the meantime?

Anyone having issues connecting to Cloud v1 at the moment??

Can someone confirm if they are having issues logging in as well? Edit: For anyone reading this Chris White responded in #prefect-ui stating that they are aware and are in the process of recovering.

Good morning dear prefect community. What is the best and suggested way to update existing deployments, if the parameter schema changes? Let's say I want to add a parameter, my workflow is: 1. Add parameter to flow 2. Run prefect deployment build 3. Add new parameter value to deployment.yaml 4. Run prefect deployment apply However, the UI never updates and never shows the new parameter. As a workaround, as step 2, I delete the deployment before recreating it - is this the preferred way at the moment - or am I missing something?

How can be the start_time is null ? even though the flow_run and task_run was success, any possible reasons for this ? or it is some edge case problem while writing to database ?

I have a problem with: Task missing UUID in get_or_create_task_run_info - Expected type UUID! - INTERNAL_SERVER_ERROR My colleague also wrote on this some time ago: https://github.com/PrefectHQ/prefect/issues/5999 Now it happens again. What should I do?

Is there a way to pass the K8S namespace and PREFECT_API_URL values to prefect deployment build ./my-favorite-function.py:validation_flow -n saman-test -t kubernetes -i kubernetes-job so I don’t have to edit the YAML file afterwards?

Hi! Question about Prefect blocks, security and confidentiality. When we use, let's say S3 or Amazon blocks, it is necessary to provide access keys to the Prefect. In theory, with those keys Prefect team can secretly access our files and do with them whatever the heart desires. Is there anything other than gentlemen's agreement that can guarantee that Prefect team won't perform such shenanigans?😁

Hello, What postgress schema does prefect 2.0 use? Can we use different. Schema?

Hi With Prefect 1 - Any tips for detecting if a task was used within another task without using

task._*run*()_

? I want to add a pytest to check and prevent adding incorrect task uses like this.

import prefect

@prefect.task
def task1(par):
    task2()  # this is incorrect! it should be task2.run()

@prefect.task
def task2():
    return 1

with prefect.Flow("Sample flow") as flow:
    aa = prefect.Parameter("aa")
    task1(aa)

flow.validate()

doesn't return errors/warnings, since it doesn't check what's happening inside each task. Meanwhile, in the task, there's another task hiding and being used incorrectly. It will fail when running the flow.

ValueError: Could not infer an active Flow context while creating edge to <Task: task2>. This often means you called a task outside a `with Flow(...)` block. If you're trying to run this task outside of a Flow context, you need to call `task2.run(...)`

I could do a

flow.run()

in my test, but that can be cumbersome and require a lot of scaffolding in terms of setting up mocks, and even with mocks, the flow run might just be too slow. A

task.validate()

can be handly, but I'm finding some difficulty in writing such a function.

Does Prefect 2 have something similar to Artifacts from v1?

Hello. In prefect 1, is there a docs you can point me towards for the following problem: I have daily scheduled flows that run dbt projects for the previous day. Dbt dates are provided from the flow by Parameters, and the default is

date.today()

and

date.today() - timedelta(1 day)

. Now I want to make the flows resilient. I want the flow's date parameter to be the last success flow run, so that if one day the flow run fails, the next flow run automatically takes the last 2 days. Basically, I want the dates to be

date.today()

and

last_successful_flow_run_date

. I was thinking of storing the flow run result and its parameters in an S3 bucket, then query it every time to recover the last date that the flow ran successfully. But I'm not sure if this is the best way. I'd appreciate any help with this. Thank you

hey, so I have a weird problem.. 1. I have flows running with flow storage from gitlab, 2. dependencies from ecr image (run_config) 3. ecs agent (run_config) now I have another requirement of storing the results in s3😅 I register the flow from a different file(cli file) n add the flow storage and run config through (separate from the actual flow file). Everything works great till here, ecs pulls in the right image, grabs the flow from gitlab, runs fine. but I can't seem to get results working like this. 1. s3 storage works when I the storage to the flow in the flow file 2. s3 storage does not work when I add it through the cli file before registering. any idea what am doing wrong? feels like the storage I specify gets overwritten when the flow gets pulled in from gitlab (speculating though) Edit:I want results so that I can retry flows when they fail(if any1 was wondering). Also planning to write a scheduled flow that clears results of completed flows so that I don't have to worry about using too much storage (just 3mil records or so :3)

(Not an issue) Glad to see in Prefect 2.0, the flow is able to accept the dynamic parameter such as datetime.now(), so there's no need to wrap the dynamic input in a separate task now :D

Hey there, can you see logs at a task level in Prefect Cloud 2? I notice in the flow log output the task id is on logs coming from specific tasks but I can't seem to filter by task.

Not sure if correct place, but is there a technical reason why the imperative api doesn’t exist in prefect 2.0? I found it very helpful when building tasks with reusable logic and tasks that have other class methods benefiting from the encapsulation the imperative api provides. (More in thread)