Hey all! I have a Prefect flow that runs our Airbyte connections. However, we have hard-coded a dictionary of

airbyte_connection_ids

into the flow. I want to reference an externally stored dictionary (i.e. a

.json

file stored in S3) so that it's easier to change when we redeploy Airbyte connections & get new

airbyte_connection_ids

. Is this possible? Can a Prefect parameter take values from an external source? We are running on prefect version 1.2.4

Hey, I’m using a dedicated Github repository as a storage for all my flows. Some of the codes of my flows are stored in another Github repository. I’m using Github block to migrate the code (with

get_directory

function) and execute the code with flow decorator. For example:

from prefect import flow


@flow
def example:():
    with open("my/code/path.py") as f:
        exec(f.read())

I want to know if this is the best practice to do it and also how can I make sure the code that I’m migrating from the other repository is up-to-date before the flow start to run Thanks

Hi, is it possible to override task_definition in ECSTask? Specifically, I would like to specify

secrets

as follows.

"secrets": [
                {
                    "name": "AWS_ACCESS_KEY_ID",
                    "valueFrom": "/copilot/app/prd/secrets/AWS_ACCESS_KEY_ID"
                },
            ]

Thank you.

Hi, we have set up our prefect server and had configured slack notifications for the same too. But we are noticing that sometimes notifications are not getting sent when a flow has been completed/failed. How can we check what went wrong, maybe some logs or how do we fix this?

Given the news around 30 day termination of Docker Hub hosting for OSS projects, do users of the prefecthq image need to do anything to migrate to another source, or are we good?

Upon login, magic links are already expired - I'm wondering if this is a google chrome issue specifically or if others are seeing this?

I’m unsure how to fix this issue… I have 3 flows that are scheduled for the 10th of every month. These three flows are now double scheduled for the 10th. So within a minute or even just seconds of each other, I receive the output from those runs twice. How do I fix this?

Hello! Any clues about why am I getting some database connection issues (see thread) on my environment? The UI seems to work fine but the server keeps raising these exceptions.

Hey Everyone, on Ive been trying to add upload a json file onprem to an s3 bucket, everytime i try to load the S3Bucket though, it keeps raising a ValueError, saying its unable to find the block document. Is there something that I need to load first?

async def uploadJsontoS3(path):
    s3_bucket = await S3Bucket.load("my_bucket")
    with open(r"\\filepathtoJson", "rb") as f:
        s3_bucket.upload_from_file_object(f, "aws-batch-args/testArgs.json")

deployment naming strategy question: is it possible to have separate flows deployed with the same entry point name (e.g. name every entrypoint 'main')? It looks like in the prefect ui, flow runs only reference the entry point, so am I right that it would be confusing to distinguish flow runs from different flows if I were to do this?

from prefect import task

from google.cloud import dataproc_v1beta2 as dataproc

@task

_def_ submit_pyspark_job():

# Create a client to interact with the Dataproc API

client = dataproc.JobControllerClient()

# Set the Dataproc cluster and job parameters

project_id = "<project-id>"

region = "<region>"

cluster_name = "<cluster-name>"

main_python_file_uri = "<main-python-file-uri>"

args = ["<args>"]

# Create the job request

job = {

"placement": {"cluster_name": cluster_name},

"pyspark_job": {

"main_python_file_uri": main_python_file_uri,

"args": args

}

}

# Submit the job to the Dataproc cluster

result = client.submit_job_as_operation(

_request_={

"project_id": project_id,

"region": region,

"job": job

}

)

# Get the ID of the submitted job

job_id = result.name.split("/")[-1]

print(_f_"Submitted PySpark job with ID: {job_id}")

I'm unclear on how to gather/reduce results from using

.submit()

. I want to make a bunch of api calls in parallel and build a list of results. Can anyone help me on this?

hi prefect, is there a simple way to check metadata on deployed agents? ie. suppose i have a work queue with 2 agents listening, is there a way to see even a count of recently polled agents? assuming this exists in the backend as a queue would not know when it becomes unhealthy. It would be great to see things like user, device name etc. as well

Yesterday I left about 1000 pipelines running every 1h with a concurrency limit of 250 and when I went back to see the result, I realized that the process consumed almost 100Gb of ram. I was very confused, because none of the pipelines store values locally, they just query an external database and send the data to another API. But even without storing any value in cache it consumed all the cache storage power of my machine. I imagine that it must be the Prefect's own sqlite database that must be consuming so much resource. Is there a way to limit this or make Prefect stop using cache memory?

Hi prefect community, I have a model that was trained with Prefect workflow and saved in AWS S3 Bucket.

def upload_dir_to_s3(local_model_dir, logger, target_model_dir=None):
    s3_bucket = S3Bucket.load(BLOCK_NAME_IRS_TRAIN_DATA_S3)

    if not target_model_dir:
        target_model_dir = local_model_dir
    try:
        s3_bucket.upload_from_folder(from_folder=local_model_dir, to_folder=f"{target_model_dir}")
        <http://logger.info|logger.info>(f"Success! Directory '{local_model_dir}' was uploaded to '{s3_bucket.bucket_name}' Bucket.")
    except Exception:
        raise IOError(f"Saving to '{s3_bucket.bucket_name}' Bucket Failed!")

Then I download this model from S3 Bucket, however not within Prefect Workflow, but separately with

boto3.client

. Locally it works, but in K8s Cluster I get an Forbidden Error.

botocore.exceptions.ClientError: An error occurred (403) when calling the HeadObject operation: Forbidden

