Hi all, is there any way to specify the path that the manifest file will be generated into?

I just found out that for a work queue to pick up a flow run, all of the the work queue's tags must present in the deployment. For example • if my work queue has

tag_1

,

tag_2

, and the deployment only has

tag_1

, it wont be picked up • If I modify the deployment to have both

tag_1

,

tag_2

, it will be picked up • If I add another

tag_3

to the deployment, the deployment is also picked up Is this the expected behavior?

If there was any thoughts or ideas to separate prefect.client (to interact with REST Prefect API) from server and whole library? Prefect is a very huge package & if you want only rest api client it is no make sense to download all those dependencies that is important, for example, in lambdas where you care about package size. I mean REST API client usually used in 'users' of Prefect, not in server, and I don't want load all server things with it

I'm looking into the CI/CD process to deploy new & update workflows. A few questions: • When I update the deployment setup from the UI (like tags, schedules...) where is the updated info stored? Is it possible to sync it back to the original YAML file? • Follow up from the first point. If a deployment with

name_a

already existed with some modified configs, and now I run

deployment apply

on the same original deployment file, I will lose all of the configs. Is there any way to avoid this, or what's the best practice when handling deployments?

Hi all, Can anyone tell me how to pass python dependence to Github storage? and is there a way we can pass pyproject.toml file to Github storage to install dependencies?

Hey, we are *currently experiencing some performance problems with prefect 2. As soon as we decorate a function with

@flow

it runs slower by a factor of 2-10x. This is our test setup:

def run_flow():
    calculate_something()


@flow(
    name=FLOW_NAME,
    task_runner=get_default_task_runner(),
    version=get_file_hash(__file__),
    timeout_seconds=get_default_timeout(),
)
def run_prefect_flow():
    global USE_PREFECT_LOGGER
    USE_PREFECT_LOGGER = True
    run_flow()


if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == "--no-prefect":
        # Normal performance 
        run_flow()
    else:
        # Bad performance
        run_prefect_flow()

We can't share any code from inside

calculate_something

- are there any circumstances that you know of where this might happen? What overhead brings `@flow`into play? Does it analyze http or DB requests for debugging purposes which might explain the performance degradation? What I can offer is a pstats profile/graph in a dm because it also may contain sensitive data. *currently = as long as we can remember using prefect 2.

Hello! What is a replacement for prefect.signals from 1.0? I used it a lot, especially for skipping unnecessary tasks. Should I do it in a native pythonic way, like passing None as a result of a task to show that there are no need to do something in downstream tasks? The second question is about manually setting downstream/upstream tasks. I was able to do it with

.set_upstream()

method in 1.0. But it looks like there are no more such method. I.e. there are two tasks: loading data into some BigQuery table and then merging it with another table. Obviously these tasks need to be run in exact order. But there are no need to pass any results from the first task to the second, so it wasn’t obvious for Prefect 1.0 that there is some kind of order. That’s why I used

.set_upstream()

. What should i do now? And the third question is about

.map()

for tasks. So, now i should just use standard pythonic map? Will it be parallelized in the same manner as it was in Prefect 1.0 with Concurrent/Parallel task runner?

Hi, a quick question for registering flow: within the same folder, we have

c_flow.py

which orchestrates

b_flow.py

and

a_flow.py

, when we register all three flows together under this folder, will Prefect know the order of register? (like need to register b_flow and a_flow first, and then register c_flow)

Does anyone have experience using map() to call a function taking a **kwargs parameter? I'm not sure if this is even a Prefect question or more of a Python question, but it's not obvious to me how I can get this to work, if it is at all possible. Specifically, I am trying to call trigger_dbt_cli_command which is part of the prefect-dbt package, on prefect 2.0.3. I'm not sure what other information I should include, since the types of error (or error messages) I get varies depending on how I try to approach the problem. 😅

Hey, I have have two questions about logging in Prefect Orion (already read through the docs 1. Is there also an option to have all logs coming out of Docker, i.e. the Python process in the UI? I know, there’s

PREFECT_LOGGING_EXTRA_LOGGERS

and it works, but for example it does not cover logs, like coming from the Tensorflow CPP logs, as e.g.:

2022-08-08 13:59:27.431966: W tensorflow/stream_executor/platform/default/dso_loader.cc:64] Could not load dynamic library 'libcudart.so.11.0'; dlerror: libcudart.so.11.0: cannot open shared object file: No such file or directory
2022-08-08 13:59:27.432037: I tensorflow/stream_executor/cuda/cudart_stub.cc:29] Ignore above cudart dlerror if you do not have a GPU set up on your machine.

