Hi, is there a way to trigger a flow run similar to a webhook logic without using prefect cloud? I have all of my infrastructure on Azure, and I wanted to trigger a flow run when a file is created on datalake.

@Marvin what are recommended specs for runing a local prefect server, with4-5 workpools wach with 2 workers

Morning! I have an issue with the latest prefect release (2.16.0), the version 2.15.0 works fine. I have this minimal workable code async code.

@flow
async def myflow(experiment_id: int):
    engine = create_async_engine(
        url=get_db_url(),
        connect_args={
            'statement_cache_size': 0,
            'prepared_statement_cache_size': 0,
        },
    )

    async with AsyncSession(engine) as session:
        await session.get(Experiment, experiment_id)

if __name__ == '__main__':

    loop = asyncio.run(myflow(189))

But if I use a task, I got the following error:

RuntimeError: Task <Task pending name='Task-27' coro=<AsyncSession.close() running at /Users/william/.pyenv/versions/myenv/lib/python3.11/site-packages/sqlalchemy/ext/asyncio/session.py:1016> cb=[shield.<locals>._inner_done_callback() at /Users/william/.pyenv/versions/3.11.5/lib/python3.11/asyncio/tasks.py:881]> got Future <Future pending cb=[Protocol._on_waiter_completed()]> attached to a different loop

Full stracktrace on thread. Should I open a github ticket?

Hello, consider I have a prefect server on EC2 (AWS), can I deploy a flow in a sagemaker processing job and see the flow in the EC2 server?

Hello, I'm very new here so apologies for the question if it sounds newbie-like - I'm currently reviewing some prefect 1.0 code with mind to refactor into prefect 2.0, and one of the functions for prefect that is currently being used is prefect.core.parameter.DateTimeParameter, which doesnt seem to be supported in prefect 2.0. Is that the correct interpretation? If so, what would be prefect 2.0 alternative for this? Thanks in advance

Hey, does anyone in the community did incremental etl with Prefect ? We are considering doing it and maybe there are mistakes that can be avoided ^^ Thanks !

Hello everyone, does someone knows how to disable the SSL Verification when using prefect? Is there's something like the "*--set ssl_verify True*"? I'm facing some problems with auth and running the agent due to a corporate network. The following error occurs when I use the "*prefect auth login --key*":

requests.exceptions.SSLError: HTTPSConnectionPool(host='<http://api.prefect.io|api.prefect.io>', port=443): Max retries exceeded with url: / (Caused by SSLError(SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1129)')))

Hey guys, running into this issue with prefect, and I'd appreciate some insight. I'm working on just the tutorial from the docs, and the page isn't loading

Hello, I am currently trying to use a DaskExecutor in an ECSRun to parallelize a flow. I’m following the documentation to create a temporary cluster with a specified worker image. My flow currently looks something like this:

def fargate_cluster(n_workers=4):
    """Start a fargate cluster using the same image as the flow run"""
    return FargateCluster(n_workers=n_workers, image=prefect.context.image)

class LeoFlow(PrefectFlow):

    def generate_flow(self):
        with Flow(name=self.name, storage=S3(bucket="raptormaps-prefect-flows")) as flow:
            ...
        flow.executor = DaskExecutor(
            cluster_class=fargate_cluster,
            cluster_kwargs={"n_workers": 4}
        )
        return flow

In the dockerfile for the image that I’m specifying in the ECSRun, I have included the following line to install dask-cloudprovider:

RUN pip install dask-cloudprovider[aws]

However, when I execute the flow, I am hitting the following error:

Unexpected error: AttributeError("module 'aiobotocore' has no attribute 'get_session'",)
Traceback (most recent call last):
  File "/usr/local/lib/python3.6/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "/usr/local/lib/python3.6/site-packages/prefect/engine/flow_runner.py", line 442, in get_flow_run_state
    with self.check_for_cancellation(), executor.start():
  File "/usr/local/lib/python3.6/contextlib.py", line 81, in __enter__
    return next(self.gen)
  File "/usr/local/lib/python3.6/site-packages/prefect/executors/dask.py", line 238, in start
    with self.cluster_class(**self.cluster_kwargs) as cluster:
  File "/rprefect/leo_flow.py", line 58, in fargate_cluster
    return FargateCluster(n_workers=n_workers, image=prefect.context.image)
  File "/usr/local/lib/python3.6/site-packages/dask_cloudprovider/aws/ecs.py", line 1361, in __init__
    super().__init__(fargate_scheduler=True, fargate_workers=True, **kwargs)
  File "/usr/local/lib/python3.6/site-packages/dask_cloudprovider/aws/ecs.py", line 726, in __init__
    self.session = aiobotocore.get_session()
AttributeError: module 'aiobotocore' has no attribute 'get_session'

Is there a specific version of dask_cloudprovider that Prefect requires?

Hi, a newbie question… If the trigger for all tasks in a flow is “all successful”, and one of the upstream task fails, all of the downstream tasks are “trigger failed” (even if the upstream failed task is restarted and succeeds). This forces me to restart the trigger-failed tasks one by one. Is this the intended behavior? or am I missing something? Should I change the trigger to “all finished” in order to avoid this manual restarting of the downstream tasks when one of the upstream tasks fails at least once?