Hi, is there a way to delete a specific flow-run? or the logs associated with a specific flow-run? (for the case where logs may have non-redacted private data in, for example)

Hi. I have a mapped task that pulls data from an API and passes it to a reduce task that saves the data to S3. Some of the API calls may fail and that is OK. I am using a

FilterTask

as explained here to allow the reduce task to save the data. It works ok with small results, but when I try it with the real API output, I am getting

State payload is too large

I tried setting

result=None

for this task but that did not fix it. Any suggestions?

Failed to set task state with error: ClientError([{'path': ['set_task_run_states'], 'message': 'State payload is too large.', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}])
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/prefect/engine/cloud/task_runner.py", line 91, in call_runner_target_handlers
    state = self.client.set_task_run_state(
  File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 1917, in set_task_run_state
    result = self.graphql(
  File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 569, in graphql
    raise ClientError(result["errors"])
prefect.exceptions.ClientError: [{'path': ['set_task_run_states'], 'message': 'State payload is too large.', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}]

I also tried setting the trigger on the reduce task to

trigger=any_successful

and it worked but I wonder why the filter task is not working. Thanks!

hi there !!! nice to meet you all guys

I am trying to get the flow ID after register… is there any way ?

hello, is it possible to run dependent flows (flow of flows) like shown here https://docs.prefect.io/core/idioms/flow-to-flow.html without registering to the BE server. For example create the flow runs and call

from prefect import Flow
from prefect.schedules import CronSchedule
from prefect.tasks.prefect import create_flow_run, wait_for_flow_run


weekday_schedule = CronSchedule(
    "30 9 * * 1-5", start_date=pendulum.now(tz="US/Eastern")
)


with Flow("parent-flow", schedule=weekday_schedule) as flow:
    
    # assumes you have registered the following flows in a project named "examples"
    flow_a = create_flow_run(flow_name="A", project_name="examples")
    wait_for_flow_a = wait_for_flow_run(flow_a, raise_final_state=True)
    
    flow_b = create_flow_run(flow_name="B", project_name="examples")
    wait_for_flow_b = wait_for_flow_run(flow_b, raise_final_state=True)
    
    flow_c = create_flow_run(flow_name="C", project_name="examples")
    wait_for_flow_c = wait_for_flow_run(flow_c, raise_final_state=True)
    
    flow_d = create_flow_run(flow_name="D", project_name="examples")
    wait_for_flow_d = wait_for_flow_run(flow_d, raise_final_state=True)
    
    b = wait_for_flow_b(upstream_tasks=[wait_for_flow_a])
    c = wait_for_flow_c(upstream_tasks=[wait_for_flow_a])
    d = wait_for_flow_d(upstream_tasks=[b, c])
flow.run()

I'm trying to do a conditional flow like the one shown below:

@task
def check_cond():
    return False


@task
def run_task(anchor_date, days_from):

    logger = prefect.context.logger
    logger.critical(f"{days_from=}")


with Flow('foo') as flow:

    anchor_date = Parameter(name="anchor_date", default=None)
    cond = check_cond()

    with case(cond, True):
        anchor_date = run_task(anchor_date, days_from=0)

    with case(cond, False):
        anchor_date = run_task(anchor_date, days_from=-1)

however the

False

condition never gets run. When

check_cond

returns

True

- the flows runs as expected. Am I missing something?

Join us in 5 minutes for a FIRE~sale~side Chat with @Jeremiah and @Chris White! We'll be discussing 🔥 topics and, as promised, The Matrix 4 (I'm not joking).

Hi — when is the

today

template filled in a result

location

or task

target

string? I keep getting a `KeyError`:

from prefect.engine.results import LocalResult

# Create a new local result pointing to current working directory.
result = LocalResult(dir=".", location="{today}.txt")
assert result.location == "{today}.txt"

# This raises `KeyError: 'today'`.
result.write(value_="hello!")

I’m following this doc: https://docs.prefect.io/core/concepts/results.html#choose-a-serializer (under “Templating

Result

locations”)

Hi, I am experiencing some weird behavior using prefect server. I can run a flow on prefect core, it executes completely but when I spin up the UI and run the exact same flow on a local agent (same machine and environment), I get an ModuleNotFoundError. I have modules I import as part of my flow