Plus, it’s kind of tedious collecting all the libs I want to include. 2. When starting a flow run, many valuable logs and especially errors are only visible on the work queue. So if e.g. I misconfigured a block and the deployment run fails to start, an end user cannot see the error message, since the queue logs are not visible. E.g. one would need to check the work queue Kubernetes pod and manually filter out the error logs for yourself. A user might not have access and with many flow runs, this is also hard. Am I missing something, or how should one work with that properly?

Hi, How to deal with tasks which output multiple outputs. Currently, it seems if I use those outputs as inputs elsewhere Prefect won't detect the dependency.

Hi! I suddenly get a strange intermittent failure on one of my flows:

Error during execution of task: ValueError('A secret name must be provided.')

The secret task runs for 18 minutes (?) and it downstream tasks succeeds somehow, see image. I am on Prefect 1.2.4

Hello community, I have, what I feel is a rather stupid question for prefect 1.x. I have a simple task, that needs an array as an argument (the list can be of arbitrary length and mixed types in my example, but using

List[int]

for simplicity here)

@task
def test_task(lst: List[int]):
    log = prefect.context.logger
    val = sum(lst)
    <http://log.info|log.info>(val)
    return val

What I’d like to do is to run a mapped task over, but I’m not entirely sure how to invoke it using

unmapped

(if this is even possible). When I call it like:

with Flow("test flow") as flow:
    p = Parameter("p", default=[[1, 2, 3], [2, 3, 4]])
    test_task.map(p)

it works as expected and the

test_task

is called with:

test_task([1,2,3])
test_task([2,3,4])

I’d like to achieve the following to be run, but with the use of unmapped task:

test_task(10, 1, 2)
test_task(20, 1, 2)
test_task(30, 1, 2)

so I tried

with Flow("test flow") as flow:
    p = Parameter("p", default=[10, 20, 30])
    test_task.map([p, unmapped(1), unmapped(2)])

this one fails with

int not being iterable

exception, while if I remove the array like this:

with Flow("test flow") as flow:
    p = Parameter("p", default=[10, 20, 30])
    test_task.map(p, unmapped(1), unmapped(2))

I’m getting “too many positional arguments error” I dont know how better should I put it to get what I need. Other solution is to come up with a task per array size, but that is just a lot. Any advice? Thank you very much for your help!

Hi there, I have just migrated from prefer server to cloud. I am getting this

ERROR
prefect-server.ZombieKiller.TaskRun
No heartbeat detected from the remote task; marking the run as failed.

Can anyone help me with workaround for this? Its trying to run a sql query but timing out. Thanks

Hi all - we are running

prefecthq/prefect:2.0.2-python3.8

and trying to execute the following flow from this tutorial on a Kubernetes cluster pointing to an S3 storage block. When doing so, we receive the following error:

RuntimeError: File system created with scheme 's3' from base path 's3://<mybucket>' could not be created. You are likely missing a Python module required to use the given storage protocol.

It looks like others that have been experiencing this have had to manually run

pip install s3fs

in their execution environment to get S3 external storage working with the Kubernetes execution environment. Is this the recommended deployment pattern for now? If so, is there a plan to start including these dependencies in the prefect filesystems package? It seems strange that we’d need a custom image for something that is supposed to already be tightly integrated with Prefect 2.0 and a fundamental requirement for using k8s infrastructure.

import prefect
from prefect import task, flow, get_run_logger
from prefect.filesystems import S3

s3_block = S3.load("aws-s3")


@task
def hello_world():
    logger = get_run_logger()
    text = "hello from orion_flow!"
    <http://logger.info|logger.info>(text)
    return text


@flow(name="orion_flow")
def orion_flow():
    logger = get_run_logger()
    <http://logger.info|logger.info>("Hello from Kubernetes!")
    hw = hello_world()
    return

Dear Prefect Community, Is it possible to run Prefect Orion (2.0) UI using https:// instead of http://. For example instead of running http://server_host:4200 can we run https://sever_host:4443. If so, what kind of configuration I have to do? Please provide instructions. Thanks, Xavier Babu

