In writing this message I think I got there myself, but since I typed it (and prototyped it in code) I figure I will ask anyway. A question for the Prefect devs and other users. What is the rational on a

@task

decorated function blocking when calling it as is, e.g.

@task 
def square_num(a):
    return a**2

@flow 
def main_flow():
    squares = [square_num(i) for i in range(4)]

here each time

square_num

is called in the loop, it is not a

PrefectFuture

that is stored, it is the evaluated number. To get the behaviour that is similar to the

@dask.delayed

type decorated function, the

.submit

method should be used. I ask this question after mocking up a port of a

dask

workflow over to

prefect

. If the

__call__

of a

Task

instance behaved similar to its

submit

method, the conversion would basically be as simple as replacing

@delayed

with

@task

and calling it a day, I mocked up a extra

Task

keyword argument in my own

prefect

fork called

block_on_call

, and could tweak this

__call__

vs

.submit()

behaviour on a per

@task

basis. Maybe it is because of my previous

dask

exposure, but this kind of feels more natural to me, and kind of makes this toy porting problem pretty effortless. Is it a case that the current

Task.__call__

behaviour is to support a DAG-less type use case/style? Would there be much interest in a pull request of the above make sense/be useful, or am I missing the larger intent here? Does my rambling make sense?

Hi Team, Can we start prefect agent with multiple pools? I tried below but it doesn't work, it only picks up the last -p pool mentioned

prefect agent start -p my-pool -p my-pool-2 -p my-pool-3

Hello, does anyone know if it is possible to restart a task run and not entire flow in Prefect 2 cloud using Prefect client or API callbacks?

Hi, i want replace local DaskTaskRunner to dask remote cluster, i have a simple code

import asyncio
from typing import List

from prefect import flow, task
from prefect_dask import DaskTaskRunner


@task
async def test_task(numbers: List[int]):
    print(numbers)


@flow(task_runner=DaskTaskRunner('<tcp://10.10.10.200:8786>'))
async def my_parent():
    numbers = [i for i in range(2)]
    coros = test_task.map(numbers)
    await asyncio.gather(*[coros])


if __name__ == "__main__":
    asyncio.run(my_parent())

Does somebody tell me what i'm doing wrong ? Without mapping test_task, it works, mapping + local dask works too. but if i want use remote dask cluster and mapping i've receive this error:

15:19:09.646 | INFO    | distributed.batched - Batched Comm Closed <TCP (closed) Client->Scheduler local=<tcp://10.61.14.2:60674> remote=<tcp://10.10.10.200:8786>>
Traceback (most recent call last):
  File "/home/robin/.cache/pypoetry/virtualenvs/algoseek-adjustments-inserter-myRRPE6N-py3.11/lib/python3.11/site-packages/distributed/batched.py", line 115, in _background_send
    nbytes = yield coro
             ^^^^^^^^^^
  File "/home/robin/.cache/pypoetry/virtualenvs/algoseek-adjustments-inserter-myRRPE6N-py3.11/lib/python3.11/site-packages/tornado/gen.py", line 767, in run
    value = future.result()
            ^^^^^^^^^^^^^^^
  File "/home/robin/.cache/pypoetry/virtualenvs/algoseek-adjustments-inserter-myRRPE6N-py3.11/lib/python3.11/site-packages/distributed/comm/tcp.py", line 269, in write
    raise CommClosedError()
