Hi guys, I'am using

prefect 2.3.1

,

prefect-dask: 0.2.0

and got several errors running code similar to this pseudo code :

@task
def dask_context(item):
    name="myblock-name"
    config = await JSON.load(name)

@flow(task_runner=DaskTaskRunner())
def do_stuff():
   with tags("only_4"):
       result_list = dask_context.map(data_list)

I got this Exception multiple times after it running for a while (sometimes minutes, sometimes hours):

Encountered exception during execution:
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/httpcore/backends/asyncio.py", line 33, in read
    return await self._stream.receive(max_bytes=max_bytes)
  File "/usr/local/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 1265, in receive
    await self._protocol.read_event.wait()
  File "/usr/local/lib/python3.9/asyncio/locks.py", line 226, in wait
    await fut
asyncio.exceptions.CancelledError

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/httpcore/_exceptions.py", line 8, in map_exceptions
    yield
  File "/usr/local/lib/python3.9/site-packages/httpcore/backends/asyncio.py", line 35, in read
    return b""
  File "/usr/local/lib/python3.9/site-packages/anyio/_core/_tasks.py", line 118, in __exit__
    raise TimeoutError
TimeoutError

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/httpx/_transports/default.py", line 60, in map_httpcore_exceptions
    yield
  File "/usr/local/lib/python3.9/site-packages/httpx/_transports/default.py", line 353, in handle_async_request
    resp = await self._pool.handle_async_request(req)
  File "/usr/local/lib/python3.9/site-packages/httpcore/_async/connection_pool.py", line 253, in handle_async_request
    raise exc
  File "/usr/local/lib/python3.9/site-packages/httpcore/_async/connection_pool.py", line 237, in handle_async_request
    response = await connection.handle_async_request(request)
  File "/usr/local/lib/python3.9/site-packages/httpcore/_async/connection.py", line 90, in handle_async_request
    return await self._connection.handle_async_request(request)
  File "/usr/local/lib/python3.9/site-packages/httpcore/_async/http11.py", line 105, in handle_async_request
    raise exc
  File "/usr/local/lib/python3.9/site-packages/httpcore/_async/http11.py", line 84, in handle_async_request
    ) = await self._receive_response_headers(**kwargs)
  File "/usr/local/lib/python3.9/site-packages/httpcore/_async/http11.py", line 148, in _receive_response_headers
    event = await self._receive_event(timeout=timeout)
  File "/usr/local/lib/python3.9/site-packages/httpcore/_async/http11.py", line 177, in _receive_event
    data = await self._network_stream.read(
  File "/usr/local/lib/python3.9/site-packages/httpcore/backends/asyncio.py", line 35, in read
    return b""
  File "/usr/local/lib/python3.9/contextlib.py", line 137, in __exit__
    self.gen.throw(typ, value, traceback)
  File "/usr/local/lib/python3.9/site-packages/httpcore/_exceptions.py", line 12, in map_exceptions
    raise to_exc(exc)
httpcore.ReadTimeout

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

Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/prefect/engine.py", line 1185, in orchestrate_task_run
    result = await task.fn(*args, **kwargs)
  File "/tmp/tmpq25d58myprefect/lib/dask_context.py", line 14, in _dask_task
    with await get_context(flow_run.parameters[block_name]):
  File "/tmp/tmpq25d58myprefect/./lib/config_context.py", line 40, in get_context
    config = await JSON.load(name)
  File "/usr/local/lib/python3.9/site-packages/prefect/blocks/core.py", line 606, in load
    block_document = await client.read_block_document_by_name(
  File "/usr/local/lib/python3.9/site-packages/prefect/client.py", line 1268, in read_block_document_by_name
    response = await self._client.get(
  File "/usr/local/lib/python3.9/site-packages/httpx/_client.py", line 1751, in get
    return await self.request(
  File "/usr/local/lib/python3.9/site-packages/httpx/_client.py", line 1527, in request
    return await self.send(request, auth=auth, follow_redirects=follow_redirects)
  File "/usr/local/lib/python3.9/site-packages/prefect/client.py", line 257, in send
    await super().send(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/httpx/_client.py", line 1614, in send
    response = await self._send_handling_auth(
  File "/usr/local/lib/python3.9/site-packages/httpx/_client.py", line 1642, in _send_handling_auth
    response = await self._send_handling_redirects(
  File "/usr/local/lib/python3.9/site-packages/httpx/_client.py", line 1679, in _send_handling_redirects
    response = await self._send_single_request(request)
  File "/usr/local/lib/python3.9/site-packages/httpx/_client.py", line 1716, in _send_single_request
    response = await transport.handle_async_request(request)
  File "/usr/local/lib/python3.9/site-packages/httpx/_transports/default.py", line 353, in handle_async_request
    resp = await self._pool.handle_async_request(req)
  File "/usr/local/lib/python3.9/contextlib.py", line 137, in __exit__
    self.gen.throw(typ, value, traceback)
  File "/usr/local/lib/python3.9/site-packages/httpx/_transports/default.py", line 77, in map_httpcore_exceptions
    raise mapped_exc(message) from exc
httpx.ReadTimeout

I started running this code with a tag concurrency limit of 20 then switched to 4. The more threads the exception was raised faster. Currently running on a server of 44 cores and 256 GB.

Hi - Prefect Experts, Im running into an error that indicates I am unable to schedule a flow that has required parameters. Is that expected? 🧵

Is there something that has to be done to my account to enable login through the CLI? I have created an API key but when I run

prefect cloud login

i get

Unable to authenticate with Prefect Cloud. Please ensure your credentials are correct.

Hi all, I was having issues with the logging in Prefect. I wanted to share my solution in case it helps someone else. Issue 1 - Prefect logger does not work outside of Prefect context Issue 2 - Prefect logger needs to have get_run_logger() in every functions Solution (proof-of-concept): See https://discourse.prefect.io/t/hybrid-logger-a-logger-that-works-both-inside-and-outsite-of-a-prefect-context/1559 @Taylor Curran

Is it possible to get a replica of the database of historical runs etc that Prefect cloud manages? Or something like incremental backups provided to end users

How do I add mapped tasks in the imperative API style? so for example:

flow = Flow("My imperative flow!")

# define some new tasks
name = Parameter("name")
second_add = add.copy()

# add our tasks to the flow
flow.add_task(add)
flow.add_task(second_add)
flow.add_task(say_hello)

# create non-data dependencies so that `say_hello` waits for `second_add` to finish.
say_hello.set_upstream(second_add, flow=flow)

# create data bindings
add.bind(x=1, y=2, flow=flow)
second_add.bind(x=add, y=100, flow=flow)
say_hello.bind(person=name, flow=flow)

If I wanted to give the

add

task a mapped input?

Hi all, I am getting this error when building a deployment..

Hi 👋 I am trying to use the GitHub repo feature that was added in 2.3.0 I have this code for creating the deployment (🔥*which doesn’t work*🔥):

from weekly_bonus_calc import weekly_bonus_calc
from prefect.deployments import Deployment
from prefect.filesystems import GitHub

storage = GitHub(
    repository="<https://github.com/yuvital/yuvital-prefect-github.git>"
)

deployment = Deployment.build_from_flow(
    flow=weekly_bonus_calc,
    name="weekly-bonus-calc",    
    work_queue_name="work-queue",
    storage=storage,
    schedule={'cron': '*/5 * * * *', 'timezone': "Asia/Jerusalem"}
)

deployment.apply()

Also, I’ve created a GitHub block via the UI (attaching a screenshot). Can you point out the correct way to setup this. Another thing I don’t understand is where are the Git credentials should go since this is obviously a private repo. Thank you for any help on the issue 🙏

hello everyone 👋, I am using prefect server 1.0 deployed on k8 with git as storage. I want to try event based triggering of prefect flows by a third party api , Is it possible in prefect server, if yes then how ?

hello everyone, I am using create_flow_run() task in a parent flow to trigger a flow.So when this flow failed in halfway.So my parent flow should restart it instead creating a new flow all together. but how to achieve the former case?

def random_fn():
   print("Random")

@flow
def main_flow():
        prefect_flow = Flow(name="random_flow", fn=random_fn)
        deployment = Deployment.build_from_flow(
            flow=prefect_flow,
            name=random_deploy,
            version=1,
            work_queue_name="test-queue",
        )
        deployment.apply()

I'm trying to build a deployment for a flow that is created dynamically. While the deployment 'random_flow' gets created successfully, when I try to run the same, I get this error.

Flow could not be retrieved from deployment.
Traceback (most recent call last):
  File "/home/malavika/.virtualenvs/lib/python3.8/site-packages/prefect/engine.py", line 256, in retrieve_flow_then_begin_flow_run
    flow = await load_flow_from_flow_run(flow_run, client=client)
  File "/home/malavika/.virtualenvs/lib/python3.8/site-packages/prefect/client.py", line 103, in with_injected_client
    return await fn(*args, **kwargs)
  File "/home/malavika/.virtualenvs/lib/python3.8/site-packages/prefect/deployments.py", line 54, in load_flow_from_flow_run
    await storage_block.get_directory(from_path=deployment.path, local_path=".")
  File "/home/malavika/.virtualenvs/lib/python3.8/site-packages/prefect/filesystems.py", line 134, in get_directory
    shutil.copytree(from_path, local_path, dirs_exist_ok=True)
  File "/usr/lib/python3.8/shutil.py", line 555, in copytree
    with os.scandir(src) as itr:
FileNotFoundError: [Errno 2] No such file or directory: '/tmp/tmpbkxm8cdtprefect'

What can be done to link this flow to the deployment properly and get it to run through the Orion UI?

Anyone in the community that is using Prefect for CI/CD? And if so, how do you like it so far?

Hey guys, I might be missing something in v2, but in v1 I belive we can re-run just the failed tasks from a flow run, can we do the same in v2?

What are other ways to restart a flow apart from restarting in ui?

How can you stream logs to the UI from functions called in a Task that are using their own logger outside prefect? I can see those logs in the terminal, but never in the UI. I only can see logs that are defined with the

get_run_logger

Hi @Anna Geller and everyone, I wonder if there is API endpoint to check if the Agent is busy executing flows or in Idle mode? Thank you.

This message was deleted.

Hi. Should I install s3fs in addition to prefect to run the flow, because I see an error

ModuleNotFoundError: No module named 's3fs'

?

I receive an error that agent can't create a queue even if it exists, I use service account API key with role member. Should it be role with more permissions?

Prefect 2.3.1 - CI/CD: I run a freshly deployed flow at the end of my CI/CD pipeline, and would like to know whether or not the run was successful. My current idea is to poll the API for the state of the submitted run until it reaches a terminal state. Has anyone implemented a solution like this, or do you have another idea on how to solve this? 🙏

Hi. Does Prefect 2 support extras like previous version?(prefect[aws,kubernetes])

Are there any details anyone needs me to fill in? Is this the right place to be posting this? Don’t know how we like to “bump” messages here…

Hi @Chris Reuter and everyone. Is there someone that can help with answering this issues that I posted: https://github.com/anna-geller/prefect-docker-deployment/issues/1?

Hi prefect experts, I'm two days new since discovering prefect, I have a question about deployments. So If I understand correctly, unless I register a flow as a deployment, I can't trigger the flow in the orion UI Is that right? I saw docs that taught how to register a deployment (CLI or Python), but it looks like either method I can run multiple times and it will register the flow as deployments multiple times. Is there a way to create a file that simply "define" the list of deployments like flows does right now? Also how do I remove a deployment? is it through the orion UI only? Also, there seems to be no way to trigger a part of the flow (e.g. if the flow runs subflows 1,2,3,4,5 sequentially, there's no way to just run 3,4,5). Is that correct? Thanks!

Hey dear Prefectionists, I am running into weird errors with my K8s cluster workloads. The flows actually keep crashing with a 502 Bad Gateway error that mentions a Prefect Cloud URI:

prefect.exceptions.PrefectHTTPStatusError: Server error '502 Bad Gateway' for url '<https://api.prefect.cloud/api/accounts/><ACCOUNT>/workspaces/<WORKSPACE>/logs/'
For more information check: <https://httpstatuses.com/502>
Worker information:
    Approximate queue length: 0
    Pending log batch length: 1
    Pending log batch size: 360
The log worker will attempt to send these logs again in 2.0s

Is it possible that there are too many log messages sent and the Prefect Cloud API can not respond in time? Cheers!

Hello, Can't multiple users use same prefect db? I have setup postgress db and connecting with dbuser and password. User who start server first time it works for him however next user who try to start server connecting to same db there is error alembic.until.exc.CommandErro:Can't locate revision identified by 'd3....'

Hi! can a Prefect project be moved between teams? I've created a project in a personal account and recently joined a company account (with other members) but can't find info anywhere about how to make this movement (currently using Prefect 1.0)

Hello. In Prefect 1, I had a few flows stuck in Submitted for more than 1 hour, but Lazarus never did anything (although it's enabled for these flows). Is there a way to change Lazarus settings, or make sure it's working? (In the UI it's only enable/disable) I couldn't find any info in the documentation. Thank you

Do prefect 2's blocks have any sort of locking functionality? (imagine a queue-like block with some flows adding to it and others removing, how could they maintain reliability that no item gets lost?)