Hello everyone, there's a way to configure Prefect so he can use a proxy server? I'm trying to log in on Prefect cloud (using the CLI), but I'm receiving the following SSLError:

requests.exceptions.SSLError: HTTPSConnectionPool(host='<http://api.prefect.io|api.prefect.io>', port=443): Max retries exceeded with url: / (Caused by SSLError(SSLEOFError(8, 'EOF occurred in violation of protocol (_ssl.c:852)'),))

Hey guys, I'm kinda struggling to set up a Prefect on our company server. So it's running and all that, but I had to setup a reverse proxy (Nginx) as well, so that people can access the UI via standard https at 443 and at a company subdomain (internal that is). The UI itself is accessible, but it's trying to make REST API requests to http://127.0.0.1/api/something which is obviously not working. Is there any env setting I can use to make it understand it should use https://prefect.internal.company.com/api instead? I couldn't find anything on that topic anywhere 😞 Thanks!

How should web applications (Node.js) listen/subscribe to Prefect workflows completion? I have a task that trains a deep learning model through a Web Application. The “*Start training*” button on the web app sends a

POST /model

REST API call to the Node.js server which then in turn calls Prefect 2.0 using the REST API to start a new run (of an existing deployment). The workflow will train the model, run inference on some data and then persist the output to a database. This is obviously an asynchronous operation that may take a few minutes (or more) to complete. Assuming the workflow succeeded, I would like to notify the users (those who have Chrome opened to my web app) that something has happened, i.e. training completed. How should the Node.js be notified when the flow has finished (either successfully / failed)? Is there like a middleware / RabbitMQ / other message queue that the Node.js app can subscribe to, onto which Prefect publishes event? If not, does Prefect expose other broadcast events? And if not, should I poll periodically from my app and maintain state diff? Thanks!

@Marvin my flow run is stuck in "awaiting concurrency slot" state, but there is no other ongoing runs. Why?

@Marvin How do I run .deploy to a process type worker

Hi all, I'm overhauling our current Prefect 2.19.1 to Prefect 3.1.0. Because our codebase dynamically generates a lot of deployments we use Prefect Python and not Prefect YAML to manage everything. Surprisingly the Python code has really stayed mostly the same between versions, which is great. Unsurprisingly some of our side scripts we use to manage the Cloud are now broken. One in particular we use to clear backlogs of Late runs is giving me fits. This code used to work

import asyncio

from prefect.states import State
from prefect.client import get_client

MAX_RUNS_TO_DELETE = 5000
PREFECT_OFFSET_MAX = 200


async def remove_all_flows():
    client = get_client()
    for i in range(0, MAX_RUNS_TO_DELETE, PREFECT_OFFSET_MAX):
        flow_runs = await client.read_flow_runs(offset=i)
        for flow_run in flow_runs:
            if flow_run.state_name == "Late":
                flow_id = flow_run.id
                print("deleting", flow_id)
                await client.set_flow_run_state(flow_run_id=flow_id, state=State(type="CANCELLED"))


asyncio.run(remove_all_flows())

But now returns an error

pydantic.errors.PydanticUserError: `StateCreate` is not fully defined; you should define all referenced types, then call `StateCreate.model_rebuild()`.

No amount of defining State parameters (or passing directly a StateCreate object) fixes the issue. I'm increasingly convinced the old code is now a dead end. Could the Prefect team advise on what the idiomatic way to cancel a Late flow run would be under Prefect 3.1+?

Hello, I am new to prefect and investigating several use-cases to assess if Prefect would be a good fit in the toolbox to build a team of engineers dealing with data integration between our company's API and our customers systems. For some use-cases we have to periodically pull files from a remote SFTP server, do some manipulations with the file and some APIs then finally upload a sort of "response" file over the SFTP server. I couldn't find something related to SFTP in the built-in tasks list https://docs.prefect.io/api/latest/tasks/function.html What would be the best way to deal with SFTP file exchange with prefect ? Should I just use a custom function task and a python sftp library or is there a more suitable task ?

Hello friends! I am running into some issues using an ECSCluster with the DaskExecutor. A brief overview of my setup: using Prefect cloud tied in with AWS infrastructure (e.g. running prefect-agent as a service on an ECSCluster) A flow might look something like this:

file_list = ['./file_1.txt', './file_2.txt', './file_3.txt', './file_4.txt', './file_5.txt', './file_6.txt', './file_7.txt', ..., './file_100.txt']

@task
def test_task(input_file):
    result = some_function(input_file)
    return result

def ecs_cluster(n_workers=4):
    """Start a ecs cluster using the same image as the flow run"""
    return ECSCluster(
        n_workers=n_workers, image=prefect.context.image, region_name="us-east-1"
    )


with Flow(
    "test flow",
    storage=S3(bucket="storage"),
    run_config=ECSRun(task_definition=task_definition),
) as flow:
    test_task.map(input_file=file_list)

flow.executor = DaskExecutor(
    cluster_class=ecs_cluster, cluster_kwargs={"n_workers": 10}
)

RuntimeError: Timed out while attempting to connect to ephemeral Prefect API server.

Is this a known issue? We wrap controlflow runs via a retriable mechanism, so it eventually completed. But is this potentially something that is on our end?

@Marvin I get this error when trying to set my Prefect profile. I use Prefect 2.10.3:

Unknown setting 'PREFECT_SERVER_ALLOW_EPHEMERAL_MODE'. (type=value_error)

When I run

prefect config unset PREFECT_SERVER_ALLOW_EPHEMERAL_MODE

I get the same error