Hi all! I am trying to generate a manifest YAML file using the

prefect deployment build

CLI. As for the block options, I have defined the following custom

LocalFileSystem

block (shown in the image below). However, when building the manifest file with the command

prefect deployment build src/flows/example.py:healthcheck --name test --storage-block local-file-system/mount

, I get the following error:

Found flow 'healthcheck'
Traceback (most recent call last):
  File "/home/user/Miniconda3/envs/prefect/lib/python3.9/site-packages/prefect/cli/_utilities.py", line 41, in wrapper
    return fn(*args, **kwargs)
  File "/home/user/Miniconda3/envs/prefect/lib/python3.9/site-packages/prefect/utilities/asyncutils.py", line 201, in coroutine_wrapper
    return run_async_in_new_loop(async_fn, *args, **kwargs)
  File "/home/user/Miniconda3/envs/prefect/lib/python3.9/site-packages/prefect/utilities/asyncutils.py", line 152, in run_async_in_new_loop
    return anyio.run(partial(__fn, *args, **kwargs))
  File "/home/user/Miniconda3/envs/prefect/lib/python3.9/site-packages/anyio/_core/_eventloop.py", line 70, in run
    return asynclib.run(func, *args, **backend_options)
  File "/home/user/Miniconda3/envs/prefect/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 292, in run
    return native_run(wrapper(), debug=debug)
  File "/home/user/Miniconda3/envs/prefect/lib/python3.9/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/home/user/Miniconda3/envs/prefect/lib/python3.9/asyncio/base_events.py", line 647, in run_until_complete
    return future.result()
  File "/home/user/Miniconda3/envs/prefect/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 287, in wrapper
    return await func(*args)
  File "/home/user/Miniconda3/envs/prefect/lib/python3.9/site-packages/prefect/cli/deployment.py", line 853, in build
    deployment = await Deployment.build_from_flow(
  File "/home/user/Miniconda3/envs/prefect/lib/python3.9/site-packages/prefect/deployments.py", line 720, in build_from_flow
    await deployment.upload_to_storage()
  File "/home/user/Miniconda3/envs/prefect/lib/python3.9/site-packages/prefect/deployments.py", line 572, in upload_to_storage
    file_count = await self.storage.put_directory(
  File "/home/user/Miniconda3/envs/prefect/lib/python3.9/site-packages/prefect/filesystems.py", line 187, in put_directory
    shutil.copytree(
  File "/home/user/Miniconda3/envs/prefect/lib/python3.9/shutil.py", line 568, in copytree
    return _copytree(entries=entries, src=src, dst=dst, symlinks=symlinks,
  File "/home/user/Miniconda3/envs/prefect/lib/python3.9/shutil.py", line 467, in _copytree
    os.makedirs(dst, exist_ok=dirs_exist_ok)
  File "/home/user/Miniconda3/envs/prefect/lib/python3.9/os.py", line 225, in makedirs
    mkdir(name, mode)
FileNotFoundError: [Errno 2] No such file or directory: ''
An exception occurred.

Any idea why this is happening? I'm currently using

python=3.9.13

and

prefect=2.6.4

You can find the

healthcheck

flow code in the following link: https://discourse.prefect.io/t/how-to-deploy-prefect-2-0-flows-to-run-as-a-local-process-docker-container-or-a-kubernetes-job/1246

In the Prefect 2.0 Cloud UI, where can I change my email address? When I goto the profile editing it is grayed out.

Follow up question. Now that self-service organizations are available in Prefect Cloud 2.0, is it possible to convert an existing workspace into an org workspace?

Hello! Is there any way in Prefect to use a mount point ("/home/user/mount/central-storage") as remote storage block? I have tried to set it to 'basepath' of LocalFileSystem, but this way I cannot run deployments on remote agents. Is there a "hacky" way to use RemoteFileSystem for this purpose?

Hey everyone - I’m trying to figure out what would be a good architecture on AWS for running a large number of flows in a burst (ex: X00,000 flows run once a week, broken into smaller batches). Ideally I’d want the backing infrastructure that these flows run on to be ephemeral, so it seems like I could use any of the following to do this: • Spinning up more agents temporarily (current plan) • Kubernetes jobs • ECS (?) • Dask + Fargate (?) ◦ I know Dask parallelism operates at task level rather than flow level I’m wondering if anyone here has similar use cases. If so, what works well for you?

Hi All - hopefully an easy question: What is the proper way to redefine an existing task? E.g., In interactive mode I create my_task, then edit my_task, and I get an error of "OSError: could not get source code". Using 2.6.4

Hi, I was creating a new deployment earlier and tried to use the '--skip-upload' option in building my deployment, but got an error that it was not an available option. It's listed as an option to use in the documentation but doesn't show up when checking through 'prefect deployment build --help' in terminal. Has this feature been removed?

What is the recommended way to test flows that reference external blocks? For instance this test fails because result_storage is set to a block name.

from prefect import flow, get_run_logger, task
from prefect.testing.utilities import prefect_test_harness


def test_flow():
    @task
    def _add(x: int, y: int) -> int:
        get_run_logger().info(f"Adding {x} + {y} in task")
        return x + y

    @flow(persist_result=True, result_storage="s3/mldc-result-storage", result_serializer="json")
    def add(x: int, y: int) -> int:
        get_run_logger().info(f"Adding {x} + {y}")
        result = _add(x, y)
        return result

    with prefect_test_harness():
        assert add(5, 6) == 11

Hey folks. Is there a best practice for conditionally triggering a deployment for a flow, from another flow in Prefect 2 i.e.

flow a deployment x

triggering

flow b deployment y

?

👋 hullo! we're thinking through our infrastructure story with prefect right now, and are aiming to deploy to EKS. a question i have is, is there a standard model of # of agents to # of nodes in a cluster? Or should you be having a single agent per flow that you want to execute in the cluster? Also apologies in advance if there's already a recipe or something i should be looking at 🙂

hi - I am hitting an error in prefect-aws, get_directory

@Anthony Head @Ryan Peden @Zanie @Andrew Brookins @Nate… Just wanted to say a big 🎉THANK YOU🎉 to each of you and others who have helped me. My client loves the idea of us replacing Celery with Prefect and was impressed with what we had working and how we can leverage Prefect in the future for more traditional and non-traditional workflow solutions!

[SOLVED] Hi, I am getting 404 when trying to enter my workspace in cloud (never happened to me before). I didn’t do anything different in changing settings. Can you help me? Update: According to this https://prefect-community.slack.com/archives/CL09KU1K7/p1666181421767899?thread_ts=1666168584.461719&cid=CL09KU1K7

Another one - my deployed flow cannot retrieve environmental variables from my local .env file. The file is in

.prefectignore

when I upload to my storage block and I am using a local agent. I am pretty sure that in the past this worked, but anyway - How can I set the environmental variables of my execution from an .

env

file?

hey guys, I am following this docker-compose setup for local development https://github.com/fraibacas/prefect-orion, but for some reason, even though

s3fs

is installed, I cant add any kind of block in the UI, do you think the issue is related to docker?

Hi everyone! Have been having a lot of success with migrating Prefect 1 to Prefect 2 on GKE. It has been a great experience!! I'm not sure this is the correct place for feedback but there are 2 situations that would be great to have a better view on: 1. When the underlying Kubernetes pod runs out of/over resource requirements the Pod gets killed by the GKE cluster but the job in Prefect Cloud remains in the state of

running

. Nothing is ever messaged back to the logs and it never

fails

, some sort of response or indication that the pod is no longer active would be great! 2. When you run a job and it never makes it to the actual Flow, the failure comes through in the Prefect Cloud UI but the kubernetes Pod remains active. I'm assuming this is due to the

finished_job_ttl

not getting applied yet, but it is set at the deployment level so not sure why it keeps the pod around on a failure like this.

Hi, Is it correct that one can't run a flow (even locally) if the Prefect API is down??

Where can I find the docs for the cli for 2.0? They seam to be missing from der API references.

Hello all, I would like to ask if you could help me with a problem I'm having in creating a deployment with an SFTP server as remote storage? I'm using Prefect server version 2.6.4 on Windows 10. Specifically, I locally created an example project with the following structure:

C:\USERS\TESTBENUTZER\PREFECT_TEST_PROJECT
├───.idea
│       some files
│           
├───prefect_flows
│       prefect_flows.py
│       
└───prefect_tasks
        prefect_tasks.py

In addition, I created a remote storage block via the UI named

sftp-server

and the base path

<sftp://ACTUAL-SFTP-URL/prefect_flows/api_flow>

. After that, I used the following command to create a deployment:

PS C:\Users\Testbenutzer\prefect_test_project> prefect deployment build "C:\Users\Testbenutzer\prefect_test_project\prefect_flows\prefect_flows.py:api_flow" -n "ftp-test" -q "ftp-test" -sb "remote-file-system/sftp-server"

However, only the folder

prefect_flows

with the file

prefect_flows.py

is uploaded to the

api_flow

directory of the SFTP server, not the folder

prefect_tasks

. Also, Prefect gives the following error:

File "C:\Users\Testbenutzer\anaconda3\envs\prefect_test_project_v2\lib\site-packages\fsspec\callbacks.py", line 65, in relative_update
  self.value += inc
TypeError: unsupported operand type(s) for +=: 'int' and 'NoneType'
An exception occurred.

I would write the full traceback into the thread, as well as the code of the two Python files and a screenshot of the storage block definition (without the actual SFTP URL and login credentials). If anyone had any idea, what I could be doing wrong here, that would be a great help! Many thanks in advance!

hello, anyway to specify the parameter info in the deployment.yaml through the CLI, seems just can be modified after

prefect deployment build

?

Does .prefectignore support negative cases? This did not seem to work

dbt_projects/*
!dbt_projects/.gitkeep

Trying to calculate if a task has failed and I am doing it all wrong. How would I go about it? Right now I have

isinstance(blah, Failed)

but Failed isn't a type. Here is a code snippet:

if isinstance(bq_result, LoadJob) and bq_result.state == "DONE":
    return Completed(message="Load Finished!")
elif isinstance(bq_result, Failed):
    slack_webhook_block = SlackWebhook.load("data-pipeline-notifications")
    slack_webhook_block.notify("Hello from Prefect 2.0!")
    return bq_result
else:

Hello, Is there any guide on how you are able to prepare a blob storage in azure to work with as an storage block? thanks in advance

Hey, I’m struggling in 2.0 without a Flow State Handler and wondering what the best approach might be to replicate my 1.0 logic for a 2.0 flow. In 1.0, I had a flow state handler which did the following pseudo logic: • if flow just started: ◦ send slack message • if flow cancelled: ◦ make post request to an endpoint which does thing A • if flow successful and some other params are true: ◦ make post request to an endpoint which does thing B • once flow finished (failed or successful) ◦ post to slack with a custom message including task state counts for failed/success etc (queried from the backend) I know there are Notifications in 2.0 for state changes, but they can’t do the custom logic I need so wondering if there are any suggestions? FWIW I really, really miss state change handers - they were incredibly powerful customisation tools 🙂 I have a feeling you’ll tell me I need to create a subflow and put all this logic in the parent flow, but hoping to avoid that as it seems overkill… hoping someone has some ideas..! 🙏

Is it possible to load a

Block

inside the class in which another

Block

is defined using the Python API?

class Foo(Block):

    target_name: str
    _block_type_slug = "foo-block"

    def block_initialization(self) -> None:
        bar_block = JSON.load("bar-block").value
        another_block = JSON.load(target_name).value

I have tried with the above code but I get the following warning, which results in a coroutine being returned instead of a block:

RuntimeWarning: coroutine 'Block.load' was never awaited

Hi All, just upgraded to an Org account and reconfiguring K8's to use the new service accounts - I cant get my k8s agent to start, the agent looks to be connected fine as it writes out the api url but then 404's on getting the {url}/work_queues

Agent started! Looking for work from queue(s): default-work-queue...Mon, Oct 24 2022 6:17:45 pm17:17:45.536 | ERROR | prefect.agent - Failed to create work queue 'default-work-queue'.Mon, Oct 24 2022 6:17:45 pm

prefect.exceptions.PrefectHTTPStatusError: Client error '404 Not Found' for url '<https://api.prefect.cloud/api/accounts/9396b5ff-23c0-42b4-91d9-5892450xxxx/workspaces/de803906-c6f7-45%0A77-a3aa-d300b56fxxx/work_queues/>'

This feels smelly.... can anyone suggest a better way to catch rnadom exceptions in your Prefect 2.0 flow?

if __name__ == "__main__":
    try:
        flow_result = main()
    except Exception as e:
        slack_webhook_block = SlackWebhook.load("data-pipeline-notifications")
        slack_webhook_block.notify(f"{e.__class__}: {str(e)}")

Hmmm... how would I get the flow name while inside a flow in Prefect 2.0? Is that still something in a context somewhere?

I just come into a strange problem. I used to run

rm -rf ~/.prefect/storage/*

to clear cache so that the task would run again next time. But I find it won't work this time and I have to clean the whole

~/.prefect

directory instead of just storage to make it work. Is this a bug or I just miss something? Does prefect provide some command line interface to clear cache?

Hi Team, We came across this error twice in a month, and it makes us worry about the reliability of Prefect. Since Prefect is the backbone of our infrastructure, everything stops working when this happens. The retries and everything simply fails and all our flows fail until this issue persists. Our deployment: Prefect Cloud (1.0) + GCP Kubernetes + LocalDaskExecutor What can be done to safeguard against this? Is there a reliable way to retry and not fail? Task Run for Reference. CC: @Christina Lopez @Yash Joshi

[20 Oct 2022 12:22pm]: Error during execution of task: SSLError(MaxRetryError("HTTPSConnectionPool(host='<http://api.prefect.io|api.prefect.io>', port=443): Max retries exceeded with url: / (Caused by SSLError(SSLEOFError(8, 'EOF occurred in violation of protocol (_ssl.c:1131)')))"))