Hi, when creating a class in a separate library. What is best practice so that the prefect logger picks up the log lines? Currently I'm on this path but I'm struggling to see them in the gui.

from logging import getLogger

logger = getLogger(__name__)

Class MyClass:
    def __init__(self):
         pass

    def my_method(self):
         <http://logger.info|logger.info>(‘hi’)

Does that make sense?

Hi. Is it possible to restrict number of parallel map tasks with DaskExecutor?

Our Team uses the task BigQueryLoadGoogleCloudStorage to load in blob uris to Bigquery. The docstring specifies that uri is an optional string https://github.com/PrefectHQ/prefect/blob/master/src/prefect/tasks/gcp/bigquery.py However, uri is used as the source_uris parameter for the load_table_from_uri method of the bigquery client. And according to the gcp bigquery client documentation, that parameter can be either a string or an iterable of strings https://cloud.google.com/python/docs/reference/bigquery/latest/google.cloud.bigquery.client.Client#google_cloud_bigquery_c[…]ent_load_table_from_uri Is there any reason why the prefect Task BigQueryLoadGoogleCloudStorage has it specified that its an optional string and not Union[str, Sequence[str]] ? Looking at the code it looks like it should be fine, but I want to make sure we dont fall into an anti-pattern.

Hi, I was wondering if there was a way to stop a scheduled flow if it failed X number of times?

What’d cause this error? I’m using an eagerly loaded template now (that’s in the project tree) as opposed to an S3 file path for the

job_template_path

now. Same template loaded via S3 during flow init didn’t throw this error, but registration blows up with this error now.

I've been looking at the nice

prefect_test_harness

in 2.0 and am wondering if any such fixtures exist for 1.2? Are there any general guides for setting up tests for projects using prefect?

Hey, just noticed that the API key i created last month doesen't show up in the prefect cloud API key tab. I can still see it whith

prefect auth list-keys

command but not in prefect cloud. Is it normal ?

Can you set task dependencies across flows? 🧐 We have some models that depend on prod models in our main flow but it would be nice to separate them into their own pipeline- however I don't want an unrelated task failure to prevent the dependent pipeline from running, so it would be better to set it on a task level

I am looking to setup Docker storage with multiple flows. I found this example in docs: https://docs.prefect.io/orchestration/recipes/multi_flow_storage.html but this example assumes all flows are in the same file, and there is no way to specify the file_path in

storage.add_flow()

method. Is there a way around it?

We are using prefect cloud for our org. If we use Git or Github storage, I believe the GITHUB_ACCESS_TOKEN is going to be stored on prefect cloud server? That’s basically like allowing prefect access to our private repos. Is there a way to avoid storing the access token on the cloud?

What’d cause a KubernetesRun to get stuck in this state?

Hi all - I'm trying out Orion w/ Kubernetes + Dask, and Google Cloud Storage for storage. I'm running into a pickling error (see thread) - any idea what I'm doing wrong?

Hey guys! Awkward new-guy first question incoming: I just installed Orion (

pip install prefect>=2.0b

) into an environment with mypy, and since there doesn’t appear to be a

py.typed

file in

prefect

, mypy is skipping it:

Skipping analyzing "prefect": module is installed, but missing library stubs or py.typed marker

I threw a py.typed in there (

touch py.typed

), but obviously this is less than ideal. Am I missing something obvious?

One more question! How can I specify resource requirements for Kubernetes in Orion? In v1 I used the JOB_MEM_REQUEST env variables (and similar): https://github.com/PrefectHQ/prefect/blob/master/src/prefect/agent/kubernetes/deployment.yaml#L36, but those don't seem to have an effect on the Orion agent.

Hi. Im creating my first flow - which is an daily ETL flow that reads data from csv-files and writes the data to big query. But now i wonder if there are any recommended ways to safeguard that duplicates aren’t written to big query if the flow it’s executed twice. I was thinking of using the cached_key_fn to not rerun the write task but feel unsure if that’s how it’s supposed to do. (I would rather have the task to be skipped..)

Hey ! Thanks to Kevin, I'm trying out a method to use

loguru

logger in Prefect tasks as described here. At first glance it works pretty well but it looks like

DEBUG

logs aren't sinked into Prefect. Any idea what I could do ?

Hi, wrt keeping the database size and performance good, is there a way to set some sort of retention rate policy on the database that maintains the flows, flowruns, etc?

Hi folks, our prefect kubernetes agent sometimes fails to deploy flow runs due to a connection error - namely a ReadTimeoutError (more details in the thread) Note that the same agent is able to deploy tens to hundreds of flows every day but seems to hiccup with these errors, leading to flow runs never getting submitted on our kubernetes cluster.

