Hello, I'm trying to deploy prefect agents on an on prem environment, but I get a

prefect.exceptions.PrefectHTTPStatusError: Client error '403 Forbidden' for url '<https://api.prefect.cloud/api/accounts/xxxxxxxx/workspaces/xxxxx/work_queues/name/dev-prem-wq>'

I tried changing the api keys (service account) / giving ownership to the agent on the workspace, but nothing worked so far. I tried deploying the agents on the cloud on a kubernetes cluster and it works just fine, but when I try to do the same thing on our on prem environment, it returns me this error. Can I have some insights please ? Thanks

Hi guys ! I’m trying to setup a

Prefect Agent

and a

prefect server

on my local computer, everything residing inside of

containers

. My

Prefect Server

`container(s)`is based on the

docker-compose

file retrieved from

prefect server config > server-docker-compose.yaml

. When spinning up a container for an Agent and running

prefect agent local start --api <http://host.docker.internal:4200>

, it says that I must be authenticated to communicate with the containerized server living in my local machine. My first question is: Is

authentication

mandatory in this kind of situation , for local run ? When I authenticate by creating a new tenant using

prefect server create-tenant -n default

outside the Agent container, I’m able to spin up the

Agent Container

and run everything. The problem is that I wanted to run the

create-tenant

inside the container too, but it’s not possible as I’m getting connection refused between containers.

Hey guys, A question about sub-flows concurrency basic control issue: I have the need to create a flow to run sub-flow that are a bit heavy (on memory mostly, take ~10 min to run), and i need it with a few dozens different-inputs. i'm using prefect 2.20 I tried two approaches: - simply trigger each sub-flow (through

run_deployment

in order to allocate separate resources). this creates dozens sub-flows but the parent-job trying to keep updated and manage it crashes due to connection-limit reached error (an internal

httpx

crash). I asked marvin about configuring it on prefect and didn't see any immediate solution, perhaps more than 4-5 running in parallel is too much for prefect? am i missing something? - create many sub-flows but run it through a managed concurrency-limit queue (say, 3 at each given time - once one is done another kicks in). since it's a flow and not a task i had to go through work-pools /work-queue limits configuration. the thing is that since it takes some time to get the actual machine (mostly for the first batch from cluster), during that time it seems there's an error for the sub-flow yet to be executed (-1 signal until later it succeeds). and due to that temp-failure, other sub-flows also start to run and i'm missing my 3-limit (so 6-7 are running in parallel) which is odd as it seems like a common glitch. I looked for a parameter that controls how much time to wait for the first sub-flow-submit failure to avoid this race-edge-case but the only parameter (

job_watch_timeout_seconds

) that the docs were promising about, was also described as: "Number of seconds to wait for each event emitted by a job before timing out" - so raising this to 1-2 minutes seems too important/risky. - The third fallback option i use is creating less but chubbier jobs - each running serially 10 different inputs - and running them in parallel. but that seems like a bad solution for such a platform that should take care of that.. Did i miss any simple way out of this? Thanks!

@Marvin how can I trigger a flow run from receiving a SQS message?

@Marvin how can I skip a task if the task passed to the

wait_for

arg has failed?

@Marvin I'm suddenly getting 401 unauthorized errors without making any changes to my prefect setup. I've tried creating a new api key, verified my environment variables, etc. and still have no luck. can you help me debug this? is it possible I hit some free tier limit?

Nesting multiple asynchronous flows in Prefect can be managed using the concept of subflows or nested flows. Here's how it works, based on both Prefect 2.x and 3.x: ### Prefect 2.x In Prefect 2.x, you can create subflows by calling a flow function inside another flow. The parent flow is the main flow, while the child or "subflow" is executed within it. Subflows behave like regular flow runs and have their own task runners. Importantly, subflows block the parent flow's execution until completion, but with asynchronous subflows, you can run them concurrently using

asyncio.gather

. For more on this, you can check the Prefect 2.x documentation on nested flows. ### Prefect 3.x In Prefect 3.x, nested flows offer several benefits, such as using different task runners and enabling conditional execution of grouped tasks. They provide first-class observability, meaning you can track the status of nested flows easily within the Prefect UI. Nested flows also support asynchronous execution, allowing them to be awaited or run concurrently, adhering to async Python standards. For a deeper dive into creating asynchronous nested flows in Prefect 3.x, review the Prefect 3.x documentation on writing flows. In essence, whether you're using Prefect 2.x or 3.x, the key to nesting asynchronous flows lies in leveraging subflows with async capabilities and managing their execution so they can handle concurrent workloads effectively. Ah, the joys of managing asynchronous tasks—always making you feel like a juggler who needs just one more hand!

@Marvin We are using (currently) prefect 2.18 and now when running our tests, we get an import error

ModuleNotFoundError: No module named 'griffe.dataclasses'

. Does griffe not get installed with prefect 2.18?

I'm getting a QueuePool limit of size 5 overflow 10 reached when calling .map with around 700 tasks. How do I handle this and prevent the overflow/timeout from occuring?

# create an api request for each self link
api_results = api_get.map(map(lambda x: x["self"], employee_self_links))

@task(
    retries=3,
    retry_delay_seconds=exponential_backoff(backoff_factor=10),
)
def write(employee: Response):
    employee = employee.json()

    dest = f"raw/employees/details/{employee['id']}.json"
    write_to_file_system.fn(dest, employee)

# save the results to storage
storage_results = write.map(api_results)

Hello, Is there any way to implement a dropdown selection on the UI when starting a custom run for a deployment? Update: Solved!