Is there a way to reference a task output in a result

location

or task

target

? I know the Prefect context gives you a bunch of variables – notably including

parameters

, which looks promising! – but what if we have some data (the strings

"foo"

or

"bar"

) computed by some task

compute_foo_or_bar

and we’d like to reference that output in a subsequent task’s output target?

Hey all, my docker agent stopped being able to auth (error 500) to my Gitlab storage. I don't think I changed anything to the Prefect Agent/Gitlab settings so I don't know what caused it. Any ideas?

hi guys i am not sure I am using correctly prefect … anyone know where I can buy a course to learn more easy and deeply about prefect and best practices ?

Hi everyone, I am running prefect server on my linux and I want someone else in my team to be able view the UI on their own machine. Typically to do this, I simply share <IP>:<port_number> and this works like a charm (I do this with airflow and streamlit while testing). We are on the same network. I tried the same for prefect (<IP>:8080) but it can't be reached. I am still learning about prefect and not sure why this behavior is different. Would appreciate any help Update: I tried to start a local agent from a different machine and I get

ConnectionRefusedError: [WinError 10061] No connection could be made because the target machine actively refused it

. I can't tell if this is related. I created a config.toml file

# base configuration directory 
home_dir = "C:/Users/XXX/.prefect"

backend = "server"

[server]
host = "<http://XXXXXX>"
port = "4200"
host_port = "4200"
endpoint = "${server.host}:${server.port}"

I am certain the server is running on the host ip address

Hi all! how to kick-off flow defend on another flow is success? Thanks

Hi, is there any prefect graphql api video tutorial or blog posts.. Thanks!!!

Hey folks 😒imple_smile: We are using ECS Agent to execute flows. Is there a way to apply custom tags to the ECS task that is executed?

Hi all! Is anyone aware of a way of timing out a flow itself and not just the tasks? Ie Task has

timeout

which we can pass in, but Im currently experience some odd issues where tasks seem to be lost in dask somewhere (they are submitted to dask but never come back, never time out), and this means my flows never end. Ideally Ill try to dig into our env and dask and prefect and figure out what is causing silent untracked failure, but as an interim solution, does anyone know of a way I can say “Cancel the flows and all tasks if its been an hour since you started?”

Is anyone else having trouble logging into the UI? Status page is all green, but no one on my team can login

Yes

@jars @Brian Phillips and everyone else: We are aware of the issue and investigating it. We’ll update in the announcements channel once we know more.

All services are available again - we’ve updated the status page.

Thank you!

Hey everyone, what is the best way of finding out what labels a flow run was run with while executing inside the flow?

Hey everyone, I have a couple of flows that are running just fine and now I get this new error when trying to register a new flow -

raise InterruptedError(line.get("error"))
      InterruptedError: name invalid: Request contains an invalid argument.

In order to register my flow I use the following command -

prefect register --project "Project_name" --path path_to_flow_file.py --name "brands-recognition"

Project_name is the project I have at Prefect cloud path_to_flow_file is the path to the file which contains my flow brands-recognition is the flow name. This is my flow snippet (for debug purposes - just 2 tasks)-

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

with Flow("brands-recognition",
          storage=Docker(registry_url="us-central1-docker.pkg.dev/xxx/",
                         dockerfile="./Dockerfile"), schedule=schedule) as flow:  # , schedule=schedule
    mode = Parameter('FULL_UPDATE', default=None)
    task_a(mode)

flow.run_config = VertexRun(machine_type='e2-standard-8', labels=["ml"],
                            service_account='prefect-integration@xx')

# flow.run()

Now when running locally - the flow runs smoothly and generates a result. It seems like the problem is only at the pushing phase (when I register the flow). What am I missing? Thanks.

hi, is this behavior normal in the UI? I find it pretty frustrating that the

Run History

keeps re-materializing as I scroll

Hi every body !!! Let me share with you all an issue I have: First that is my code

This is the python that run the flow after …

Good morning, I have a question about the Prefect Cloud incident earlier today...

Hello, Is it possible to get a run history for all the flows using the GraphQL query rather than looking at each flow?