I'm trying to start a prefect orion flow from a flask route handler but I have an error when calling the

flow

decorated function "MissingContextError: No settings context found" What am I missing ?

hey, i'm trying to start a flow from cli with param flag. I have a boolean param that i want to set to true so i use

--param bool_param=True

but

True

is treated as a String during flow execution. How can i pass a boolean param in cli for flow run ?

Hi!, I saw an update in Prefect 2.0 doc that R will also be supported as flow language, Will Golang also be supported?

i have prefect flows running on GKE using prefect cloud, and GKE is killing my pods while they are running flows, and im not sure why? can anyone help me? in prefect i see

Kubernetes Error: pods failed

and

no container statuses found for pod

and in GKE i see

SyncLoop DELETE

and

audit_log, method: "io.k8s.core.v1.pods.delete", principal_email: "system:serviceaccount:kube-system:generic-garbage-collector"

and

SyncLoop REMOVE

Hello, I'm trying to register flows that are generated dynamically based on a yaml file and that are stored in a list. In short, I'm using code like this (flows is a simple python list of Flow objects):

for flow in flows:
        flow.register(
                project_name=f"Data Platform - {os.environ.get('PREFECT_TARGET_ENV', 'Develop')}",
                add_default_labels=False,
                set_schedule_active=True if os.environ.get('PREFECT_TARGET_ENV', 'Develop') == 'Production' else False
            )

This has worked for me before but since I upgraded prefect to 1.2.1, I'm getting an error message related to the Azure storage I'm using for the flows:

azure.core.exceptions.ResourceExistsError: The specified blob already exists

Any idea why this strategy is not working anymore?

Hello! I'm trying to clone a bitbucket repository into a Prefect Docker image (using the

ShellTask

task) but keep running into this error. Any suggestions? The SSH key is set up correctly, but there doesn't appear to be a user associated with the

ShellTask

commands. This wasn't an issue previously, as older flow images can pull git projects just fine. Only came up recently after I rebuilt the image again (Prefect version & python version are still the same).

Cloning into './dbt_project'...
No user exists for uid 1000190000
fatal: Could not read from remote repository.

Please make sure you have the correct access rights
and the repository exists.

cloudpickle

v2.1.0 was just released an hour ago and it has broken the healthcheck when we try to register a flow. We're using the

prefecthq/prefect:0.15.6-python3.7

as a base image. We now get this error when building the docker image.

Step 21/21 : RUN python /opt/prefect/healthcheck.py '["/opt/prefect/flows/prefect-dbt-run-modelling.prefect"]' '(3, 7)'
 ---> Running in 1d14c333ced9
Beginning health checks...
System Version check: OK
Traceback (most recent call last):
  File "/opt/prefect/healthcheck.py", line 152, in <module>
    flows = cloudpickle_deserialization_check(flow_file_paths)
  File "/opt/prefect/healthcheck.py", line 44, in cloudpickle_deserialization_check
    flows.append(cloudpickle.loads(flow_bytes))
AttributeError: Can't get attribute '_make_function' on <module 'cloudpickle.cloudpickle' from '/usr/local/lib/python3.7/site-packages/cloudpickle/cloudpickle.py'>

Hello, more a python issue but encountered when trying to execute this tuto. I have a project structures that looks like the attached image. This structure is the same as @Anna Geller github repo for the tuto. In her flow codes, she imports functions from a

config.py

file in

flow_utilities

folder like this

from flow_utilities.config import funcA funcB

. When i do so, i get an error. So i tried to reproduced it with the dummy code you have on the image. I get the same error which is

ModuleNotFoundError: No module named 'flow_utilities'

. This really seems stupid to me as it is just importing some func from local file but somehow i can't make it work with this file architecture. What am i doing wrong ?

New issue: When using

Secret.get()

it apparently only search for local secrets. The doc says i have to change

prefect.config.use_local_secrets

to false but i can't understand how... I suppose it is in the config.toml file but as the flow has to run on an eks cluster, how can i specify this for the cluster ?

does anyone have a link to an example Dockerfile for a custom image that a flow might run on?

When attaching a log handler (e.g., slack) to prefect logger, then whenever running a flow with dask, it will log multiple times with the handler - at least this is what we see with our implementation - Is there a good example for attaching a custom log handler to prefect orion logging?

if I want to create flows dynamically like this (registration happens after)

def build(...):
    with Flow(...) as Flow:
        ...
    return flow

is it possible to use docker storage? thinking on possible issues trying to find the path of the flow inside the docker file, there won't any "flow" variable on the module level