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)

Is there a way to control the max concurrency of the

ConcurrentTaskRunner

?

I'm in need of an easy way to login to google account in prefect, but I didn't want to use Selenium. Can I use Playwright in prefect?

I’m trying to set up CI on a flow repo using the template here: https://github.com/anna-geller/dataflow-ops/blob/main/.github/workflows/main_simple.yaml but am running into a 404 issue with the API similar to the one reported here: https://prefect-community.slack.com/archives/CL09KU1K7/p1661349987899419 I’ve double-checked the API URL being used in CI against the one on my local CLI and have verified that the API Key being used is active as well. Is there a better way to debug why I can’t register flows to Prefect Cloud?

Hi Everyone, I'm having an issue with deployment storage. This deployment is using the

process

infrastructure and the

s3

block for filesystem storage. I build my deployment, the yaml is generated, and my flow code is uploaded to s3. I have prefect orion running locally, and this deployment applied to my local prefect. I then start a local agent within my python virtual environment. In orion, I click

Run

on my deployment. The flow run is moved to my local work queue, my local agent picks up the flow run, and builds the infrastructure. The agent then runs the prefect engine, which begins pulling my flow code from s3. In doing so, I receive this flow run failure log:

Flow could not be retrieved from deployment.
Traceback (most recent call last):
  File "C:\Users\BStefansen\.virtualenvs\lbx-broadband-pipeline-4k_4F-H8\lib\site-packages\prefect\engine.py", line 254, in retrieve_flow_then_begin_flow_run
    flow = await load_flow_from_flow_run(flow_run, client=client)
  File "C:\Users\BStefansen\.virtualenvs\lbx-broadband-pipeline-4k_4F-H8\lib\site-packages\prefect\client.py", line 104, in with_injected_client
    return await fn(*args, **kwargs)
  File "C:\Users\BStefansen\.virtualenvs\lbx-broadband-pipeline-4k_4F-H8\lib\site-packages\prefect\deployments.py", line 55, in load_flow_from_flow_run
    await storage_block.get_directory(from_path=None, local_path=".")
  File "C:\Users\BStefansen\.virtualenvs\lbx-broadband-pipeline-4k_4F-H8\lib\site-packages\prefect\filesystems.py", line 399, in get_directory
    return await self.filesystem.get_directory(
  File "C:\Users\BStefansen\.virtualenvs\lbx-broadband-pipeline-4k_4F-H8\lib\site-packages\prefect\filesystems.py", line 260, in get_directory
    return self.filesystem.get(from_path, local_path, recursive=True)
  File "C:\Users\BStefansen\.virtualenvs\lbx-broadband-pipeline-4k_4F-H8\lib\site-packages\fsspec\spec.py", line 801, in get
    self.get_file(rpath, lpath, **kwargs)
  File "C:\Users\BStefansen\.virtualenvs\lbx-broadband-pipeline-4k_4F-H8\lib\site-packages\fsspec\spec.py", line 769, in get_file
    outfile = open(lpath, "wb")
FileNotFoundError: [Errno 2] No such file or directory: 'C:/Users/BSTEFA~1/AppData/Local/Temp/tmp5t7vj96aprefect/aurora\\aurora_constants.py'

Expectation I would expect the prefect engine to use s3 file paths for pulling down the flow code, but it seems to be using local file paths. Is this an issue with my deployment/flow setup, or is this a bug? Environment • Windows 10 • Python 3.10.5 • Prefect 2.2.0

@Anna Geller The team at my company saw this issue you opened with

s3fs

for

DockerContainer

infrastructure blocks, but this is also an issue for me with the

process

infrastructure when using prefect cloud. If I run a deployment in the cloud using

s3

for filesystem storage and

process

for infrastructure, our flow runs error out saying

ModuleNotFoundError: No module named 's3fs'

I'm not sure if this is due to our agent being a

kubernetes

agent and our deployment infrastructure being

process

, or something else. If the

process

just runs a subprocess on a docker image made by the kubernetes agent, then I would assume it's an issue with the default image used by the kubernetes agent

Can I set add a key to

context

in a task such that the key would be available to other downstream tasks within that run?

Hi, I would like to have an input from you guys. I am looing at prefect2.0. I was expected that in a deployment I could specify an environment (docker image) per subflow as my subflows can have a lot of dependencies that are not always compatible between each other. How I envision things is to have 1 image per subflow. As the main flow expands, it seems that allowing such deployments would make everything more compartmented and maintainable. Have you encountered such request before? Is there a way to do that in prefect 2.0?

I am not a pip expert, how do you install optional packages locally when doing development?

pip install -e ".[dev]"

does not install the packages

Hi, after I executed a flow on agent, I did nothing, then the agent will pop up those exception......it doesn't seem to impact function, but it's pretty annoying. Do you have a clue how can i get rid of these? (exceptions will be pasted in thread)

If I'm getting this:

UserWarning: Block document has schema checksum sha256:28a9d16c2589998d5281b0496b25edd76c1a0b159095d0c7fe5eb5dc5ab62549 which does not match the schema checksum for class 'GCS'. This indicates the schema has changed and this block may not load.

Should I delete and recreate my GCS block?

Missing something small, whats the syntax to pass multiple work queues to an agent