Hi, I have a function that runs parallelly to fetch data from external api. The

get_data_asynchronous()

function below creates 10 threads and calls the api concurrently. I am using this function in

run_factset_api()

. As a standalone code locally this works fine. But when I schedule a run on prefect, the

run_factset_api()

function exits before execution and returns coroutine object (although locally it returns the desired value). Is there something I should do to facilitate parallel run on prefect?

async def get_data_asynchronous():
    with ThreadPoolExecutor(max_workers=10) as executor:
        with requests.Session() as session:
            # Set any session parameters here before calling `fetch`
            loop = asyncio.get_event_loop()
            tasks = [
                loop.run_in_executor(
                    executor,
                    company.get_company_records,
                    *(session, [companies], {**company_info, **formulas})
                    # Allows us to pass in multiple arguments to `fetch`
                )
                for companies in companies_to_fetch
            ]
            for response in await asyncio.gather(*tasks):
                master = master.append(response, ignore_index=True)
    return master

@task
def run_factset_api():
    loop = asyncio.get_event_loop()
    future = asyncio.ensure_future(get_data_asynchronous())
    master = loop.run_until_complete(future)
    return master

@task
def save_data_to_s3(emmi_reduction):
    s3_resource = boto3.resource('s3')
    s3_resource.Object(bucket, 'factset_output_data.csv').put(Body=csv_buffer.getvalue())

with Flow('api-flow', storage=STORAGE, run_config=RUN_CONFIG) as flow:
     response = run_factset_api()        
     if response:
         save_data_to_db(response)
        
flow.register('pipeline')

We have a function defined as a task as below

@task(log_stdout=True, state_handlers=[notify_run_failure])
def submit_job_run_to_tmc(job_run):

but this is being called twice when we run a flow

Hi all, I was wondering if there's a way of setting environments for my jobs on the Kubernetes Agent. I'm trying that by using the Helm chart, and my

agent

section of the

values.yaml

file looks like this:

agent:
  enabled: true
  prefectLabels:
    - mylabel
  ...
  job:
    ...
    envFrom:
      - secretRef:
          name: gcp-credentials

The secret

gcp-credentials

exists and is correct. Unfortunately, this doesn't seem to work

Hi all, I have a Python task in my flow that has sys.exit() in it meaning I want that task to break if some condition has met and do not proceed further. But, my task keeps on running in this case and does nothing. Any help?

Is there a graphql mutation to change the location of a flow?

Is there a way to see the release notes on the alpha releases of Orion?

Hi everyone, I hope you're having a good day! I just saw that my prefect server is desync with my prefect agent due to a timezone mismatch (server prefect runs as docker, which have their /etc/localtime point to /usr/share/utc/utc and my agent runs with

prefect agent local start

so it uses my local timezone) Is there any configuration option that I am missing that could provide a custom timezone to all the dockers launched via the

prefect server start

?

Hi everyone, please is there a way to pass parameter from one flow to another? I'm using

client.create_flow_run()

to run the dependent flows but when I add parameter to it, it's not passing to the dependent flow.

When upgrading from older versions of Prefect and using KubernetesRun run_config, is it relevant what the version of the agent starting the run is? I see

AttributeError: 'Context' object has no attribute 'image'

when DaskExecutor tries to start up a cluster

I'm trying to add a team member to my prefect cloud account, but the invite button is grayed out. From what I understand, the free tier allows 3 users. Currently I only have myself and one other member under the users list.

Just wondering if there's been anyone using Singularity to run containers rather than Docker?

so from the prefect cli i can get the logs for a run, but these logs seem to come in batches - is there any way to follow the logs in the same way you might tail -f a log file?

We are getting blank white pages on cloud.prefect.io I think it is an issue of CSS. Options are there but it is of white color.. hahaha

Hey everyone, I’m trying to deploy one of my flows over Vertex AI (GCP) and Im getting the following error:

File "/usr/local/lib/python3.9/site-packages/gcsfs/credentials.py", line 84, in _connect_google_default
    raise ValueError(msg.format(self.project, project))
ValueError: User-provided project 'my_project' does not match the google default project 'some_generated_id'. Either

Some code snippets and information about my flow -

from prefect import Flow
from prefect.storage import Docker
from tasks.extract_product_data import extract_data
from prefect.run_configs import VertexRun
from prefect.schedules import IntervalSchedule
from datetime import timedelta

