Hello, is there a way to render markdown written in deployment.yaml:description in the Prefect Orion Cloud UI?

Do we have docker-compose for prefect 2.0 ?

How can we logout of prefect cloud:2.0 and switch back to prefect UI? I am running Orion start command but queue is created on cloud

Hi all~ I'm using Prefect1.0. Is there any method that trigger to immeadiatly excute the flow, if the certain task is finished in the app we developed? NOT trigger from the task.

Version:             2.0.2

API version:         0.8.0

Python version:      3.9.13

I having problems with both the

read_flow_by_name

and

read_deployment_by_name

. Neither of these method is working in when i try to use them: Sample code:

from prefect.client import get_client

import asyncio

import json

async def main():

async with get_client() as client:

data = await client.read_flow_by_name(flow_name='test_flow')

loop = asyncio.get_event_loop()

loop.run_until_complete(main())

Error:

Traceback (most recent call last):

File

data = await client.read_flow_by_name(flow_name='test_flow')

/.venv/lib/python3.9/site-packages/prefect/client.py", line 508, in read_flow_by_name

return schemas.core.Deployment.parse_obj(response.json())

File "pydantic/main.py", line 521, in pydantic.main.BaseModel.parse_obj

File "pydantic/main.py", line 341, in pydantic.main.BaseModel.__init__

pydantic.error_wrappers.ValidationError: 2 validation errors for Deployment

flow_id

field required (type=value_error.missing)

manifest_path

field required (type=value_error.missing)

Hi all, I'm having a look at prefect-email, and wonder if there is a better way to securely store the password rather than having it hard coded into the Python scripts? https://github.com/PrefectHQ/prefect-email Thank you 🙂

Is anyone else on 2.0.2 yet? I'm getting an interesting error when I try to deploy using a remote storage block (seems to work with a local file system). Code: prefect deployment build orchestration/flows/test_cleansed.py:default -n test_cleansed -t dev --storage-block azure/azure-dev --infra docker-container -o test_cleansed-deployment.yaml Error: TypeError: put_directory() got an unexpected keyword argument 'ignore_file' on line 470 of cli/deployment.py Judging by the source code in Github, it seems like the overload of put_directory() for the Azure storage block specifically omits the ignore_file argument. Perhaps these two features were not tested together, unless I'm missing something. I'll give people an hour to say that I'm doing something wrong, otherwise I'll create an issue (or a PR, might be an easy fix).

Hi all, (working on prefect 2.0.2) I am running a deployment flow on scheduled interval. I have 2 agents listening on the same work-queue and grabbing this flow. One agent gets an error : prefect.exceptions.Abort: This run has already terminated. ---> The flow is considered "Crashed". Is it the expected behavior ? How is it possible to split the load on multiple agents for Scheduled deployments ? Cheers

I'm trying to find how to use kubernetes tasks on prefect 2, is that possible? (I mean same as in prefect 1)

