Hello Everyone, Currently running Prefect 2.0 on AKS. We’ve been working on deploying Flows via Azure DevOps and have successfully deployed them to a default workqueue. However, when testing a flow run, the agent (which is started for the default workqueue) doesn’t pick up the flow run. Does anyone have any ideas on which the issue might be here?

Hello community, hello Prefect team, one of our prefect 2 flows contains a large number of parallel tasks that access a database with limited capacity. in order to avoid too much load on the db, i tried to apply tag based task concurrency limits, observed some unexpected behavior and would love to get some feedback on if i am using concurrency limits correctly. 1. when submitting a batch of tasks in a loop, the specified concurrency limits do not seem to apply immediately. there are more running tasks than active slots for the tag. only if i introduce an artificial delay between submit calls, the limit takes effect. i assume this is due to some race conditions when accessing the orion db and was wondering if there is a preferred way of submitting tasks so that limits are respected. 2. when increasing the task concurrency limit while a flow is executing (this would be very useful when having long running flows and a dynamically scaleable db), the number of "Active Tasks Runs" doesn't return to zero after all flows are completed. instead a certain number of tasks is still listed as active by the

prefect concurrency-limit inspect

command but is in fact in status completed. this effectively lowers the concurrency-limit for all subsequent tasks as these pseudo-active tasks keep occupying a slot for the tag. are we at all supposed to adapt the concurrency limit while tagged tasks are being executed or is this considered bad practice? Please find in the thread a self-contained example flow that can be used to reproduce the behavior. Any advice i much appreciated. Thank you.

Hi - I am trying to run a Prefect 2.6.1 KubernetesJob however the job when created keeps failing with - any ideas? I have tried changing path & entrypoint settings numerous times but am now stuck

/usr/local/bin/python: No module named prefect.engine.__main__; 'prefect.engine' is a package and cannot be directly executed

Just saw the release notes gratitude thank you 🎉 to the addition of results!

Hey everyone! I'm so close to deploying my first flow, with

ECSTask

infrastructure. I'm programmatically creating a

Deployment

for a simple flow and it seems to have registered on our Prefect Cloud. However, the deployment run fails without log messages in Prefect Cloud. The agent logs

botocore.exceptions.ParamValidationError

, which makes me think this might be out of my reach. I'm currently in what seems to be the exact same situation as @Anthony Desmier (link to thread) is in. Not really sure where to go next. Any help is appreciated a lot, thank you for your time!

Seems like prefect's

prefect deployment apply

command has a typo in its output:

Successfully loaded 'sampleflow_dev'
Successfully uploaded 4 files to <gcs://dev-prefect-work/flow_storage/>
Deployment 'sampleflow/sampleflow_dev' successfully created with id 'c3b605e4-ca96-45eb-b51a-b048ca6d2e08'.
View Deployment in UI: <https://app.prefect.cloud/account/a9c0f124-ca06-4646-a501-57a405ebf3c7/workspace/43f1b8a7-ed9c-46d2-a88a-55eef95b8ef7/deployment/c3b605e4-ca96-45eb-b51a-b048ca6d2e08>

To execute flow runs from this deployment, start an agent that pulls work from the 'kubernetes' work queue:
$ prefect agent start -q 'kubernetes'

...that link it shows is incorrect: https://app.prefect.cloud/account/a9c0f124-ca06-4646-a501-57a405ebf3c7/workspace/43f1[…]55eef95b8ef7/deployment/c3b605e4-ca96-45eb-b51a-b048ca6d2e08 ...it should be: https://app.prefect.cloud/account/a9c0f124-ca06-4646-a501-57a405ebf3c7/workspace/43f1[…]/deployments/deployment/c3b605e4-ca96-45eb-b51a-b048ca6d2e08 ...in other words, it's missing a "/deployments"

How should we think about job resiliency during agent stoppage and restart? When an agent stops or a flow run executor stops, how can we ensure the flow run is properly rerun, and NOT stuck in the Running state (with no execution)? I'm using local execution block, but this Github issue suggests remote runs on Kubernetes (and maybe ECS) are also a concern, for 'missing' jobs that don't get reruns. https://prefect-community.slack.com/archives/C03D12VV4NN/p1664554177863929?thread_ts=1664525816.785439&cid=C03D12VV4NN

When running using async, there are a number of pythonic ways to approach concurrency limits including queues and semaphores. The examples I have seen with Prefect 2 (and I may not be looking hard enough) are associated with process concurrency and not async concurrency. Is there a recommended way of limiting async concurrency? Are we free to use

asyncio.Semaphore

or

asyncio.Queue

and the like? Thanks for any suggestions or pointers.

Good afternoon, I managed to fix my problem

I’m trying to contribute to Prefect… I downloaded the repository and ran the tests (without modifications). This is the summary of the output I received…

10 failed, 4486 passed, 39 skipped, 3 xfailed, 11 errors in 1173.85s (0:19:33)

I wasn’t expecting to have failures… I’ve added my code and am now running the 20 minutes of tests to ensure it’s working as expected… I’m assuming so long as I don’t get to

11 failed

in the summary I should be okay right?