Could you please tell me, what the problem might be? Should I save or read the model in another way?

Hey all, is it possible to have a file stream as a parameter for a deployment?

I have noticed that tasks work on copies of the inputs and not directly with the inputs itself. For example the following flow would run forever.

from prefect import task, flow, get_run_logger
from logging import Logger
@task()
def pop_element_zero(ls: list, logger: Logger) -> list:
    val = ls.pop(0)
    <http://logger.info|logger.info>(f'Just popped {val} from the list {ls}')
    return val

@flow()
def pop_task_flow():
    logger = get_run_logger()
    ls = [1,2,3]
    while ls:
        pop_element_zero(ls=ls, logger=logger)

if __name__ == "__main__":
    pop_task_flow()

It makes sense (really good for making tasks retry-proof when the inputs are immutable objects). But I have some flows where I would like to not have this behavior. Is there a task decorator param that I am missing that would allow the user to let the task work directly on its inputs and not copies?

I’m finally getting everything moved over to 2 and wondering if there’s a way to completely delete/deactivate my 1 Cloud account. I don’t see anything the UI

Hi! In Prefect 2, when a flow run succeeded, I was able to retry the flow run. However, if there are task runs that have succeeded, the flow retry will fail with this error:

Task run 'xxxxx' already finished.
... exception stack trace ...
prefect.exceptions.MissingResult: State data is missing. Typically, this occurs when result persistence is disabled and the state has been retrieved from the API.

If I manually set all task runs in the flow run to failed, then the flow retry will succeed. Is this a bug or a feature? My expectation is when we hit the retry button for flow runs, the state of task runs should get reset. Am I missing something? Thank you!

question, when I run my flows locally, the print statements run as expected. When I deploy and run via the UI the print statements show up at the very end... can someone explain why?

Hello. I see that calling a flow inside another flow (calling a subflow) doesn't allow task methods (such as submit or map). I need to submit runs of a subflow mapped over a list of input. Is there a straightforward way to do this? I saw the docs mentioning AnyIO task groups and asyncio.gather, but it seemed a bit too complicated.

Hi Everyone: I'm running into a problem using a prefect profile. I'm on Windows 10, Prefect 2.8.5, working within a conda environment with python 3.8.16. Running through an example project using dbt, snowflake, prefect by Anna Geller on Medium. From bash, I entered the following, which completed with no errors

prefect profile create dev

then I tried to use the profile by issuing the following command and received the errors below.

prefect profile use dev

Error screen snip:

Hello! I have a flow that runs 11 subflows and for some reason the 10th flow always raises a timeout error when calling run_deployment. Any ideas? thank you!

The command

prefect block register

sometimes raises

alembic.script.revision.ResolutionError: No such revision or branch 'XXXX'

. Anyone know what causes this and how to resolve it?

Hi everyone, is it possible to delete / remove a block that was registered with an Orion server from Python code? Sort of the opposite of

my_block.save("my-block-slug")

Hi everyone, What is maximum and optimal number of queue then can be assigned to one agent ?

Hi Team, Can anyone help me in triggering a flow whenever a api is hit. i'm getting problems with prefect.client and secrets. i also want help with connecting to my prefect cloud

Hello Everyone, I am running a flow in my local system using Pycharm and trying to connect Prefect. I am using API Authentication and i am able to connect to Prefect. I wanted to access Prefect Blocks and using below code :

import prefect
from prefect import flow, task
import json
from prefect_databricks import DatabricksCredentials
from prefect_databricks.flows import jobs_runs_submit_and_wait_for_completion
import scripts
from prefect.blocks.system import Secret, String
from prefect_databricks import DatabricksCredentials
from prefect.blocks.system import Secret, String
@flow(name='Databricks-Flow')
def databricks_flow():
    secret_block = Secret.load("prd-databricks-token")
    print(secret_block)
if __name__ == "__main__":
    databricks_flow()

While executing this i am getting below error: Can anyone please help me in resolving this. FYI: This Piece of code is perfectly working in my friends system.

raise PrefectHTTPStatusError.from_httpx_error(exc) from exc.__cause__
prefect.exceptions.PrefectHTTPStatusError: Client error '403 Forbidden' for url '<https://api.prefect.cloud/api/accounts/470d1ece-3f38-482c-8641-2b9756b4781e/workspaces/b7f3ff33-c0b3-4c1e-b1a3-794ef4f674b1/flows/>'
Response: {'detail': 'Forbidden'}
For more information check: <https://httpstatuses.com/403>

Hi everyone ! I’m using nested pydantic models as input parameters for my flows. My issue here is that I’m blocked by mandatory field in optional nested model: My input looks like :

class ModelB(pydantic.BaseModel):
    mandatory: string

class ModelA(pydantic.BaseModel):
    some_values: typing.Any
    b: ModelB | None = None

In some case, I don’t need to instantiate

ModelB

and to do so, I just don’t add it to the deployment’s yaml. This was working fine all along, but this morning the UI prevents me to run a deployment

Quick Run

as my field

ModelB.mandatory

is missing (even if I did not specify

ModelB

in

ModelA

in the yaml). Did something change and I have now to change my model parameters ?

Prefect 2.7.10 Hi everyone! I am experiencing very strange issue. My one of my flows consists of one task. Lets say

@task(persist_result=True)
def task():
    ...

@flow(persist_result=True)
def flow():
    task()

As you can see the task is called synchronously and I don’t get any results back. But I keep getting.

State message: Flow run encountered an exception. MissingResult: State data is missing. Typically, this occurs when result persistence is disabled and the state has been retrieved from the API.