When should we expect a prefect docker image based on python 3.10+?

i have a general question about how best to utilized Prefect for training (and Pickling) machine learning models. This is my mini-MLE setup: I have a Gitlab repo that holds all my model definitions. (model-repo) I have a Giltab repo where Prefect lives. (prefect-repo) I use poetry to create a package with all my model definitions. I import that into the Prefect repo, and use prefect's tasks to train, pickle the models. This is great, but super super slow and a terrible development process. Every time I make a change to a model in the model-repo, I have to recreate / repackage the Repo. I have to get changes in to main, update a tag version, etc etc. Then I have to update the pyproject file in the prefect-repo (ugh). There's got to be a better way. Curious how others do MLE with Prefect. Thanks! 🙏

Is there a way to unit test tasks where

.fn

returns a coroutine? I tried this:

from prefect_aws.s3 import s3_upload

...

with prefect_test_harness():

my_upload = s3_upload.fn(bucket=S3_BUCKET_NAME, key='B5_key', data=b'55555', aws_credentials = AwsCredentials())

asyncio.run(my_upload)

But it’s not working with this runtime error:

E           RuntimeError: There is no active flow or task run context.

../valo-prefect-poc/.venv/lib/python3.7/site-packages/prefect/logging/loggers.py:91: RuntimeError

Hello all, I am trying to install prefect 2.0

pip install -U "prefect==2.0b1"

and I am encountering below error my machine

Ubuntu 20.04.4 LTS

.