Hey all, I'm in the process of evaluating whether prefect 2.0 is a viable solution to replace an internal tool at work. I've gone through setting up an ECS cluster and have prefect agents listening and able to spin up a dask scheduler and dask cuda worker. However, I'm running into some prefect issues, referenced by someone else here: https://github.com/PrefectHQ/prefect/issues/6759 Does anyone have a workaround for this issue or an ETA on when this will be fixed? I've already tried setting an environment variable in the docker image (PREFECT_API_REQUEST_TIMEOUT=60), but to no avail. Exact error I'm seeing in the dask-cuda-worker in the 🧵

Hi The Prefect schedule does not seem to work. Here is the code and command I ran. flows/tutorial.py

from prefect import flow,get_run_logger

@flow
def my_favorite_function():

    logger = get_run_logger()
    <http://logger.info|logger.info>("aaaaaaaaaaaaaaaaaaaaaaaaaa")
    print("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
    return 42

prefect deployment build flows/tutorial.py:my_favorite_function -n etl --cron "*/1 * * * *" -q etl --apply
prefect agent start -q 'etl

Is there a problem?

Hey, I’m in the process of switching to Prefect 2.x. I’m trying to figure out how to map over multiple arguments and generate child subflows. I realize that

Task.map

can do something similar to generate tasks, but I’d prefer subflows for my project. Is this possible? If so, any docs or resources y’all can point me to? Thanks!

"Can prefect execute a flow in a single thread? See comments for more details and context, I’m trying to get Prefect to run Playwright, but it is single-threaded only." Hi Can Playwright be combined with Prefect?

[tool.poetry.dependencies]
python = "^3.10"
prefect = "^2.6.1"
playwright = "^1.27.1"

import prefect
from playwright.sync_api import Page, sync_playwright

@prefect.task
def go_to_a_web(page: Page):
    page.goto("<https://formulae.brew.sh/cask/slack>")


@prefect.flow
def main_flow():
    with sync_playwright() as pw:
        browser = pw.chromium.launch(headless=False)
        page = browser.new_page()
        go_to_a_web(page)


main_flow()

I got

greenlet.error: cannot switch to a different thread

any help would be much appreciated

Hi all, Im trying to post a message in slack after the flow gets to success state. But it shows None as the message for some reason. I have used the same function for failure and it works fine on it. I have given the flow code below First flow function try:

def post_to_slack_on_success(task, old_state, new_state):
    if new_state.is_successful():
        flow_run = FlowRunView.from_flow_run_id(prefect.context.get("flow_run_id"))
        task_run = flow_run.get_task_run(task_run_id=prefect.context.get("task_run_id"))
        msg = task_run.get_result()
    SlackTask(message=msg).run()
    flow.state_handlers.append(slack)

Second flow function try:

def post_to_slack_on_success(task, old_state, new_state):
    if new_state.is_successful():
        print(new_state)
        print(old_state)
        print("Result  :  ", new_state.result, new_state.message)

Screenshot of the alert:

Hello, Trying to run multiple python scripts in prefect. My use case is that, each task should represent as an individual python script and these all scripts/ tasks can be combined to form a single flow . HOW IS THIS ACHIEVABLE? Can we run bash commands as a single task in prefect version 2. #prefect-contributors #prefect-community #prefect-dbt

Hey, I've a question about Results storage. (prefect 2.6) Currently, I do not configure any results storage - neither on flows nor tasks. According to the docs, the results then are stored in the PREFECT_LOCAL_STORAGE_PATH (~/.prefect/storage). This is also what I see - a lot of blobs in this folder. Does prefect sometimes delete these blobs/results? Or am I responsible for cleaning up the result storage location?

Hi, I'm trying to run a flow on prefect 1.0. Every task ends up with a success but at the end flow fails. When I check the logs there is an error

prefect.exceptions.ClientError: [{'path': ['get_or_create_task_run_info'], 'message': 'Expected type UUID!, found ""; Could not parse UUID: ', 'extensions': {'code': 'INTERNAL_SERVER_ERROR', 'exception': {'message': 'Expected type UUID!, found ""; Could not parse UUID: ', 'locations': [{'line': 2, 'column': 101}], 'path': None}}}]

I have tried to re-register flow and I'm sure that flow run and registered flow are the same

This message was deleted.

Hi is there issue how prefect 1.4 .0 handles newest dask 2022.10.0 and python3.9 becouse ci/cd pipeline does not work now when I am using those but local machine with python 3.9.5 works fine.

Hi, I'm having some issues with accessing prefect 2.0 cloud from behind a proxy. I can login and switch workspaces but

prefect block ls

times out. If I switch to a network without proxy everything works. Any suggestions?

hi everyone, question about Prefect 2.6.0, what is analog 'with case' in Prefect 2 or there is not anything similar? how to define a branch logic based on some check?

Rather interesting task failure. I have the following resource manager (some functions removed to keep the example simple):

@resource_manager
class LocalFileManager:

    ...

    def setup(self) -> 'LocalFileManager':
        self.temp_dir = Path(mkdtemp())
        self.logger = prefect.context['logger']
        <http://self.logger.info|self.logger.info>('Set up LocalFileManager @ %s', self.temp_dir)
        self._lock = RLock()
        self._cache = {}
        return self

    def cleanup(self, _) -> None:
        with self._lock:
            <http://self.logger.info|self.logger.info>('Cleaning up: %s', self.temp_dir)
            shutil.rmtree(self.temp_dir)

I had a flow fail, the setup task is the root failure. Here are the logs

17 October 2022,06:09:30 	prefect.CloudTaskRunner	INFO	Task 'LocalFileManager.setup': Starting task run...
17 October 2022,06:09:30 	prefect.LocalFileManager.setup	INFO	Set up LocalFileManager @ /tmp/tmp1b3_t4xs
17 October 2022,06:09:31 	prefect.CloudTaskRunner	INFO	Task 'LocalFileManager.setup': Finished task run for task with final state: 'Success'
17 October 2022,08:55:01 	prefect.CloudTaskRunner	INFO	Task 'LocalFileManager.setup': Starting task run...
17 October 2022,08:55:01 	prefect.CloudTaskRunner	ERROR	Task 'LocalFileManager.setup': Exception encountered during task execution!
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/prefect/engine/task_runner.py", line 884, in get_task_run_state
    logger=self.logger,
  File "/usr/local/lib/python3.7/site-packages/prefect/utilities/executors.py", line 468, in run_task_with_timeout
    return task.run(*args, **kwargs)  # type: ignore
  File "/usr/local/lib/python3.7/site-packages/prefect/tasks/core/resource_manager.py", line 25, in run
    return mgr.setup()
AttributeError: 'NoneType' object has no attribute 'setup'
17 October 2022,08:55:01 	prefect.CloudTaskRunner	INFO	Task 'LocalFileManager.setup': Finished task run for task with final state: 'Failed'

It seems like the task ran twice and fails for the second run. This exact flow has run multiple times without issue, and this is a common resource manager used in multiple other flows.

Hello. In Prefect 1 GraphQL, I want to query flow runs with a certain flow run parameter, but when I try to do

parameters{...}

I get this error

Field parameters must not have a selection since type "jsonb" has no subfields.

Any ideas on how I can access the subfields in parameters and use them in a where condition? Thank you

Does anyone know where I can always see the most recent version of the helm chart prefect-agent https://github.com/PrefectHQ/prefect-helm/tree/main/charts/prefect-agent It shows version 0.0.0 on this page but I don't believe that's right.

Hi, In Prefect 1, is it possible to somehow export the whole log file for the task as opposed to only what's shown on the screen?

Hello, Is it possible to launch Orion within python?

#prefect-2-orion

Hi all, question. Anything we can do in a

@task

can also be done in a

@flow

, but tasks have a limitation: they can’t call other tasks. In a complex/nested job, for every function except the top-level flow I have to think about whether it is atomic and can be a task, or whether I will want to split it out further into multiple tasks, in which case it needs to be a flow. But I realized I could just make everything a flow so I don’t have to worry about it. If I decide to split it out I can. If I don’t, no problem. Given the power and flexibility of flows, and that one arbitrary limitation of tasks, why use tasks at all? So, what am I missing here? Are tasks obsolete? What’s their unique use case?

Prefect 2.6 has brought me back... but I am getting the occasional error like this. This looks to be a failure with the Prefect API? Can anyone shed some light on it for me? Thanks

11:20:34.559 | ERROR   | Task run 'Get-Items-d8ed86f1-2199' - Crash detected! Request to <https://api.prefect.cloud/api/accounts/cafe3a79-624b-468d-87a7-97fde3358a01/workspaces/5d09d677-90b8-4ef8-a9be-9760b422937a/task_runs/a13a62a5-5e54-4b1d-a976-98635a6913c6> failed: Traceback (most recent call last):
  File "/home/tenders/.cache/pypoetry/virtualenvs/prefect-orion-HonJDUqB-py3.10/lib/python3.10/site-packages/anyio/streams/tls.py", line 130, in _call_sslobject_method
    result = func(*args)
  File "/usr/lib/python3.10/ssl.py", line 975, in do_handshake
    self._sslobj.do_handshake()
ssl.SSLWantReadError: The operation did not complete (read) (_ssl.c:997)

Sorry, I'm still confused around blocks! I understand that I might want to add a block in order to centrally coordinate access to configuration inside my flow. At this point, I'm just trying to get my flow running though, using an ECS-Task block. 1. I've defined a block in Python with

ECSTask(...).save("dev-trial", overwrite=True)

on my dev machine. It shows up in Prefect Cloud. 2. I have created a deployment with

prefect deployment build -n dev-trial -q dev -ib ecs-task/dev-trial -a flows/healthcheck.py:healcheck

again from my dev machine 3. I have started a local agent with

prefect agent start -q dev

and triggered a job. All works. 4. I have started an agent inside a container and triggered the job, but now it fails with the error

KeyError: "No class found for dispatch key 'ecs-task' in registry for type 'Block'."

It seems like in the container, it's not able to resolve the block reference. But isn't that embedded inside the yaml downloaded from Prefect Cloud?