@Marvin how can I automatically retry Crashed flow

In Prefect 2.0, I’m trying to create a Deployment which takes arguments from a configuration stored in Prefect Cloud as a custom block. Here is the sample code:

config = CustomConfig.load('my_block')

Deployment(
    name="My deployment",
    flow=my_flow,
    parameters={
        "password": config.password
    }
)

when I run this command from CLI

prefect deployment create my_deployment.py

I get this error

AttributeError: 'coroutine' object has no attribute 'password'

Failed to load deployments from 'my_deployment.py'
sys:1: RuntimeWarning: coroutine 'Block.load' was never awaited

How can I do that?

I'm creating an integration with datadog and I need to add a custom handler to prefect's flows. I need to add a handler that gets all logs from the run and spits it out as json. I've seen a lot of examples like the one bellow, but that's not what I need. I don't want to add a handler to each logger.

logger = prefect.context.get('logger')
logger.addHandler(log_handler)

I need something like this

log_handler = logging.StreamHandler()
log_handler.setFormatter(DatadogFormatter())
get_logger().addHandler(log_handler)

I add the handler to the 'root' logger and everything is logged the way I need. I've tried the latter and it works fine with

flow.run()

, but when I register the flow I can't get it to work with k8s. More details in the thread

@Marvin I have prefect 3 and pydantic 2.8 I am getting this error File "/home/ubuntu/cleargrid-dwt-prefect/dwhenv/lib/python3.12/site-packages/pydantic/_internal/_generate_schema.py", line 2227, in _extract_get_pydantic_json_schema raise PydanticUserError( pydantic.errors.PydanticUserError: The

__modify_schema__

method is not supported in Pydantic v2. Use

__get_pydantic_json_schema__

instead in class

SecretStr

.

Hi all. I just started using prefect and am trying to deploy my first flow. I can successfully deploy a python file, but the error I get is: Failed to start process for flow run 'd7c862c3-f1f2-4cbd-b00d-b1248a9add0f'. FileNotFoundError: [Errno 2] No such file or directory: '/Users/caseym/PycharmProjects/virtual' Is this because I deployed while in a virtual environment? I'm using PyCharm. Any tips?

Hello friends, Is it possible to pause a prefect workflow, have it not consume resources, then turn it back on later?

how could I retry a failed task from the UI? Or skip a failed task and continue the rest of the flow from the UI? I feel like this is a very reasonable operation, but cannot find it in the UI. (I know there is a retry mechanism in the api) Thank you

@Marvin I have a long running task that loops through a long list. Occasionally this task fails during the loop. How would you recommend I save the last entry processed so that on a retry, the task does not iterate over the records already processed in the previous task run

@Marvin how to disable task caching system

Hello here, I am new to Prefect and run my first flow in my local K8S cluster and use

KubernetesRun

. The flow is trying to download the

py

file from a local git repo in the same local K8S cluster. But I got an

SSL_VERIFICATION

error even though the

url/host

for

git storage

is

HTTP

. Is there anyone having similar issue?