Hey everyone! I'm trying to have a task that utilizes the

slack_notifier

. I want some custom text in this this task to show up nicely formatted in slack. So far, I tried to return my custom text, and

raise SUCCESS(message='my_text')

. Both aren't what I hoped for. Any ideas?

Hi, I am using GitHub storage for my flow. In my flow, I want to load a data file (a json file located in the same GH repo as the flow file) and use it in my flow code. I can't figure out how to do this. Could someone please help me with this? Thanks!

Hi, I am trying to do task level parallelism by mapping. I was wondering how to set it up so each task is independent say that one of task’s failure would not affect other tasks being paralleled. Thanks in advance!

Maybe a small question, but I cannot find the answer in the docs. Is it possible to use Prefect to process minibatches in which a flow is scheduled to run at, let's say, every 5 mins but can only run in case the previous run is completed (can be successful or failed)? Thanks in advance for the info!

Hey all, wanted to post here before creating an issue on github. But currently having an issue getting my ci/cd registration job working when trying to update

for flow in ${flows[@]}; do prefect register flow --file flows/$flow --project my_project --skip-if-flow-metadata-unchanged; done

(which currently works) to

prefect register --path flows/ --project my_project

where my flows are utilizing azure storage and are located within flows folder. Using the new register command does create a new blob, but then errors out

Error building storage: azure.core.exceptions.ResourceExistsError: Operation returned an invalid status 'The specified blob already exists.'

for each flow ultimately exiting out the job. For now I will just continue to use ``prefect register flow``, but wanted to see if anyone was having similar issues.

Hi! Just discovered that Prefect.io/slack link is on longer active, so new friends will not be able to join this workspace. Is this something that can be fixed soon?

Hey everyone — I’m trying to use a shellTask and it is working but I’m getting duplicated logs, with slightly different formatting but identical content. I can’t seem to find a way to disable/hide the duplicated logs. Here’s an example of duplication:

[2021-06-22 13:25:46+0100] INFO - prefect.ShellTask | 13:25:46 | Concurrency: 12 threads (target='dev')
INFO:prefect.ShellTask:13:25:46 | Concurrency: 12 threads (target='dev')

and I have

stream_output=True

on the shelltask

Hi, I’m trying to get on a secret on my cloud account, and having trouble with that I can successfully set a secret using

from prefect import Client

client = Client()
client.set_secret(name="MYSECRET", value="MY SECRET VALUE")

I can see ‘MYSECRET’ on my cloud UI, so I’m sure I’m authorized etc. But I am unable to get a secret with:

from prefect.client import Secret

# Load the value of `MYSECRET`
my_secret_value = Secret("MYSECRET").get()

Where I get a

*** ValueError: Local Secret "MYSECRET" was not found.

error. Reading the docs, I see that it should be looking for a local secret first, and if not found, will look for a cloud secret. I think it is not looking for a cloud secret

Does anyone know if I just run

prefect agent kubernetes start --job-template

again to amend my existing agent? I'm trying to edit the safe-to-evict setting.

Is there a way for me to define a Flow in such a way as to be generic and then I can subclass it for each child flow? I am looking to reduce code repetition for some flows I have that follow a general pattern, but have small differences, or added steps/tasks. I don't think the imperative API is what I am looking for.

Second question is whether or not there is a way to provide state (and define within the Flow) state than using the Parameter tasks. I know the

context

exists, but I am looking for something that can read a file and set the state. I don't think the context object is setup for that. It seems to be more runtime to me

Hello, we are having some issues around the logging. Some, but not all of the logs are appearing in the dashboard. Any ideas how I might troubleshoot this? We are using ECS.

TL;DR I’m seeking a source for converting AirFlow to Prefect. Hi All. I have inherited what amounts to an AirFlow demo that I would like to sort out how to convert over to a Prefect demo. I am a graduate student in a practicum and the previous cohort wrote the AF stuff. Now I am trying to answer the question as to whether AF should remain in place or if Prefect could possibly be snapped in AF’s place before the project gets truly up and running. As a point of reference, I did get a Prefect Cloud demo type of thing up and running last week (it just spun up a container in AWS and spit some stuff out in the Prefect logs). So I’m not a complete newb, but my background is more on the networking than the development side.

