Dear Prefect Community, is there any fix available to use a specific Postgres schema for Prefect Orion DB? Thanks, Xavier

@Anna Geller, hi Anna, I have non-tech question, do you have any materials/articles about main ideas under development Prefect 2.0? why it was created, what problems & gaps team tried to solve with it?

Hi! Is there an easy way to redirect a python logger to the prefect logger - so the logs would show up in console? I am using a 3rd part library, which uses its own logger (from the standard logging library), and I would want to see its logs in prefect console as well

Hiya does Prefect 2 have the concept of heartbeats and the Zombie Killer?

Question: How can I manually control a task status, without the orchestrator trying to rerun it? I'm returning a (non final) task state and getting automatically retried. I'd like to not retry until the task is

Failed

, or skip retries when it's marked as

Succeeded

, by my code

Hey everyone, just updated to 2.4.2 and all my flows stop working, did I miss a braking change. The error is Response: {'detail': 'protected block types cannot be updated.'}. The strange thing is that the block that is accessed is of type local-file-system and I do not have such block in my system.

Hi, I am using Prefect 1.x for a while now and trying to migrate to 2.x. But I feel like the documentation is missing a lot of examples and it is pretty complicated to migrate to 2.x although I consider the effort on publishing a lot of (too?) basic articles to get started. For example, I don't find anywhere an example on adding parameters to my flow. I'm here trying to register a flow in a S3 storage with pre-defined parameters at flow registration. The parameters include S3 credentials to create a bucket IN the flow.

@flow(name="get_paris_weather")
def get_paris_weather(
    minio_endpoint: str,
    minio_access_key: str,
    minio_secret_key: str,
    minio_use_ssl: bool,
    bucket_name: str,
):
    create_bucket(
        minio_endpoint,
        minio_access_key,
        minio_secret_key,
        minio_use_ssl,
        bucket_name,
    )
    city_coordinates = get_city_coordinates("Paris")
    return get_weather(city_coordinates[0], city_coordinates[1])


# --- Deployment definition

if __name__ == "__main__":
    bucket_name = os.environ.get("MINIO_PREFECT_FLOWS_BUCKET_NAME")
    minio_endpoint = os.environ.get("MINIO_ENDPOINT")
    minio_use_ssl = os.environ.get("MINIO_USE_SSL") == "true"
    minio_scheme = "https" if minio_use_ssl else "http"
    minio_access_key = os.environ.get("MINIO_ACCESS_KEY")
    minio_secret_key = os.environ.get("MINIO_SECRET_KEY")

    flow_identifier = uuid.uuid4()
    block_storage = RemoteFileSystem(
        basepath=f"s3://{bucket_name}/{flow_identifier}",
        key_type="hash",
        settings=dict(
            use_ssl=minio_use_ssl,
            key=minio_access_key,
            secret=minio_secret_key,
            client_kwargs=dict(endpoint_url=f"{minio_scheme}://{minio_endpoint}"),
        ),
    )
    block_storage.save("s3-storage", overwrite=True)

    deployment = Deployment.build_from_flow(
        name="get_weather_s3_example",
        flow=get_paris_weather,
        storage=RemoteFileSystem.load("s3-storage"),
        work_queue_name="flows-example-queue",
        parameters={
            minio_endpoint: minio_endpoint,
            minio_access_key: minio_access_key,
            minio_secret_key: minio_secret_key,
            minio_use_ssl: minio_use_ssl,
            bucket_name: bucket_name,
        },
    )
    deployment.apply()

But what I get as error is :

prefect.exceptions.SignatureMismatchError: Function expects parameters ['minio_endpoint', 'minio_access_key', 'minio_secret_key', 'minio_use_ssl', 'bucket_name'] but was provided with parameters ['False', 'minio', 'minio123', 'prefect-flows', '172.17.0.1:9000']

Could you explain me how I can pass parameters to my flow ?

Hi, I'm trying to migrate from 1.x to 2.x with Docker storage and infrastructure. The following image (from blogpost) states that Prefect can store flows data in Docker images, but the documentation doesn't refer to it. Is it still possible in 2.x ? Is there any example I can find ? Thanks.

hi! i was trying to build a s3 block on github actions . but it fails at s3.save . attaching the screenshot for the same. it would be helpful if anyone faced similar issue and got any leads. thank you