getting server timeout when running the deployment from cloud ui 100941.192 | DEBUG | prefect.agent - Checking for flow runs... 100946.444 | DEBUG | prefect.agent - Checking for flow runs... 100951.687 | DEBUG | prefect.agent - Checking for flow runs... 100956.936 | DEBUG | prefect.agent - Checking for flow runs... 101002.188 | DEBUG | prefect.agent - Checking for flow runs... 101007.439 | DEBUG | prefect.agent - Checking for flow runs... 101012.683 | DEBUG | prefect.agent - Checking for flow runs... 101018.003 | DEBUG | prefect.agent - Checking for flow runs... 101023.248 | DEBUG | prefect.agent - Checking for flow runs... 101028.530 | DEBUG | prefect.agent - Checking for flow runs... 101028.709 | INFO | prefect.agent - Submitting flow run '62df0689-3a3d-4bb0-bcaa-a16fb215df67' 101029.536 | ERROR | prefect.agent - Flow runner failed to submit flow run '62df0689-3a3d-4bb0-bcaa-a16fb215df67' Traceback (most recent call last): File "/usr/local/lib/python3.8/site-packages/prefect/agent.py", line 169, in submit_run await self.task_group.start(submit_flow_run, flow_run, infrastructure) File "/usr/local/lib/python3.8/site-packages/anyio/_backends/_asyncio.py", line 701, in start return await future File "/usr/local/lib/python3.8/site-packages/prefect/infrastructure/submission.py", line 48, in submit_flow_run return await infrastructure.run(task_status=task_status) File "/usr/local/lib/python3.8/site-packages/prefect/infrastructure/kubernetes.py", line 201, in run job_name = await run_sync_in_worker_thread(self._create_job, manifest) File "/usr/local/lib/python3.8/site-packages/prefect/utilities/asyncutils.py", line 56, in run_sync_in_worker_thread return await anyio.to_thread.run_sync(call, cancellable=True) File "/usr/local/lib/python3.8/site-packages/anyio/to_thread.py", line 28, in run_sync return await get_asynclib().run_sync_in_worker_thread(func, *args, cancellable=cancellable, File "/usr/local/lib/python3.8/site-packages/anyio/_backends/_asyncio.py", line 818, in run_sync_in_worker_thread return await future File "/usr/local/lib/python3.8/site-packages/anyio/_backends/_asyncio.py", line 754, in run result = context.run(func, *args) File "/usr/local/lib/python3.8/site-packages/prefect/infrastructure/kubernetes.py", line 408, in _create_job job = batch_client.create_namespaced_job(self.namespace, job_manifest) File "/usr/local/lib/python3.8/site-packages/kubernetes/client/api/batch_v1_api.py", line 210, in create_namespaced_job return self.create_namespaced_job_with_http_info(namespace, body, **kwargs) # noqa: E501 File "/usr/local/lib/python3.8/site-packages/kubernetes/client/api/batch_v1_api.py", line 309, in create_namespaced_job_with_http_info return self.api_client.call_api( File "/usr/local/lib/python3.8/site-packages/kubernetes/client/api_client.py", line 348, in call_api return self.__call_api(resource_path, method, File "/usr/local/lib/python3.8/site-packages/kubernetes/client/api_client.py", line 180, in __call_api response_data = self.request( File "/usr/local/lib/python3.8/site-packages/kubernetes/client/api_client.py", line 391, in request return self.rest_client.POST(url, File "/usr/local/lib/python3.8/site-packages/kubernetes/client/rest.py", line 275, in POST return self.request("POST", url, File "/usr/local/lib/python3.8/site-packages/kubernetes/client/rest.py", line 234, in request raise ApiException(http_resp=r) kubernetes.client.exceptions.ApiException: (500) Reason: Internal Server Error HTTP response headers: HTTPHeaderDict({'Audit-Id': 'c94033e6-af19-414c-8c50-72811876a885', 'Cache-Control': 'no-cache, private', 'Content-Type': 'application/json', 'X-Kubernetes-Pf-Flowschema-Uid': '1133f237-04af-4269-a466-2f5eb69e6670', 'X-Kubernetes-Pf-Prioritylevel-Uid': 'e6f943c5-ae55-45d1-ab0e-cb0161b8bd0e', 'Date': 'Wed, 03 Aug 2022 101029 GMT', 'Content-Length': '264'}) HTTP response body: {"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure","message":"The POST operation against Job.batch could not be completed at this time, please try again.","reason":"ServerTimeout","details":{"name":"POST","group":"batch","kind":"Job"},"code":500}

Hi people, Like ShellTask function in Prefect v1 (https://docs-v1.prefect.io/api/latest/tasks/shell.html#shelltask) , Is there anyway to call upon a bash script in Prefect 2.0 using any other function as I cant find any documentation for it. Any Suggestion?

Hi, How I make SSL connection while connecting to postgres db?

I know asyncpg has way to pass sslmode, ssl sertificate

Hi I’ve managed to get a POC for Prefect up and running in Heroku but issue I have now is that when I go to run the flow that run dbt, it doesn’t appear to be picking up the env var for the dbt target and is instead falling back on the dev one. I’m currently creating a task like so

dbt_shell_task = DbtShellTask(
    profiles_dir=str(DBT_PROFILES_DIR),
    environment=str(DBT_TARGET),
    overwrite_profiles=False,
    helper_script=f"cd {DBT_ROOT}",
    return_all=True,
    stream_output=True,
    state_handlers=[post_to_slack],
)

If I run dbt outside Prefect, i.e.

dbt build --target=$DBT_TARGET

it works as expected. Am I using the wrong variable for

target

?

Hi guys, has anyone done testing for mapped flow? I'm trying to write a test to verify if a task is available in the flow but I can't seem to access the tasks that are inside the mapped flow

Hi everyone, With prefect 2.0 : Is it possible to use prefect.inrastructure to run a docker on Azure outside the agent ? My VM where I install the agent have 1Go of RAM, but I want to set up a docker task (like FARGATE on AWS) with more ressources to execute the flow. Or should I go straight to kubernetes ? Someone have tried Kubernetes on Azure for this ?

Hi all, How can I get more descriptive names of tasks like "create_flow_run" & "wait_for_flow" ? Otherwise in prefect cloud it's hard to understand which child flow started or not ...

I am running flows on prefect 1.0 with an ECS Agent. The flow is successfully scheduled/submitted then gets stuck in that state. I looked on ECS and it looks like the container starts then immediately dies. There aren’t any associated logs that I can find. Any recommendations on how to debug this? EDIT - ended up being a 1.0 agent/2.0flow compatibility issue. For brevity the ECS container stop reason was

essential container in task exited

with no generated logs. The entrance command

["/bin/sh", "-c", "prefect execute flow-run"]

was failing because (I think) prefect 2.0 doesn’t have an

execute

command. I resolved this by pinning my prefect dependency to

pip install "prefect[aws, gitlab]==1.3.0"

.

Prefect : 2.0.2 I added the "version" parameter to my flow, but it doesn't show up in the UI. When I go to retrieve the flow model version is not included either.

from prefect import flow, get_run_logger

@flow(name="test", version="1.0beta")
def main():
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Hello World")

if __name__ == "__main__":
    main()

Flow model details:

id=UUID('120dea46-79e8-4f42-99f0-7e300a2b3cb9') name='test' tags=[]

This is big blocker for me as I want to the

create_flow_run

method, but it errors out after you give it a flow model without a version in it. Any suggestion on what i am doing wrong here?

hi, wanna ask a question if someone may know or have some clues 😉 Appreciate for any help!🙌 In a flow run, we defined some parameters using

MyParams = Parameter('My_Param', default="my_input")

, in Prefect UI we can input some parameters to overwrite the default at run time, my question is: is there a way to get the run time parameter input (not the MyParams object, but the actual input)? In our case, the

default

is a list (where we pass a list of ids), we need to customize the flow run name by extracting each id from

MyParams

and formalizing a new list (child flow run name) for

map

function to work. From what I understand and tested,

MyParams

is a serialized object, to access its input first needs to deserialize it or using another way, we cannot just loop over it using

[i+'_customzied_string' for i in MyParams]

Shoot, where did the TaskRunners move to? trying to update from Prefect 2.0b6 code to 2.0.2 and I cant find references to where they moved

Second question, I need to turn off checkpointing for a task, how would I do that in Prefect 2.0?

Me again, Thanks for that announcement of Prefect <> DBT. Could we add python typing in the docs (and in general)? For example, it's hard to understand what

result

is about ... Thanks

Is anyone using SSL connection for Postgres Database with prefect?

Hi all! In Prefect 2.0, what is the recommended way to have a task/flow call an existing docker image? The flow I'm writing takes some data, handles it with some tasks, feeds it to an external docker image, and then does something more with the output from the image. Been digging through the docs for a while, but I'm not really sure how to approach this. It's all running on my local machine. Thanks!

Hello all -- looking to see if anyone has a working Prefect 2.0 example of what it looks like to use a RemoteFileSystem block through the Cloud UI to setup SMB (including auth -- even if the auth can't be setup in the Cloud UI block itself) storage to use with flows deployment. Context: Our team doesn't have S3, GC, or Azure storage available to us. It all has to be kept on-premise. Our Data Center wants to provide us with SVM storage that's accessible via SMB (and it looks like fsspec does support a "smb://" connection string). Thanks!

I am trying to get my flow to run in a remote Ray (anyscale) cluster. However, my flow is failing whenever I try to submit a task. The error is that a Ray worker is trying to access something that is specific to my local machine (I am running directly on my local machine for development). I think this has something to do with local storage being used even in a distributed context. In the doc for RayTaskRunner, it recommends to use remote storage. I setup an S3 block, but then I have no idea what I suppose to do next. My flow is still using Local storage.

from prefect import flow, task, get_run_logger
from prefect_ray import RayTaskRunner


@task
def generate_numbers(n: int) -> list[int]:
    return list(range(n))


@task
def double_the_numbers(d: int) -> int:
    return d * 2


@flow(
    task_runner=RayTaskRunner(
        address="<anyscale://singularity/demo1>",
        init_kwargs={"runtime_env": {"working_dir": "."}},
    )
)
def run_workflow():
    logger = get_run_logger()
    <http://logger.info|logger.info>("Hello World!")

    numbers = generate_numbers.submit(100)
    <http://logger.info|logger.info>(numbers.result())
    doubled = double_the_numbers.map(numbers)

    <http://logger.info|logger.info>([d.wait().result() for d in doubled])


run_workflow()

ray::begin_task_run() (pid=20006, ip=<http://xxx.xx.xx.xxx|xxx.xx.xx.xxx>)
  File "/opt/homebrew/Caskroom/miniforge/base/envs/singularity2/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 193, in wrapper
  File "/home/ray/anaconda3/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 140, in run_async_in_new_loop
    return anyio.run(partial(__fn, *args, **kwargs))
  File "/home/ray/anaconda3/lib/python3.10/site-packages/anyio/_core/_eventloop.py", line 70, in run
    return asynclib.run(func, *args, **backend_options)
  File "/home/ray/anaconda3/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 292, in run
    return native_run(wrapper(), debug=debug)
  File "/home/ray/anaconda3/lib/python3.10/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/home/ray/anaconda3/lib/python3.10/asyncio/base_events.py", line 646, in run_until_complete
    return future.result()
  File "/home/ray/anaconda3/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 287, in wrapper
    return await func(*args)
  File "/home/ray/anaconda3/lib/python3.10/site-packages/prefect/engine.py", line 946, in begin_task_run
    return await orchestrate_task_run(
  File "/home/ray/anaconda3/lib/python3.10/site-packages/prefect/engine.py", line 1083, in orchestrate_task_run
    await _persist_serialized_result(
  File "/home/ray/anaconda3/lib/python3.10/site-packages/prefect/results.py", line 15, in _persist_serialized_result
    await filesystem.write_path(key, content)
  File "/home/ray/anaconda3/lib/python3.10/site-packages/prefect/filesystems.py", line 162, in write_path
    path.parent.mkdir(exist_ok=True, parents=True)
  File "/home/ray/anaconda3/lib/python3.10/pathlib.py", line 1177, in mkdir
    self.parent.mkdir(parents=True, exist_ok=True)
  File "/home/ray/anaconda3/lib/python3.10/pathlib.py", line 1177, in mkdir
    self.parent.mkdir(parents=True, exist_ok=True)
  File "/home/ray/anaconda3/lib/python3.10/pathlib.py", line 1177, in mkdir
    self.parent.mkdir(parents=True, exist_ok=True)
  File "/home/ray/anaconda3/lib/python3.10/pathlib.py", line 1173, in mkdir
    self._accessor.mkdir(self, mode)
PermissionError: [Errno 13] Permission denied: '/Users'

@Taylor Curran @Kalise Richmond

Hi, I was just curious to know how data is passed between tasks in prefect 2. Are the results always passed to the next task directly through memory even if storage is set up to persist them? Or is it just references to the storage that are passed between tasks? Thanks in advance

Hey all, we’re hosting @John Kang on PrefectLive today at 3p Eastern where he’s going to show how to use GCS as a storage for your Prefect workflow! Come join us live on Twitch!

I’m a little confused with the migration from v1 to Orion. Should we continue to use .map? Or write async tasks?