schedule = IntervalSchedule(interval=timedelta(days=1))

with Flow("extract_comparable_products_data",
          storage=Docker(registry_url="us-central1-docker.pkg.dev/xxxx/",
                         dockerfile="./Dockerfile"), schedule=schedule) as flow:
    extract_data()

flow.run_config = VertexRun(machine_type='e2-standard-16', labels=["ml"],
                            service_account='<http://prefect_service_account.iam.gserviceaccount.com|prefect_service_account.iam.gserviceaccount.com>')

The flow has only one task for now for testing purposes. My task is using data from multiple projects of my organization (google cloud projects) so In every google_client interaction I use a specific project as a parameter, for example -

storage_client = storage.Client(project='my_pro_1')

The service account that Prefect use has permissions to all of the relevant projects (In general, storage, Bigquery, Artifactory, Vertex AI) . Anyone familiar with this issue? Thanks.

Hello, I get a strange error when installing Prefect on a VM (see picture bellow) What can I try ? Environment : Ubuntu 18.04, default python 2.7 I changed to 3.6.9, tried to install with pip and pip3

Hey! Noob question here - can’t seem to find where is the .prefect folder to add the slack token in the config.toml as described here: https://docs.prefect.io/core/advanced_tutorials/slack-notifications.html#installation-instructions

Hello, can anyone point me at any kind of comparison between

Prefect-core

and

prefect-orion

? And maybe also some clarity about which of them is run when we go in the cloud path?

I can get Prefect agent running locally and triggered by Prefect Cloud UI. Now I want to move the agent to a Kubernetes cluster on IBM Cloud. There is not a tutorial to list the steps and the docs from Prefect is grouped around some concepts that I don’t know what I need to accomplish the goal. Can you please list the steps? for example: 1. set flow storage to docker 2. Push the code to container registry in the kubernetes cluster 3. … I am hoping with this list, I can look up the Prefect docs to figure out what to do.

Hi team! Im having a few stability issues with prefect again. Ive noticed some hanging tasks, so I used that big “Cancel All” button in the prefect dashboard to try and clean things out. Except the tasks are now stuck, theyre just stuck in a different state. What would be the proper way to say more forcefully “Kill all tasks and get yourself back in working order?”

one of the recurring issues i have with prefect is that things often seem to fail for no reason, and frequently with absolutely no context. i restarted this flow a minute later with no changes at all, and it ran successfully. are there any recommendations about handling these situations?

Hey! Is there a way to run async code (async def function) inside a prefect task? Should I create a loop and then pass this coro in loop.run_until_complete () inside @task?

Ever seen this error in Kubernetes?

HTTP response body: {"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure","message":"pods \"dask-root-elided\" is forbidden: User \"system:serviceaccount:default:default\" cannot get resource \"pods/log\" in API group \"\" in the namespace \"default\"","reason":"Forbidden","details":{"name":"dask-root-elided","kind":"pods"},"code":403}

Was trying to install a new Prefect agent and it seems the RBAC setup is not right? The agent was configured with

prefect agent kubernetes install --key $PREFECT_API_KEY --rbac --label my_label_here

Hi, has anyone seen this error before on Kubernetes

Failed to load and execute Flow's environment: AssertionError(None)

. Trying to access private github repo for first time. Have set the github token. Running on gke

flow.storage = GitHub(
        repo="Infrastructure/prefect",                           # name of repo
        path="_cfg/server/hello.py",                   # location of flow file in repo
        base_url="<https://github.private-repo.com/>"
    )

we are setting the env variable which contains the token

Hi, why prefect agent require role and rolebinding permissions when deploying agent on gke?

Hi, can you recommend an article which covers the following aspects? 1. How to use Prefect without Prefect Cloud, if you still want a Web UI? 2. What are the alternatives to Prefect Cloud? 3. What are the advantages and disadvantages of those alternatives? 4. Probably too specific: (How to setup Prefect on AWS Fargate without Prefect Cloud, but with UI)

In Prefect Cloud there are Cloud Hooks and Automations. What would be a typical use case for each of them? Or specifically: is notifying when a flow run fails best done with a hook or with a automation?

Has anyone ever attached a service account to their agent with helm? I’m getting 403s after attaching a service account - including error message in thread

Hi all, I'm Daniel from Sentra ✌️We're psyched to leverage Prefect and this great community