Hi, everyone! I'm new to prefect and have a question regarding some use cases that we saw whilst testing it in our environment: - Use of connection pool to connect to databases: - We actually saw this link here (https://github.com/PrefectHQ/prefect/issues/1876) and tried using it, but when we tried to register the flow to the server, it threw the error: " flow.register(project_name='name') TypeError: can't pickle _thread._local objects " - For us this is a major issue as within our flow, ideally we would have some light queries in a lot of tasks, using connection pooling is mandatory for this use case in particular. If anyone could point us in the right direction it would be very helpful! Thanks in advance, Luis

Hey all, does anyone have experience using Great Expectations? We’re having trouble running checkpoints using the prefect tasks, running into ‘object is not subscriptable’ errors. We’ve tried V2 and V3, but have not attempted going down to 0.12 version. Noticed someone else was getting a different error here but made it successfully https://github.com/PrefectHQ/prefect/issues/4411 any ideas?

Hi all! I've just started using Prefect and I was wondering if it is possible to programmatically access task logs using Python - at the moment I'm using the web interface but it would be great if I could grab and parse them via an API. Is this currently possible? Thanks in advance!

How do people set up their repos? We had a monorepo for all our Flows at my last gig, so I'm inclined towards that. But curious as to how others do it!

Hi - I had this problem a few times. Where I am not able to log on to the webpage. I keeps loading. Do you know why that happens, and how to fix it?

Hey folks, I'm getting a weird error on my Kubernetes cluster (AKS) when my agent kicks off its first job. The agent does start successfully, and subsequent jobs are ran to completion, so doesn't appear to stop anything functionally. Just wondered if anyone else has came across this? I am packaging up my dependencies to docker storage, and storing my flows in blob storage. Cheers

[2021-06-23 08:53:35,179] ERROR - agent | Error while managing existing k8s jobs
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/prefect/agent/kubernetes/agent.py", line 384, in heartbeat
    self.manage_jobs()
  File "/usr/local/lib/python3.7/site-packages/prefect/agent/kubernetes/agent.py", line 230, in manage_jobs
    pod_events.items, key=lambda x: x.last_timestamp
TypeError: '<' not supported between instances of 'datetime.datetime' and 'NoneType'
ERROR:agent:Error while managing existing k8s jobs
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/prefect/agent/kubernetes/agent.py", line 384, in heartbeat
    self.manage_jobs()
  File "/usr/local/lib/python3.7/site-packages/prefect/agent/kubernetes/agent.py", line 230, in manage_jobs
    pod_events.items, key=lambda x: x.last_timestamp
TypeError: '<' not supported between instances of 'datetime.datetime' and 'NoneType'

Hi, I have suddenly started to get this error on my flows using the Kubernetes Agent.

ERROR - prefect.CloudFlowRunner | Unexpected error: AttributeError("'Log' object has no attribute 'splitlines'")

I’m running prefect self hosted. Server, agent & flow versions are 0.14.22 Nothing has changed, any idea what could be the reason?

Hi folks, do flow concurrency limits take the flow run parameters into account ? i.e. if I don’t want two flow runs with the same parameters to run at the same time - how do I go about achieving this ?

For those not part of the #show-us-what-you-got channel.

On the Task Run views, why can’t we sort the tasks by Duration?

Good evening from Germany, is there a difference how I activate cloud secret storage on different clients? Is there an example of how I turn in on in the config? The documentation is not very helpful about how to actually do it, for me at least.

Hi — I’m in process of trying to deploy a KubernetesAgent to EKS, I have an EKS cluster with 3 different nodes. Where would I specify which node group I would want to use for a flow run? Would it be in the flow.run_config = KubernetesRun(…)?

Hi are there any best practice recommendations for prefect repositories set up in production?

Messing with the GitHub Action again.  Does this sound right for the sequence of Actions to register flows after a Commit? 1. Checkout repo into Action Runner 2. Install Prefect on the Action Runner 3. Make a config file that has a 

config.toml

 with my auth token (stored as a secret in GitHub) 4. run 

prefect register

 through the CLI Thanks!

Hey, I am using

Docker

storage to push an image to ECR. I am authenticated with

aws sts assume-role

locally in my CLI. I am trying to push the image locally at the moment and getting an error with:

File "/Users/benmuller/code/es-betfair-flows/.venv/lib/python3.8/site-packages/prefect/storage/docker.py", line 585, in push_image

raise InterruptedError(line.get("error"))

InterruptedError: denied: Your authorization token has expired. Reauthenticate and try again.

How can I make sure that my

sts

creds are honoured ?