distributed.comm.core.CommClosedError
15:19:09.659 | INFO    | Task run 'test_task-0' - Crash detected! Execution was interrupted by an unexpected exception: CancelledError: test_task-0-4fd7fc89242e4471877c70e0b40362e9-1
15:19:09.688 | INFO    | Task run 'test_task-1' - Crash detected! Execution was interrupted by an unexpected exception: CancelledError: test_task-1-1249f0856b0a4a138e0ca263f104cdea-1
15:19:09.744 | ERROR   | Flow run 'brass-worm' - Finished in state Failed('2/2 states failed.')
Traceback (most recent call last):
  File "/home/robin/.config/JetBrains/PyCharmCE2023.1/scratches/tmp_dask.py", line 21, in <module>
    asyncio.run(my_parent())
  File "/usr/lib/python3.11/asyncio/runners.py", line 190, in run
    return runner.run(main)
           ^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.11/asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.11/asyncio/base_events.py", line 653, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "/home/robin/.cache/pypoetry/virtualenvs/algoseek-adjustments-inserter-myRRPE6N-py3.11/lib/python3.11/site-packages/prefect/_internal/concurrency/api.py", line 182, in wait_for_call_in_loop_thread
    return call.result()
           ^^^^^^^^^^^^^
  File "/home/robin/.cache/pypoetry/virtualenvs/algoseek-adjustments-inserter-myRRPE6N-py3.11/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 283, in result
    return self.future.result(timeout=timeout)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/robin/.cache/pypoetry/virtualenvs/algoseek-adjustments-inserter-myRRPE6N-py3.11/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 169, in result
    return self.__get_result()
           ^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.11/concurrent/futures/_base.py", line 401, in __get_result
    raise self._exception
  File "/home/robin/.cache/pypoetry/virtualenvs/algoseek-adjustments-inserter-myRRPE6N-py3.11/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 346, in _run_async
    result = await coro
             ^^^^^^^^^^
  File "/home/robin/.cache/pypoetry/virtualenvs/algoseek-adjustments-inserter-myRRPE6N-py3.11/lib/python3.11/site-packages/prefect/client/utilities.py", line 51, in with_injected_client
    return await fn(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/robin/.cache/pypoetry/virtualenvs/algoseek-adjustments-inserter-myRRPE6N-py3.11/lib/python3.11/site-packages/prefect/engine.py", line 375, in create_then_begin_flow_run
    return await state.result(fetch=True)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/robin/.cache/pypoetry/virtualenvs/algoseek-adjustments-inserter-myRRPE6N-py3.11/lib/python3.11/site-packages/prefect/states.py", line 91, in _get_state_result
    raise await get_state_exception(state)
  File "/home/robin/.cache/pypoetry/virtualenvs/algoseek-adjustments-inserter-myRRPE6N-py3.11/lib/python3.11/site-packages/prefect_dask/task_runners.py", line 298, in wait
    return await future.result(timeout=timeout)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/robin/.cache/pypoetry/virtualenvs/algoseek-adjustments-inserter-myRRPE6N-py3.11/lib/python3.11/site-packages/distributed/client.py", line 305, in _result
    raise exception
concurrent.futures._base.CancelledError: test_task-0-4fd7fc89242e4471877c70e0b40362e9-1

Process finished with exit code 1

any tip will be appreciated

Hello, I have this task:

@task
def extract():
    logger = get_run_logger()

    result = sf_operator.set_stage(url='<S3://bucket/folder/working/>',
                                   stage_location='EXTERNAL_TABLES.BACKUP_TABLE')
    [<http://logger.info|logger.info>(log_message) for log_message in result]

    result = sf_operator.set_external_table(external_table='EXTERNAL_TABLES.BACKUP',
                                            stage_location='EXTERNAL_TABLES.BACKUP_TABLE')

    [<http://logger.info|logger.info>(log_message) for log_message in result]

Here in sf_operator.set_stage, I am using

from prefect_snowflake.database import SnowflakeConnector, snowflake_query

to run query to create a stage in snowflake. But I am getting

RuntimeError: Tasks cannot be run from within tasks. Did you mean to call this task in a flow?
prefect.flow_runs
Finished in state Failed('Flow run encountered an exception. RuntimeError: Tasks cannot be run from within tasks. Did you mean to call this task in a flow?')

Can anyone please explain where am I using task inside a task and a possible work around this? This worked fine in prefect 1 Any help will be appreciated.

Hello! We want to have possibility to continue the flow run in case K8S pod was killed and flows run restarted on the new pod. Is it possible to use some Prefect features to share that state and results from executed tasks to continue the flow run from where it was interrupted?

hello everyone, this may sound a bit dumb, but how can i get the logs from a subflow triggered from a flow that i directly run using the python API? i have been able to get the flow logs with a subprocess that calls 'prefect flow-run logs <id>' but i cannot go further.. is it possible to retrieve them directly from my python script? is it possible to track the subflow chain execution from the initial flow ID within the python api or is it unavoidable to check the UI for flow ids etc?? thanks!

Hi team! Im trying to figure out the easiest way to trigger a no-changes-to-the-defaults flow run when a cloudwatch trigger happens. Ie someone uploads a file into s3, and that triggers a prefect flow run which is designed to download that file, process it, and stick it somewhere else. Are there any existing patterns of plugins which would make this easy to do instead of our default solution (cloudwatch triggers a lambda). I notice you can trigger a flow run via the REST API, but the example request (https://docs.prefect.io/2.10.16/api-ref/rest-api-reference/#tag/Flow-Runs/operation/create_flow_run_flow_runs__post) is… massive.

Hi all - has anyone used a Github storage block with a repo that has a git submodule? Is there a convenient way to pull the git submodule at runtime?

How can I run an async task outside of a flow block? I'm trying to use

prefect_aws.secrets_manager.read_secret()

and am stuck with a coroutine object.

Probably an easy one here — I’m hosting a free plan for Prefect on a single EC2 utilizing the Prefect agent and am hoping to display a clickable link via the logs for a flow run in the Prefect UI. I’ve got the proper markdown syntax

[Link text Here](<https://link-url-here.org>)

, however the link is still not clickable. I believe I need to set

PREFECT_LOGGING_MARKUP=True

— any recommendation on the best place to configure this?

Is there a way to have Prefect flows run as a user other than

root

?

Hey, just double checking my understanding. You can't configure a Docker Container Infrastructure block (hosting locally) with podman, you need actual docker to make this work?

Hey team! The latest prefect release that I can see on anaconda is for 2.10.14, which is now a few releases behind what we're up to (2.10.17). Does anyone know what might be the cause here? https://anaconda.org/conda-forge/prefect/files

I'm having an issue where the prefect SDK is overwriting my logging configuration when the module is loaded. The codebase's logging configuration is partially determined at run time and loaded along with django, so simply specifying the

PREFECT_LOGGING_SETTINGS_PATH

env var doesn't seem sufficient. What's the best way to prevent prefect from making additional changes to the logging configuration when loaded?

Hello, I’m not understanding why use an ECS worker or an ECSTask as an infrastructure block. Can I use a local agent that creates ECS task definitions for each flow?

Folks may already know this but fyi if anyone else has been wanting to use Prefect params in Cloud and be able to run the CLI scripts the terminal, you can just add

typer.run

directly and it'll work for both. i.e.:

if __name__ == "__main__":
    typer.run(process_file_flow)

Hi there, how do I configure a Slack Webhook automation to include the last line of output from any failed tasks? In Prefect 1 we had slack notifications such as:

Failed

Propagate Affinity merges / deletions is now in a Failed state

Message

Copy code

AssertionError('Some items failed to merge')

Now the best I can get is:

<flow> in state

Failed

Flow run <flow>/fiery-hyrax observed in state

Failed

at 2023-06-23T190401.492402+00:00.

Flow run URL: https://app.prefect.cloud/.../

State message: 1/1 states failed.

from the config:

# Subject
{{ flow.name }} in state `{{ flow_run.state.name }}`

# Body
Flow run {{ flow.name }}/{{ flow_run.name }} observed in state `{{ flow_run.state.name }}` at {{ flow_run.state.timestamp }}.
Flow run URL: {{ flow_run|ui_url }}
State message: {{ flow_run.state.message }}

Is it possible to improve upon the "State message" part as this is not very informative? Or is this not available in the variables that can be accessed from this context. If that's the case how would we include this information if we built something to send a notification on failure ourselves?

Hi for the prefect api endpoint api/logs/filter, anyone know what the level corresponds to? e.g. CRITCAL, INFO, etc.

"level": {
      "ge_": 20,
      "le_": 50
    },

Oh I also guess that the logging is only for flows and tasks but not any server logs

Query. to tall -- in the same way that Dask has its visualize method to produce a graph representation of the workflow, is there a similar method one can use for prefect? This question might not make sense too much with the DAG-less nature of prefect, but curious. The only tool is great until there is a time-imbalance between tasks, e.g. A handful of tasks take 4+ hours to run, but others only a few minutes. It is hard to follow and make sense of.

hey, I am using prefect open-source on k8s with helm, I have one flow that run each day, after couples of days the ui stop working well and responding/ not showing data, also the pod cpu is looking very high and I dont understand why ( around 1.1 c), when I look on the pod server logs I see a lot of errors like this:

07:35:21.231 | ERROR   | prefect.server.services.cancellationcleanup - Unexpected error in: TimeoutError()
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/prefect/server/services/loop_service.py", line 78, in start
    await self.run_once()
  File "/usr/local/lib/python3.10/site-packages/prefect/server/database/dependencies.py", line 119, in async_wrapper
    return await fn(*args, **kwargs)
  File "/usr/local/lib/python3.10/site-packages/prefect/server/services/cancellation_cleanup.py", line 48, in run_once
    await self.clean_up_cancelled_subflow_runs(db)
  File "/usr/local/lib/python3.10/site-packages/prefect/server/services/cancellation_cleanup.py", line 96, in clean_up_cancelled_subflow_runs
    subflow_run_result = await session.execute(subflow_query)
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/ext/asyncio/session.py", line 436, in execute
    result = await greenlet_spawn(
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 190, in greenlet_spawn
    result = context.throw(*sys.exc_info())
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/orm/session.py", line 2225, in execute
    return self._execute_internal(
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/orm/session.py", line 2120, in _execute_internal
    result: Result[Any] = compile_state_cls.orm_execute_statement(
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/orm/context.py", line 293, in orm_execute_statement
    result = conn.execute(
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1412, in execute
    return meth(
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/sql/elements.py", line 483, in _execute_on_connection
    return connection._execute_clauseelement(
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1635, in _execute_clauseelement
    ret = self._execute_context(
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1844, in _execute_context
    return self._exec_single_context(
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1984, in _exec_single_context
    self._handle_dbapi_exception(    raise exc_info[1].with_traceback(exc_info[2])
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1965, in _exec_single_context
    self.dialect.do_execute(
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/engine/default.py", line 921, in do_execute
    cursor.execute(statement, parameters)
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 561, in execute
    self._adapt_connection.await_(
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 125, in await_only
    return current.driver.switch(awaitable)  # type: ignore[no-any-return]
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 185, in greenlet_spawn
    value = await result
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 540, in _prepare_and_execute
    self._handle_exception(error)
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 491, in _handle_exception
    self._adapt_connection._handle_exception(error)
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 780, in _handle_exception
    raise error
  File "/usr/local/lib/python3.10/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 528, in _prepare_and_execute
    self._rows = await prepared_stmt.fetch(*parameters)
  File "/usr/local/lib/python3.10/site-packages/asyncpg/prepared_stmt.py", line 176, in fetch
    data = await self.__bind_execute(args, 0, timeout)
  File "/usr/local/lib/python3.10/site-packages/asyncpg/prepared_stmt.py", line 241, in __bind_execute
    data, status, _ = await self.__do_execute(
  File "/usr/local/lib/python3.10/site-packages/asyncpg/prepared_stmt.py", line 230, in __do_execute
    return await executor(protocol)
  File "asyncpg/protocol/protocol.pyx", line 201, in bind_execute
asyncio.exceptions.TimeoutError

Hi everyone! After a successful PoC we are going to use Prefect in production: pretty excited on that! I have some production experience with Prefect 1.0 but I would like to go for prefect 2.0 now. Is there any way to get in touch with some of your Developer Advocates to help me in solving a few issues? Best 🙂

Hi everyone and Prefect community 👋, I’m trying to configure Persistent Results of my flows and tasks to be stored on a S3 Bucket. I have configured my flow with the following. For the PoC, I’m using the RemoteFileSystem from

prefect.filesystems

that use fsspec to point the configuration of my S3 bucket via the keyword argument

settings=

like below:

from prefect import task, flow
from prefect.context import get_run_context
from prefect.filesystems import RemoteFileSystem

bucket_endpoint = "..."
access_key = "..."
secret_key = "..."
bucket_name = "s3-prefect-results-bucket"
s3 = RemoteFileSystem(basepath=f"s3://{bucket_name}",
           settings={"client_kwargs": {"endpoint_url": bucket_endpoint}, "key": access_key, "secret": secret_key})

@task(persist_result=True)
def task_persisted(name):
  ctx = get_run_context()
  task_id = ctx.task_run.id
  ret = f"Task id persisted {task_id} by {name}"

  return ret

@flow(name="persist_flow", result_storage=s3)
def persist_flow(server_id: int):
  result = task_persisted("sbrabez")
  print(f"Result persisted: {result}")

  return 0

But when I executed it, it reports the following errors. i.e. It seems it cannot

PubObject

(S3 API) and the problem seems located in the

aiobotocore

client side. Any ideas what’s wrong here or what I’m doing wrong? My S3 Credentials are OK here.

OSError: [Errno 22] Invalid Argument.
 | flow_name:persist_flow | flow_run_name:important-dragonfly | flow_run_id:211feee3-4ab1-4b81-86ea-8e76856ed500 | level_name:ERROR | message:Finished in state Failed('Flow run encountered an exception. Traceback (most recent call last):
File "/opt/prefect/lib/python3.8/site-packages/s3fs/core.py", line 113, in _error_wrapper
    return await func(*args, **kwargs)
File "/opt/prefect/lib/python3.8/site-packages/aiobotocore/client.py", line 371, in _make_api_call
    raise error_class(parsed_response, operation_name)
botocore.exceptions.ClientError: An error occurred (InvalidArgument) when calling the PutObject operation: Invalid Argument.

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

OSError: [Errno 22] Invalid Argument.\n')
Traceback (most recent call last):
  File "/opt/prefect/lib/python3.8/site-packages/s3fs/core.py", line 113, in _error_wrapper
    return await func(*args, **kwargs)
  File "/opt/prefect/lib/python3.8/site-packages/aiobotocore/client.py", line 371, in _make_api_call
    raise error_class(parsed_response, operation_name)
botocore.exceptions.ClientError: An error occurred (InvalidArgument) when calling the PutObject operation: Invalid Argument.

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

Traceback (most recent call last):
  File "persist.py", line 32, in <module>
    persist_flow(server_id=19959)
  File "/opt/prefect/lib/python3.8/site-packages/prefect/flows.py", line 468, in __call__
    return enter_flow_run_engine_from_flow_call(
  File "/opt/prefect/lib/python3.8/site-packages/prefect/engine.py", line 182, in enter_flow_run_engine_from_flow_call
    retval = from_sync.wait_for_call_in_loop_thread(
  File "/opt/prefect/lib/python3.8/site-packages/prefect/_internal/concurrency/api.py", line 137, in wait_for_call_in_loop_thread
    return call.result()
  File "/opt/prefect/lib/python3.8/site-packages/prefect/_internal/concurrency/calls.py", line 173, in result
    return self.future.result(timeout=timeout)

FYI, I’m using the Prefect version

2.9.0

Version:             2.9.0
API version:         0.8.4
Python version:      3.8.10
Git commit:          69f57bd5
Built:               Thu, Mar 30, 2023 1:08 PM
OS/Arch:             linux/x86_64
Profile:             objectStorage
Server type:         server

I’ll appreciate if you can point me any example or resource to configure this persistent result feature via S3. Thanks for your assistance, Cheers! 🤝

Hi. How can I best limit the execution of a flow? In Prefect 1.0, for each flow, we would set the "Flow Run Concurrency" in the UI for each flow. In Prefect 2.0, I create a work queue for each flow and limit the "Concurrency Limit" to 1. However, I notice that multiple flows are being launched without waiting for a previously launched flow to finish its execution.

Hi, I have 2k shell commands to run - I divide them to batches of 8 commands and launch a flow(similar to `process_pokemon_batch`in this orchestrator recipe) for each--> 250 flows, each running 8 commands. Now I want each flow to run the 8 commands in parallel, each one using it's own CPU. It works great when I test it locally with

concurrent.futures.ProcessPoolExecutor

or

multiprocessing.Pool

but get stuck on the ECS agent. You could think of it as if the `get_pokemon_info`is a CPU bound command that takes 1 minute to run and therefore it makes sense to run these in parallel (not like the current

get_pokemon_info

, which is i/o bound and is a pretty light API call and therefore makes sense to run concurrently using async). So my question is how can I enable my own parallelism? (Great comment about Concurrency versus parallelism is in the Prefect docs)

I set up the following environment variables: prefect config set PREFECT_SERVER_API_HOST=192.168.1.103 prefect config set PREFECT_API_URL="http://192.168.1.103:4200/api" Then running prefect server start works. I can access http://192.168.1.103:4200/docs I can execute api health. But the main UI is just a white page, and nothing else. I suspect this is related to Orion UI makes requests only to 127.0.0.1:4200 and ignores PREFECT_API_URL · Issue #5648 · PrefectHQ/prefect · GitHub

Hi

Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 528, in _prepare_and_execute
    self._rows = await prepared_stmt.fetch(*parameters)
  File "/usr/local/lib/python3.9/site-packages/asyncpg/prepared_stmt.py", line 176, in fetch
    data = await self.__bind_execute(args, 0, timeout)
  File "/usr/local/lib/python3.9/site-packages/asyncpg/prepared_stmt.py", line 241, in __bind_execute
    data, status, _ = await self.__do_execute(
  File "/usr/local/lib/python3.9/site-packages/asyncpg/prepared_stmt.py", line 230, in __do_execute
    return await executor(protocol)
  File "asyncpg/protocol/protocol.pyx", line 201, in bind_execute
asyncpg.exceptions.ActiveSQLTransactionError: ALTER TYPE ... ADD cannot run inside a transaction block

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

Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1965, in _exec_single_context
    self.dialect.do_execute(
  File "/usr/local/lib/python3.9/site-packages/sqlalchemy/engine/default.py", line 921, in do_execute
    cursor.execute(statement, parameters)
  File "/usr/local/lib/python3.9/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 561, in execute
    self._adapt_connection.await_(
  File "/usr/local/lib/python3.9/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 125, in await_only
    return current.driver.switch(awaitable)  # type: ignore[no-any-return]
  File "/usr/local/lib/python3.9/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 185, in greenlet_spawn
    value = await result
  File "/usr/local/lib/python3.9/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 540, in _prepare_and_execute
    self._handle_exception(error)
  File "/usr/local/lib/python3.9/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 491, in _handle_exception
    self._adapt_connection._handle_exception(error)
  File "/usr/local/lib/python3.9/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 778, in _handle_exception
    raise translated_error from error
sqlalchemy.dialects.postgresql.asyncpg.AsyncAdapt_asyncpg_dbapi.Error: <class 'asyncpg.exceptions.ActiveSQLTransactionError'>: ALTER TYPE ... ADD cannot run inside a transaction block

When I deployed prefect server, Its getting this error. I am using Postgresql 13.3 version and prefect 2.10

I'm trying to create a

prefect.yaml

file for easier deployments. Both of my deployments use the same Docker image. Thus, the Docker image only needs to be build once. However, with my current implementation, the Docker image is build for both deployments. Is it possible to build the Docker image only once, while still deploying two flows?

Hello Prefect team, hello community, I am looking into the new project feature that was introduced in 2.10 and how to revamp our current CI to this new setup. Currently, all our deployments are created via CI using separate stages for building the image, linking them to a infrastructure block, and creating the deployments. With prefect projects, these could be merged into one deploy CI stage that executes all of the previously independent stages as steps of the deployment process. However, having them as separate stages comes with the huge benefit that we can reuse the image that is created in the first stage for all later stages, meaning that we have to install dependencies only once. With prefect projects, as far as i understand, we would need to install all dependencies as part of the docker image creation and separately on the machine that is running our CI steps, so that the deploy command can succeed. This would increase CI runtime substantially as we are dealing with complex dependencies. Is there any way how we could execute the deploy step inside of the image that was created in the build step or is such a feature on the roadmap? Thank you for any input on this matter.