Does DaskTaskRunner support designated workers? Specify in the dask as follows:

future = client.submit(func, *args, workers=['Alice'],
                       allow_other_workers=True)

Hello. I am getting below error while executing Databricks pipeline using Prefect flow on windows machine. Environment details - Python - 3.10 Prefect - 2.4.0, also tested with 2.4.1, same issue persists. Error -

prefect.exceptions.ParameterTypeError: Flow run received invalid parameters:
 - run_name: none is not an allowed value

in jobs_runs_submit_and_wait_for_completion
    jobs_runs_state, jobs_runs_metadata = await jobs_runs_wait_for_completion(
ValueError: too many values to unpack (expected 2)

Hi all, as I understand, to run a flow, this is the steps: • first you create a work queue, and then you start an agent polling that work queue. • if you start an agent without creating a work queue first, the queue will be created automatically If you follow the first case, there maybe work queues created without agents. How do you check which agents are running? I don't see that option in the CLI.

How can I locally logout from prefect cloud?

@Anna Geller I think I have challenge with a GitHub template, I’m trying to implement the serverless with Aws Lambda template on GitHub. Under the environment section of the

serverless.yml

file I don’t understand the KVPs

PREFECT_API_KEY

and

PREFECT_API_URL

are the values supposed to be variables or secrets and from where?

Hi! I've had some problems with flows getting stuck after all tasks are finished and did not receive support so I've tried to recreate a minimal reproduceable example see code below. What is happening for me: The tasks can run fine, but after the tasks completes the memory usage explodes (goes to 100%). Depending on the size of the data the flow will either: • get stuck after last finished task • finish succesfully • either of above intermittently I can run the same flow perfectly fine in Prefect 1, it also runs a lot faster even when increasing data a lot. I attach a log of a unsuccesful prefect 2 run in the thread:

Hello, guys! I believe I´ve found an unexpected behavior that obligates us to use scapes in JSON parameter. I´m running Prefect 2.4.1 under Windows 11. To reproduce the error, just try something similiar this command: prefect deployment build testetalend.py:talend --name talend_deploy --apply --params='{"job_name": "myjobname", "batch_file": "C:/Users/myuser/Documents/testeTalend/myscript.bat"}' After hit Enter, we got this message: json.decoder.JSONDecodeError: Expecting property name enclosed in double quotes: line 1 column 2 (char 1) To workarround the problem, type de same command using scapes in JSON string: prefect deployment build testetalend.py:talend --name talend_deploy --apply --params='{\"job_name\": \"myjobname\", \"batch_file\": \"C:/Users/myuser/Documents/testeTalend/myscript.bat\"}' Anyway, this behavior is very similar that was reported by the issue below: https://github.com/PrefectHQ/prefect/issues/3295 Thanks in advance!

Hello am getting this error when i run a deployment

Hey ! Prefect 1 question : is there no way for me to schedule a Flow with required arguments, even though the schedule has

default_params

that fill required parameters ? I want to make sure users can't run the flow without specifying some arguments but also schedule it.

Hey y'all, I'm getting this weird error when running a prefect 2 deployment from a custom agent running in an ubuntu docker container. My flow code cannot be found when downloading from my s3 storage block during a deployment run. I normally upload the flow code when building my deployment. I can confirm that my flow file definitely exists in my s3 bucket. Any ideas?

When I use REST API with Scheduler enabled (e.g., RRULE), I get the following error in Prefect Orion 2.4. It was working fine in 2.0. Even though I have dedicated storage directory and REST API way of running workflow is pointing to default /tmp folder, not the one I have it using the parameter PREFECT_LOCAL_STORAGE_PATH. But If I run using .yml, it uses the path specified in PREFECT_LOCAL_STORAGE_PATH and it works fine. Please let me know what am I missing in the payload? Flow could not be retrieved from deployment. Traceback (most recent call last): File "<frozen importlib._bootstrap_external>", line 879, in exec_module File "<frozen importlib._bootstrap_external>", line 1016, in get_code File "<frozen importlib._bootstrap_external>", line 1073, in get_data FileNotFoundError: [Errno 2] No such file or directory: '/tmp/tmpz0cmn1ddprefect/AOPS_SQL_Workflow_v1.py'

Hi - for prefect v1 I have a task that fails some of the time. I'd like to have the final flow state determined as success if any of the mapped tasks are a success, and fail if all of them are failed. eg

@task(name="Task A")
def task_a(val):
     if random.random() > 0.5:
        raise ValueError(f"Non-deterministic error has occurred.{val}")

@task
def send_log(val):
    <http://logger.info|logger.info>(f"hello {val}")

l = [1,2,3,4]
with Flow("Trigger example") as flow:
    ta_s = task_a.map(val=l)
    sl_s = send_log.map(val=l, upstream_tasks=[ta_s])
# flow run should be success if ANY of task_a's succeeded. 
flow.set_reference_tasks([something])

Do Parameters work for Task that are classes and not functions? https://docs-v1.prefect.io/core/concepts/parameters.html I am trying to pass a Parameter to a ShellTask but it does not look like it works. Any example of that?

Hi, for prefect 0.15.13 i'm getting this error when trying to set up a mapped task run

Traceback (most recent call last):
  File "/home/wwolfe-mcguire/.conda/envs/dev_env/lib/python3.9/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "/home/wwolfe-mcguire/.conda/envs/dev_env/lib/python3.9/site-packages/prefect/engine/flow_runner.py", line 569, in get_flow_run_state
    executors.prepare_upstream_states_for_mapping(
  File "/home/wwolfe-mcguire/.conda/envs/dev_env/lib/python3.9/site-packages/prefect/utilities/executors.py", line 681, in prepare_upstream_states_for_mapping
    value = upstream_state.result[i]
KeyError: 0
[2022-09-26 15:42:45+0000] ERROR - prefect.ml-graph-energy | Unexpected error occured in FlowRunner: KeyError(0)

I wasn't able to find other examples of a similar error online and was hoping someone could give me some hints about how to start debugging this. thanks

Hi Team - For Prefect 2.0, we would like to use our S3 bucket to store some allocations files. These will be emailed/sftp to clients. I'm not sure what I'm doing wrong with the state and await from the write_path function.

aws_credentials_block = AwsCredentials.load("****-user")
s3_block = S3.load("***-s3")

s3_bucket_block = S3Bucket(
    bucket_name=s3_block.bucket_path,
    aws_credentials=aws_credentials_block,
    basepath=f"Allocations/{date.today().year}"
)
output_file = f'''SR Allocations {datetime.now().strftime('%Y%m%d-%H%M%S')}.csv'''
bytes_to_write = df_alloc.to_csv(None).encode()
csv_file = s3_bucket_block.write_path(path=output_file, content=bytes_to_write)

<http://logging.info|logging.info>('Filtering for **** Trades: Rows = %s, Accounts = %s' % (len(df_alloc), df_alloc.custodianAccnt.nunique()))

return csv_file

@flow(name = "***** Allocations")
def ****AllocationsFlow():
    try:
        slack_webhook_block = SlackWebhook.load("****-webhook")
        state = allocations_process(return_state = True)
        excel_file = state.result()
        if 'Completed'== state.name:
            slack_webhook_block.notify("**** Allocations was successful")
        else:
            slack_webhook_block.notify("***** Allocations failed")

Error: cannot pickle 'coroutine' object 112331.736 | INFO | Task run 'Fidelity Allocations-e728df66-0' - Crash detected! Execution was interrupted by an unexpected exception. 112332.443 | ERROR | Flow run 'visionary-cat' - Finished in state Failed('1/1 states failed.') Traceback (most recent call last): File "/opt/pysetup/.venv/lib/python3.9/site-packages/anyio/_core/_eventloop.py", line 70, in run return asynclib.run(func, *args, **backend_options) File "/opt/pysetup/.venv/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 292, in run return native_run(wrapper(), debug=debug) File "/usr/local/lib/python3.9/asyncio/runners.py", line 44, in run return loop.run_until_complete(main) File "/usr/local/lib/python3.9/asyncio/base_events.py", line 647, in run_until_complete return future.result() File "/opt/pysetup/.venv/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 287, in wrapper return await func(*args) File "/opt/pysetup/.venv/lib/python3.9/site-packages/prefect/client/orion.py", line 82, in with_injected_client return await fn(*args, **kwargs) File "/opt/pysetup/.venv/lib/python3.9/site-packages/prefect/engine.py", line 239, in create_then_begin_flow_run return state.result() File "/opt/pysetup/.venv/lib/python3.9/site-packages/prefect/orion/schemas/states.py", line 159, in result state.result() File "/opt/pysetup/.venv/lib/python3.9/site-packages/prefect/orion/schemas/states.py", line 145, in result raise data TypeError: cannot pickle 'coroutine' object

Hi there I'm wondering how I should store config-files in prefect 2. I saw the options with json-blocks, but I think it would be great to also have an yaml-block. Is there something like this? Maybe with even a bigger field or even a yaml editor helper. Background for this request is: we have a lot of configuration (connection details, mail server settings, special modes) and not everything should be possible to configure for every flow run as parameters. In prefect 1.0 we moved them to the config which was read by prefect.config (still possible to overwrite it with env-vars, even just from a flow run). As parameters we would like to use only such which are important for the daily execution. For all the password I will use the secrets. But I ask more about the normal configuration. Thanks for any hint!

Hi all! I am running my Prefect 2 flows via Prefect Cloud and a GKE k8s autopilot cluster running Docker images. We are working on building up our deployments to be automated and have a couple of questions that are not clear to me from the docs and was wanting to make sure my intuition is correct. We are attempting to move our deployments to Python from the cli and have the following working for requesting custom k8s configurations.

from prefect2_flow import extract_load_transform
from prefect.deployments import Deployment
from prefect.infrastructure import KubernetesJob
from prefect.orion.schemas.schedules import CronSchedule


schedule = CronSchedule(cron="15 * * * *", timezone="UTC")

deployment = Deployment().build_from_flow(
    flow=extract_load_transform,
    name="test_hourly_elt",
    parameters={
        'collection_duration': '1h',
        'uri': '<https://google.com>',
    },
    skip_upload=True,
    schedule=schedule,
    tags=["test"],
    version=2,
    work_queue_name="test-kubernetes",
    infrastructure=KubernetesJob(
        finished_job_ttl=30,
        image="us-central1-docker.pkg.dev/.../prefect2-flows/elt:latest",
        image_pull_policy="Always",
        namespace="test-prefect2",
        pod_watch_timeout_seconds=180,
        job={
            "apiVersion": "batch/v1",
            "kind": "Job",
            "metadata": {"labels": {}},
            "spec": {
                "template": {
                    "spec": {
                        "parallelism": 1,
                        "completions": 1,
                        "restartPolicy": "Never",
                        "containers": [
                            {
                                "name": "prefect-job",
                                "env": [],
                                "resources": {
                                    "requests": {
                                        "memory": "5Gi",
                                        "cpu": "2",
                                        "ephemeral-storage": "1Gi",
                                    }
                                }
                            }
                        ],
                    }
                }
            },
        }
    )
)


if __name__ == "__main__":
    deployment.apply()

1. Is it okay to set

skip_upload

to true? Since we are using Docker I'm not sure what the benefit of uploading our project to GCS is. 2. The Prefect API has a parameter for

is_schedule_active

but it doesn't look like that parameter has made it over to the Python API b/c when I try to add it above it gives me an error about not including more parameters than needed, is this something I can contribute to or add functionality for? 3. The k8s job configuration is pretty verbose to request resources above the default, is there a better way to set these that I am missing? Thank you for the time and any help provided!

If I were to run prefect's docker container in an aws lambda, how would I tell it what my handler is?

FROM prefecthq/prefect:2.4.2-python3.9
COPY service.py .
COPY requirements.txt .
RUN pip install -r requirements.txt
CMD [ "service.handler" ]

For example this doesn't work because I am not using aws' base image so CMD [ "service.handler" ] isn't properly setting the handler

What is the calculation for number of users in your tenent? I have 13 users, and a max of 20... yet I cant invite anyone. I just get the error "This Tenent has the maximum number of users every time I try to invite a new email address. What else is included in that number?

Can I access task state inside of a flow without explicitly passing task result objects to a task? I am trying to emulate the behavior of a state handler from prefect < 2.0

Hey all, trying to read a file with the

RemoteFileSystem

.

rfs = RemoteFileSystem(basepath="<s3://bucket-name/some/path/on/bucket>")
cfg = rfs.read_path("base_config.yaml")

I end up with

cfg

being a

coroutine

. I imagine this returns bytes but there aren't any helper methods and it's not clear how I'm supposed to proceed? I am interested in reading this file into memory.