Hi @Marvin I want to run a flow from another flow, I am able to do it But the catch is, my main flow should not wait until the child flow finishes, it needs to show the status as completed/ success/fail . After it triggers the child flow. I can't use automations option as I need to trigger child only in few conditions of the parameters passed from deployment. How can this be achieved any guidance please

Hello. I'm trying to pass an aiohttp client session to a flow/task. Following an error in the functional block of a task it will ping/wait for an internet connection and then raise the error for the flow or task decorator to handle. I noticed that the session can't be reused seemingly because the event loop changes for each flow or task run? Hard for me to copy paste the stack from my jupyter notebook because of how it prints, it's Runtimeerror: Event loop is closed. The first time _client is assigned in Session() it works because the context is probably in the current event loop.

def Session():
    global _client
    if _client is None:
        _client=aiohttp.ClientSession(connector=aiohttp.TCPConnector(limit=100,limit_per_host=10,force_close=True,enable_cleanup_closed=True),timeout=aiohttp.ClientTimeout(total=30))
    return _client

@Marvin how do I create my own Custom Pickle Serializer?

Hi, I have a very newbie question related to objects returned by a task when composing a Flow. I have a couple of procedures that I want to execute and they return objects. I would like to pass to the subsequent procedure a parameter based on a property which is a member of the object returned by the first procedure. Something like the code below:

import random
from prefect import task, Flow


class Generator:
    def __init__(self, data):
        self.particle = data


@task
def run_generator():
    return Generator(random.random())


@task
def compute(particle):
    print(particle**2)


with Flow('flow1') as flow:
    gen = run_generator()
    compute(gen.particle)

When I try to create the flow I get the following error:

---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
/var/folders/7g/2_wk8ddx7h5_4r12flp9031m0000gn/T/ipykernel_3073/2988601640.py in <module>
      1 with Flow('flow1') as flow:
      2     gen = run_generator()
----> 3     val = compute(gen.particle)

AttributeError: 'FunctionTask' object has no attribute 'particle'

Is there any chance I can have the delayed execution of the parameter until the flow is executed? I know that changing

compute

to receive the object and handling the data access there works but not always I have the possibility to change the function and it would be bad in my case to create many wrappers for users of my library.

@Marvin how do I set the number of concurrent flows for a deployment? if I don't specify one, is the default 1?

@Marvin is it possible to access the results of a flow run from its state via a state transition hook without persisting the result? Something like this? @flow(validate_parameters=False, name=“MyFlow”, on_failure=[store_results], # this is a hook async def store_results(flow:Flow, flow_run: FlowRun, state: State[Any]) -> None: result = state.result() print(state.result()) The flow run is not persisted in this case, and the result can be quite a large object, so it would be nice to have access to the result from the hook

Hi all! I’m having a RAM issue when running one of my flows (diagram in thread). Essentially, I have a list of calls I need to make to a source system that I store in a list and the flow is for each item in that list

ping source system for dataset -> send dataset to pandas DF -> store dataset as a file in S3 -> push contents of said file to our datastore

I’m using the

LocalDaskExecutor

to do DFE and that’s helped a ton as the flow used to fail at the first step because it was doing Breadth first and holding all of the datasets in memory. I expected that once the task tree was completed for each dataset that it’d release that memory but that doesn’t seem to be the case. Am I doing something wrong/poorly or is there a way to free up that memory once the terminal task has completed for each branch?

@Marvin can I execute my flow with different parameters in parallel using multiprocessing? Will it share task runners? If so, can you show an example?

@Marvin I'm trying to implement UV in my flows. How do I add a custom entrypoint to a prefect.yaml file? It should be something like "uv run main.py"

Hey I was using a Prefect2's sdk func

build_from_flow

to deploy a flow by passing

entrypoint

and `path`:

build_deployment = await Deployment.build_from_flow(
        flow=hello_world_flow,
        name=name, 
        ...
        entrypoint=entrypoint,
        path=path)

Im starting now on 3.0 but now build_from_flow is missing:

prefect.exceptions.PrefectImportError: `prefect.deployments:Deployment` has been removed. Use `flow.serve()`, `flow.deploy()`, or `prefect deploy` instead.

Docs mention this args but it is unclear with no examples. Does someone know how to pass entrypoint & path when creating deployment from a flow?