The basic tutorials (https://orion-docs.prefect.io/tutorials/first-steps/) works even with the error but I am facing another issue while running a simple POC Regression model with prefect. Just want to make sure if the issue is not due to the above error.

flow.register is giving me this error

Newbie ALERT: Trying to run a basic regression model using

prefect 2.0

. Facing below error:

Traceback (most recent call last):
  File "/home/kku/work/prefect_poc/env/lib/python3.8/site-packages/prefect/engine.py", line 467, in orchestrate_flow_run
    result = await run_sync_in_worker_thread(flow_call)
  File "/home/kku/work/prefect_poc/env/lib/python3.8/site-packages/prefect/utilities/asyncio.py", line 52, in run_sync_in_worker_thread
    return await anyio.to_thread.run_sync(context.run, call, cancellable=True)
  File "/home/kku/work/prefect_poc/env/lib/python3.8/site-packages/anyio/to_thread.py", line 28, in run_sync
    return await get_asynclib().run_sync_in_worker_thread(func, *args, cancellable=cancellable,
  File "/home/kku/work/prefect_poc/env/lib/python3.8/site-packages/anyio/_backends/_asyncio.py", line 818, in run_sync_in_worker_thread
    return await future
  File "/home/kku/work/prefect_poc/env/lib/python3.8/site-packages/anyio/_backends/_asyncio.py", line 754, in run
    result = context.run(func, *args)
  File "car_linearregression.py", line 104, in do_regression
    X,y = get_feat_and_target(df_car,target)
TypeError: cannot unpack non-iterable PrefectFuture object

It's a simple serial execution where a

flow

function calls different

Task

functions serially. very similar to the below tutorial on the official website.

import requests
from prefect import flow, task

@task
def call_api(url):
    response = requests.get(url)
    print(response.status_code)
    return response.json()

@task
def parse_fact(response):
    print(response["fact"])
    return 

@flow
def api_flow(url):
    fact_json = call_api(url)
    parse_fact(fact_json)
    return

So far I have tried creating a new virtual env and install minimal packages required to run the ML model but had no luck. Could you please help me with this.

Seeing a strange error when scaling up a prefect workflow

File "/venv/lib/python3.8/site-packages/prefect/orion/orchestration/rules.py", line 534, in __aexit__
    await self.after_transition(*exit_context)
  File "/venv/lib/python3.8/site-packages/prefect/orion/database/dependencies.py", line 112, in async_wrapper
    return await fn(*args, **kwargs)
  File "/venv/lib/python3.8/site-packages/prefect/orion/orchestration/core_policy.py", line 190, in after_transition
    cache_key = validated_state.state_details.cache_key
AttributeError: 'NoneType' object has no attribute 'state_details'

Any idea how to debug/interpret this?

Hey Prefect, long time! I'm getting some buggy flows stuck running for 24 hrs + that should be done in 10 minutes approx. Is there any way I can set a maximum run time for my flows? I use an ECSRun config. Cheers.

Hello everyone! How should I manage Secrets on Prefect 2.0? I cant find information on the orion-docs website 😞

Hi, we are using prefect Local Agent wrapped in a docker container, we would like to switch to Docker Agent. Does someone knows if it's possible to setup a Docker Agent with Docker IN Docker (dind) ? Thanks !

Hi, How to install prefect CLI on mac ?

Has anybody successfully installed Prefect on a GCP Compute Engine instance? I'm following this tutorial but running into some issues during the installation: https://medium.com/the-prefect-blog/prefect-server-101-deploying-to-google-cloud-platform-47354b16afe2

ModuleNotFoundError: No module named 'markupsafe'

Hi Prefect team. I am trying to use the great Expectation task in my prefect flow and i cannot mange to get it work with custom expectations that i wrote myself - https://docs.prefect.io/api/latest/tasks/great_expectations.html#rungreatexpectationsvalidation. It doesnt seem to pick them up. Would you know if we can use custom expectations with prefect, and if we can, does it require any additional import in the code? Thank you for your help !

I maintain a tool to package (set

flow.storage

and

flow.run_config

) and registration flows for my enterprise. Recently we wanted to duplicate all Prefect Cloud UI logging to Cloudwatch. Inside an individual flow I can add this code to get the logs there, but I was wondering if there was a way I could do this through a central utility?

with Flow("My First Flow") as flow:
    logger = context.get("logger")
    logger.addHandler(
        watchtower.CloudWatchLogHandler(
            log_group_name="prefect-logs",
        )
    )

Aka, would something like this work?

from prefect.utilities.storage import extract_flow_from_file

flow = extract_flow_from_file("path")
flow.logger.addHandler()?
. . .
flow.register()

Managed to run my prefect2 flow on an agent and consuming from a work queue, with no problem! Not bad, #PrefectJoys When can we expect multiple workspaces and team-workspaces? (if ever?)

I'm getting an error attempting to schedule a flow. The "Schedule" toggle, in this case, is on and was turned on without error. The GraphQL error only pops up after submitting the schedule

Hi again! I’ve got a Prefect2.0

prefect-gcp

authentication issue/question. I’m trying to follow the example docs, but getting an error Importing

GcpCredentials

:

NameError: name 'SecretManagerServiceClient' is not defined

. FIY the prefect1.0 GCP/bigquery tasks are working fine in my environment. Any advise?

Hey all, I'm working on a Prefect 2.0 implementation on docker running on GCP compute engine. One issue is the bash script to get everything up and running. This is what I came up with, which seems to work. I'm wondering if I am overcomplicating this and there is an easier way to do it? The same thing was much simpler with Prefect 1.0 (set api key, register flow, then start agent).

prefect cloud login --key <MY_KEY> -w <MY_WORKSPACE>

rand="GCS_"$(cat /dev/urandom | tr -cd 'a-f0-9' | head -c 16)

output=$(printf '%s\n' 2 <MY_BUCKET> <MY_PROJECT> $rand | prefect storage create)

storage_id=$(echo $output | grep -oP "(?<=identifier \').+?(?=\')")

prefect storage set-default $storage_id

prefect deployment create my_flow.py

output=$(prefect work-queue create my_queue | grep -oP "(?<=UUID\(\').+?(?=\'\))")

prefect agent start $output

follow up

prefect-gcp

issue using the

bigquery_insert_stream

task. I’m not able to pass

Timestamp

data-types -getting:

TypeError: Object of type Timestamp is not JSON serializable

Hi again! I am using

create_flow_run

and

wait_for_flow_run

(see previous thread). The child flow takes a few hours to run, and in the meantime, the parent flow decides that it must have failed and reports that the whole parent flow failed. Is there a way to extend the timeout period for

wait_for_flow_run

?

Hi there, I'm getting a new graphql client error and trying to figure out the cause. This is running on CI/CD and no related code has changed recently...

Hello! Is flow registration to prefect cloud asynchronous? I am trying to register multiple flows concurrently. Are there any examples of that?

Any solution for this: https://github.com/PrefectHQ/prefect/issues/4737 ?

server2.0 UI idea: it would be nice to show the start time of each flow/task run in the UI

Hello, I use a cron schedule for my prefect flows with the following format "m h * * *". Today a subset of my flows did not run at all, and there were no changes/new registrations made recently. Has anyone else encountered this issue? The previous and scheduled flows look good to me

Hello, for some reason we are not seeing any logs in the prefect cloud UI today, even though the flows we ran yesterday were logging normally. Is there any recommended way to debug this? We are running flows via an ECS agent and all the logs for the task running the agent look normal. Flows are still completing successfully, just no logs.

Is there a convenient way to override the flow execute command?

Is there a way to customize the

name

of a

SnowflakeQuery

task similar to @task to give it a human meaning name? I didn't see it in https://docs.prefect.io/api/latest/tasks/snowflake.html

Hi, I'm trying to find out how I can set up a flow that depends on some other flow , but where the other flow is defined elsewhere say in a separate repo. Can I for example reference the flow id in the dependent flow? Or how should I do that?