Hi. I'm having trouble running a deployment with docker infrastructure and s3 storage. I think it has to do with the aws region. Our S3 is in a special east region; how do we configure docker to use the aws s3 config on the server that has the region?

Flow could not be retrieved from deployment.
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/s3fs/core.py", line 646, in _lsdir
    async for i in it:
  File "/usr/local/lib/python3.8/site-packages/aiobotocore/paginate.py", line 32, in __anext__
    response = await self._make_request(current_kwargs)
  File "/usr/local/lib/python3.8/site-packages/aiobotocore/client.py", line 265, in _make_api_call
    raise error_class(parsed_response, operation_name)
botocore.exceptions.ClientError: An error occurred (InvalidAccessKeyId) when calling the ListObjectsV2 operation: The AWS Access Key Id you provided does not exist in our records.

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/prefect/engine.py", line 247, in retrieve_flow_then_begin_flow_run
    flow = await load_flow_from_flow_run(flow_run, client=client)
  File "/usr/local/lib/python3.8/site-packages/prefect/client.py", line 104, in with_injected_client
    return await fn(*args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/prefect/deployments.py", line 47, in load_flow_from_flow_run
    await storage_block.get_directory(from_path=None, local_path=".")
  File "/usr/local/lib/python3.8/site-packages/prefect/filesystems.py", line 373, in get_directory
    return await self.filesystem.get_directory(
  File "/usr/local/lib/python3.8/site-packages/prefect/filesystems.py", line 251, in get_directory
    return self.filesystem.get(from_path, local_path, recursive=True)
  File "/usr/local/lib/python3.8/site-packages/fsspec/asyn.py", line 111, in wrapper
    return sync(self.loop, func, *args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/fsspec/asyn.py", line 96, in sync
    raise return_result
  File "/usr/local/lib/python3.8/site-packages/fsspec/asyn.py", line 53, in _runner
    result[0] = await coro
  File "/usr/local/lib/python3.8/site-packages/fsspec/asyn.py", line 536, in _get
    rpaths = await self._expand_path(rpath, recursive=recursive)
  File "/usr/local/lib/python3.8/site-packages/fsspec/asyn.py", line 736, in _expand_path
    out = await self._expand_path([path], recursive, maxdepth)
  File "/usr/local/lib/python3.8/site-packages/fsspec/asyn.py", line 754, in _expand_path
    rec = set(await self._find(p, maxdepth=maxdepth, withdirs=True))
  File "/usr/local/lib/python3.8/site-packages/s3fs/core.py", line 718, in _find
    out = await self._lsdir(path, delimiter="", prefix=prefix)
  File "/usr/local/lib/python3.8/site-packages/s3fs/core.py", line 669, in _lsdir
    raise translate_boto_error(e)
PermissionError: The AWS Access Key Id you provided does not exist in our records.

hey Prefect Community, has anyone had any luck with hosting/running Prefect 2.0 in Kubernetes? In particular the UI? We were able to get prefect 2.0 running in Kubernetes using the prefect kubernetes command to generate a manifest deployment yaml, but the issue we're facing is the UI is making requests calls to 127.0.0.1. To complicate matters... we have a kubernetes ingress setup to handle ssl termination for us

Hi! I think I found a bug when mixing sync and async tasks in an async flow, for example, when running the next example:

import asyncio

from prefect import task, get_run_logger, flow


@task
async def task1():
    await asyncio.sleep(1)


@task
def task2():
    logger = get_run_logger()
    <http://logger.info|logger.info>("hi")


@flow
async def test_flow():
    await task1.submit()
    task2.submit()

The following exception is raised:

Encountered exception during execution:
Traceback (most recent call last):
  File "/opt/conda/lib/python3.10/site-packages/prefect/engine.py", line 550, in orchestrate_flow_run
    result = await flow_call()
  File "/app/src/streamcast/agent/flows/test.py", line 22, in test_flow
    task2.submit()
  File "/opt/conda/lib/python3.10/site-packages/prefect/tasks.py", line 491, in submit
    return enter_task_run_engine(
  File "/opt/conda/lib/python3.10/site-packages/prefect/engine.py", line 691, in enter_task_run_engine
    return flow_run_context.sync_portal.call(begin_run)
  File "/opt/conda/lib/python3.10/site-packages/anyio/from_thread.py", line 283, in call
    return cast(T_Retval, self.start_task_soon(func, *args).result())
  File "/opt/conda/lib/python3.10/concurrent/futures/_base.py", line 439, in result
    return self.__get_result()
  File "/opt/conda/lib/python3.10/concurrent/futures/_base.py", line 391, in __get_result
    raise self._exception
  File "/opt/conda/lib/python3.10/site-packages/anyio/from_thread.py", line 219, in _call_func
    retval = await retval
  File "/opt/conda/lib/python3.10/site-packages/prefect/engine.py", line 783, in create_task_run_then_submit
    task_run = await create_task_run(
  File "/opt/conda/lib/python3.10/site-packages/prefect/engine.py", line 823, in create_task_run
    task_run = await flow_run_context.client.create_task_run(
  File "/opt/conda/lib/python3.10/site-packages/prefect/client.py", line 1635, in create_task_run
    response = await <http://self._client.post|self._client.post>(
  File "/opt/conda/lib/python3.10/site-packages/httpx/_client.py", line 1842, in post
    return await self.request(
  File "/opt/conda/lib/python3.10/site-packages/httpx/_client.py", line 1527, in request
    return await self.send(request, auth=auth, follow_redirects=follow_redirects)
  File "/opt/conda/lib/python3.10/site-packages/prefect/client.py", line 258, in send
    await super().send(*args, **kwargs)
  File "/opt/conda/lib/python3.10/site-packages/httpx/_client.py", line 1614, in send
    response = await self._send_handling_auth(
  File "/opt/conda/lib/python3.10/site-packages/httpx/_client.py", line 1642, in _send_handling_auth
    response = await self._send_handling_redirects(
  File "/opt/conda/lib/python3.10/site-packages/httpx/_client.py", line 1679, in _send_handling_redirects
    response = await self._send_single_request(request)
  File "/opt/conda/lib/python3.10/site-packages/httpx/_client.py", line 1716, in _send_single_request
    response = await transport.handle_async_request(request)
  File "/opt/conda/lib/python3.10/site-packages/httpx/_transports/default.py", line 353, in handle_async_request
    resp = await self._pool.handle_async_request(req)
  File "/opt/conda/lib/python3.10/site-packages/httpcore/_async/connection_pool.py", line 253, in handle_async_request
    raise exc
  File "/opt/conda/lib/python3.10/site-packages/httpcore/_async/connection_pool.py", line 237, in handle_async_request
    response = await connection.handle_async_request(request)
  File "/opt/conda/lib/python3.10/site-packages/httpcore/_async/connection.py", line 90, in handle_async_request
    return await self._connection.handle_async_request(request)
  File "/opt/conda/lib/python3.10/site-packages/httpcore/_async/http11.py", line 105, in handle_async_request
    raise exc
  File "/opt/conda/lib/python3.10/site-packages/httpcore/_async/http11.py", line 84, in handle_async_request
    ) = await self._receive_response_headers(**kwargs)
  File "/opt/conda/lib/python3.10/site-packages/httpcore/_async/http11.py", line 148, in _receive_response_headers
    event = await self._receive_event(timeout=timeout)
  File "/opt/conda/lib/python3.10/site-packages/httpcore/_async/http11.py", line 177, in _receive_event
    data = await self._network_stream.read(
  File "/opt/conda/lib/python3.10/site-packages/httpcore/backends/asyncio.py", line 33, in read
    return await self._stream.receive(max_bytes=max_bytes)
  File "/opt/conda/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 1265, in receive
    await self._protocol.read_event.wait()
  File "/opt/conda/lib/python3.10/asyncio/locks.py", line 211, in wait
    fut = self._get_loop().create_future()
  File "/opt/conda/lib/python3.10/asyncio/mixins.py", line 30, in _get_loop
    raise RuntimeError(f'{self!r} is bound to a different event loop')
RuntimeError: <asyncio.locks.Event object at 0x7fd237ee4310 [unset]> is bound to a different event loop

I'm trying to get a prefect 2 agent running on ecs using: https://github.com/PrefectHQ/prefect-recipes/tree/main/aws/prefect. I might be trolling, but where can I find my [ACCOUNT-ID] for the api url? I was able to find the [WORKSPACE-ID] through the attached screenshot, but idk how to found the [ACCOUNT-ID]... Here's the docs I'm following: https://docs.prefect.io/concepts/settings/#prefect_api_url.

Hi guys, porting my code to prefect 2.0 and I have got it running local, but when creating a deployment to run it on an agent I'm getting this error on the UI. Deployment cmd:

PYTHONPATH=/home/alburati/Proyectos/pipeline/src prefect deployment build src/sources/biorxiv/flows/biorxiv_main_flow.py:biorxiv_main_flow -n 'source_flow' -t source_work

Error:

Flow could not be retrieved from deployment.
Traceback (most recent call last):
  File "/home/alburati/anaconda3/envs/nlp_env/lib/python3.7/site-packages/prefect/engine.py", line 246, in retrieve_flow_then_begin_flow_run
    flow = await load_flow_from_flow_run(flow_run, client=client)
  File "/home/alburati/anaconda3/envs/nlp_env/lib/python3.7/site-packages/prefect/client.py", line 104, in with_injected_client
    return await fn(*args, **kwargs)
  File "/home/alburati/anaconda3/envs/nlp_env/lib/python3.7/site-packages/prefect/deployments.py", line 325, in load_flow_from_flow_run
    await storage_block.get_directory(from_path=None, local_path=".")
  File "/home/alburati/anaconda3/envs/nlp_env/lib/python3.7/site-packages/prefect/filesystems.py", line 95, in get_directory
    shutil.copytree(from_path, local_path)
  File "/home/alburati/anaconda3/envs/nlp_env/lib/python3.7/shutil.py", line 324, in copytree
    os.makedirs(dst)
  File "/home/alburati/anaconda3/envs/nlp_env/lib/python3.7/os.py", line 223, in makedirs
    mkdir(name, mode)
FileExistsError: [Errno 17] File exists: '.'

Any suggestions?

Hi. This is a new error I got. Anyone know why it just started appearing? Fails like this once out every 100 tasks. Anyone know why it just started appearing?

Flow could not be retrieved from deployment.
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/dist-packages/prefect/engine.py", line 247, in retrieve_flow_then_begin_flow_run
    flow = await load_flow_from_flow_run(flow_run, client=client)
  File "/usr/local/lib/python3.9/dist-packages/prefect/client.py", line 104, in with_injected_client
    return await fn(*args, **kwargs)
  File "/usr/local/lib/python3.9/dist-packages/prefect/deployments.py", line 47, in load_flow_from_flow_run
    await storage_block.get_directory(from_path=None, local_path=".")
  File "/usr/local/lib/python3.9/dist-packages/prefect/filesystems.py", line 98, in get_directory
    shutil.copytree(from_path, local_path, dirs_exist_ok=True)
  File "/usr/lib/python3.9/shutil.py", line 568, in copytree
    return _copytree(entries=entries, src=src, dst=dst, symlinks=symlinks,
  File "/usr/lib/python3.9/shutil.py", line 522, in _copytree
    raise Error(errors)
shutil.Error: [('/home/app/app/.prefect/orion.db-shm', './.prefect/orion.db-shm', "[Errno 2] No such file or directory: '/home/app/app/.prefect/orion.db-shm'")]

Hi, I'm using prefect1.0 on Cloud and I'm trying to save the flow's information like ID, excuetion time, version, e.t.c to my own database. Is there an API already provided by prefect to do that? If not, isn't there a good way to save the corresponding contents using the functions that have already been created? Give me a hand... 🙌

Is there a way to programmatically create secrets in prefect 2.0? It is easy to load secrets that were created via the UI but it would be useful to be able to write secrets from python.

can we use github for cloud storage block in prefect 2.0 ?

Hi all, using prefect 2.0.3, is there a way to forcibly expire a cache key?

Hi guys, can I use S3 block to download files from S3 bucket? or am I just using the block as secret to the AWS S3Download task? any examples?

Hi, I am trying out prefect 2 to schedule flows. I want to dynamically schedule a flow with different input arguments and schedules (using api or python client) through code. Should I create deployment for each combination (input and schedule)? Earlier we are using

Deployment

object for creating flows, now it is deprecated. Is there any other way to run flows dynamically with schedule. Haven't found any way to run flows dynamically through code.