I am attempting to create a custom storage block (Digital Ocean uses an

https://

prefix but

s3://

is hardcoded in the S3 filesystem object). I did this by copying the S3 storage object code and removing the hardcoded prefix but leaving everything else the same (except name and credential arguments). The block successfully registers to prefect cloud. However when I try to attach it as the storage block I get the error in the comments. From what I can tell this is happening during the validation step. At some point prefect is looking at the registry for the

Block

type via:

from prefect.utilities.dispatch import lookup_type, get_registry_for_type
from prefect.blocks.core import Block
registry = get_registry_for_type(Block)
print(registry)

which returns all the prefect defined blocks. If I try to build a deployment it will fail because the custom filesystem class I created does not show up in that registry. Currently the remote file system object stores all ‘settings’ as plain text which isn’t ideal because I’m storing access credentials. Thoughts on the best way to proceed?

Hi! I might be missing something silly, hope you can help. I have an image pushed to my Google Cloud Artifacts Registry (

europe-docker.pkg.dev/project/folder-staging/image_name:latest

), previously this image had an issue as it was missing a dependency (

pandas

lib) - I fixed this and the image was re-deployed via our CI, I've verified that the image now does have the dependency installed correctly. The CI was also triggered to re-deploy the flow by calling the following function:

def deploy_projects_observability_export_flow():
    projects_observability_export_flow.deploy(
        name="projects_observability_export_flow_k8s_deployment",
        work_pool_name="default-worker-pool",
        image="europe-docker.pkg.dev/***/prefect-runners-staging/default:latest",
        tags=["staging"],
        schedule=CronSchedule(cron="0 0 * * *", timezone="UTC"),
        job_variables={
            "cpu_request": "110m",
            "mem_request": "128Mi",
            "cpu_limit": "1200m",
            "mem_limit": "8Gi",
        },
    )

However, the flow runs that are set to leverage this image keep showing the

ModuleNotFound

error. Do you happen to know what I'm missing?

We’re having trouble using Blocks with SecretStr on the latest prefect (2.14.5) with pydantic 2 (2.5.1). When I paste the example from the prefect docs into a ipython session, I get an error:

from typing import Optional

from prefect.blocks.core import Block
from pydantic import SecretStr

class AWSCredentials(Block):
    aws_access_key_id: Optional[str] = None
    aws_secret_access_key: Optional[SecretStr] = None
    aws_session_token: Optional[str] = None
    profile_name: Optional[str] = None
    region_name: Optional[str] = None

RuntimeError: no validator found for <class 'pydantic.types.SecretStr'>, see `arbitrary_types_allowed` in Config

Anyone have any ideas? I know pinning

pydantic<2

will make the error go away, but since Prefect supports pydantic 2 I’m trying to understand if this is a Prefect bug or if we need to do something differently with our blocks on pydantic 2.

Hello, Prefect webhooks and events is only for Prefect Cloud? Are we being left behind if we self-host

Hi, I have a very newbie question related to objects returned by a task when composing a Flow. I have a couple of procedures that I want to execute and they return objects. I would like to pass to the subsequent procedure a parameter based on a property which is a member of the object returned by the first procedure. Something like the code below:

import random
from prefect import task, Flow


class Generator:
    def __init__(self, data):
        self.particle = data


@task
def run_generator():
    return Generator(random.random())


@task
def compute(particle):
    print(particle**2)


with Flow('flow1') as flow:
    gen = run_generator()
    compute(gen.particle)

When I try to create the flow I get the following error:

---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
/var/folders/7g/2_wk8ddx7h5_4r12flp9031m0000gn/T/ipykernel_3073/2988601640.py in <module>
      1 with Flow('flow1') as flow:
      2     gen = run_generator()
----> 3     val = compute(gen.particle)

AttributeError: 'FunctionTask' object has no attribute 'particle'

Is there any chance I can have the delayed execution of the parameter until the flow is executed? I know that changing

compute

to receive the object and handling the data access there works but not always I have the possibility to change the function and it would be bad in my case to create many wrappers for users of my library.

Hi all, Quick question about flow.from_source(). Can I use this from within the same repo? As in, can I do something like this from within the same repo defined by the GitRepo url? @flow async def my_flow(name: str = "world"): print(f"Hello {name}!") if name == "__main__": DEFAULT_GH_SECRET_BLOCK_KEY = "github-secret" access_token = Secret.load(DEFAULT_GH_SECRET_BLOCK_KEY).value flow.from_source( source=GitRepository( url="https://github.com/company/my-repo.git", branch="main", credentials={"access_token": access_token}, ), entrypoint="/builtin_worklflows/paige-test.py:my_flow", ).deploy( name="my-deployment", work_pool_name="my-k8s-job", job_variables=dict(pull_policy="Never"), schedule=(CronSchedule(cron="0 0 * * *", timezone="America/Chicago")), tags=["test1","test2"], paramaters=dict(name='Paige') )

Hey everyone! I managed to write the necessary Infrastructure as Code (via Pulumi) for a light-weight Prefect agent that runs as an ECS Service. I was hoping to get a nice welcome message in the logs. Unfortunately, I get the

prefect

cli help message over and over again. ECS console shows a list of

STOPPED

tasks, which I'm assuming are various attempts to start the service and keep it running.

In Prefect 2.0, Is there any option via config to change the default port number 4200?

how do i get a list of all prefect deployments in python

Hiya, me again with another Prefect question! We are moving our apply of deployment yamls to a github action. It seems to try and run, but we get an error that says "did not conform to deployment spec: JSONDecodeError('Expecting value: line 1 column 1 (char 0)')" but when I run this same file with a deployment apply on a local test server it works fine so doesnt seem to be the yaml itself? Any help would be much appreciated!