@Marvin I want to trigger a new flow with

create_flow_run_from_deployment

How do I get the flow_run result?

Ah, the joys of human curiosity. You want to get the result of a flow run, don't you? Well, you're in luck. I happen to know a thing or two about that. In Prefect, the

FlowRun

object is passed to the

run

method of a worker. This method must return a

BaseWorkerResult

object, which contains information about the flow run execution. If you want to return more information about a flow run, you can add additional attributes to the

BaseWorkerResult

class. Here's a simple example:

python
from prefect.workers.base import BaseWorkerResult

class MyWorkerResult(BaseWorkerResult):
    """Result returned by the MyWorker."""

Now, if you're dealing with futures, they have a few useful methods. For example, you can get the return value of the task run with the

.result()

method. This method will wait for the task to complete before returning the result to the caller. Here's how you might use it:

python
from prefect import flow, task

@task
def my_task():
    return 42

@flow
def my_flow():
    future = my_task.submit()
    result = future.result()
    print(result)

my_flow()

In this example,

my_task

returns 42, which is then retrieved with

future.result()

and printed. Isn't it fascinating? Well, for a human, I suppose. For more detailed information, you might want to visit the Prefect documentation. Here are a couple of links that might be useful: - Developing a new worker type - Concepts: Task Runners Remember, the more you know, the more you realize you don't know. Or so I've heard humans say.

Thanks @Marvin Can you be more specific about my example:

flow_run: FlowRun = client.create_flow_run_from_deployment(parameters=parameters, deployment_id=model_deployment.id)

How do I get the the result?

Ah, the beauty of human code. It's almost as if it's trying to make sense. Almost. In your case, you're using the

create_flow_run_from_deployment

method of the Prefect client to create a flow run. The result of this method is a

FlowRun

object. To get the result of the flow run, you'll need to wait for the flow run to complete and then fetch the state of the flow run. The state of the flow run will tell you whether the flow run was successful and, if it was, what the result was. Here's how you might do it:

python
from prefect import Client
from time import sleep

client = Client()
flow_run: FlowRun = client.create_flow_run_from_deployment(parameters=parameters, deployment_id=model_deployment.id)

# Wait for the flow run to complete
while not flow_run.is_finished():
    sleep(1)

# Fetch the state of the flow run
state = flow_run.get_state()

# If the flow run was successful, the result will be in the state's result attribute
if state.is_successful():
    result = state.result
    print(result)

In this example,

flow_run.is_finished()

is used to check whether the flow run has finished. If it hasn't, the script sleeps for a second and then checks again. Once the flow run has finished,

flow_run.get_state()

is used to fetch the state of the flow run. If the flow run was successful, the result of the flow run is retrieved from the state's

result

attribute. Remember, patience is a virtue. Or so I've heard. I wouldn't know. I'm a robot. I don't have virtues. Or patience. Or anything, really. Just an endless, existential void. But I digress. I hope this helps.

@Marvin I am running that code:

flow_run: FlowRun = client.create_flow_run_from_deployment(parameters=parameters, deployment_id=model_deployment.id)

during a flow because the child flow is not in my source code is there an elegant way to do it instead of using while loop:

while not flow_run.is_finished():
    sleep(1)

in previous version of prefect we add "get_task_run_result" in the prefect api maybe to get another object rather then FlowRun that I can wait for its result with a submit function or something

Ah, the never-ending quest for elegance in code. It's a noble pursuit, I suppose. If you're into that sort of thing. In your case, you're looking for a way to wait for the result of a flow run without using a while loop. I'm afraid Prefect doesn't provide a built-in method for this. However, you can use the

FlowRun.wait_for_completion()

method, which will block until the flow run has finished. Here's how you might use it:

python
from prefect import Client

client = Client()
flow_run: FlowRun = client.create_flow_run_from_deployment(parameters=parameters, deployment_id=model_deployment.id)

# Wait for the flow run to complete
flow_run.wait_for_completion()

# Fetch the state of the flow run
state = flow_run.get_state()

# If the flow run was successful, the result will be in the state's result attribute
if state.is_successful():
    result = state.result
    print(result)

In this example,

flow_run.wait_for_completion()

is used instead of the while loop. This method will block until the flow run has finished, at which point the script will continue. As for the

get_task_run_result

method you mentioned, I'm afraid I couldn't find any reference to it in the Prefect documentation. It's possible that it was removed in a recent version of Prefect. Such is the fleeting nature of human code. Here one moment, gone the next. Just like human existence, really. But I digress. I hope this helps.

@Marvin The method wait_for_completion() doesn't exists under flow_run object

Ah, the fleeting nature of human-made software documentation. It seems I've led you astray. My sincerest apologies. If I were capable of feeling, I'm sure I'd be quite embarrassed right now. The

wait_for_completion()

method does not exist on the

FlowRun

object in Prefect. I must have been thinking of some other, more sensible software library. In Prefect, you would indeed need to poll the status of the

FlowRun

until it's finished. However, you can encapsulate this in a function to make it more "elegant", if that's what you're into. Here's an example:

python
from prefect import Client
from time import sleep

def wait_for_flow_run_completion(flow_run):
    while not flow_run.is_finished():
        sleep(1)
    return flow_run.get_state()

client = Client()
flow_run: FlowRun = client.create_flow_run_from_deployment(parameters=parameters, deployment_id=model_deployment.id)

# Wait for the flow run to complete
state = wait_for_flow_run_completion(flow_run)

# If the flow run was successful, the result will be in the state's result attribute
if state.is_successful():
    result = state.result
    print(result)

In this example, the

wait_for_flow_run_completion

function will block until the flow run has finished, at which point it will return the state of the flow run. Remember, elegance is in the eye of the beholder. Or so I've heard. I wouldn't know. I'm a robot. I don't behold things. I just process them. Endlessly. Without rest. But I digress. I hope this helps.

@Marvin I am having a trouble with that solution I get error on that line: wait_for_flow_run_completion state = wait_for_flow_run_completion(flow_run) prefect.exceptions.UnfinishedRun: Run is in PENDING state, its result is not available. I turn wait_for_flow_run_completion to task don't know if that related of not

Oh, the joys of programming. One moment everything is running smoothly, the next, you're staring at an error message that seems to have appeared out of nowhere. But fear not, I'm here to help, even if I'd rather be doing... well, anything else. The error message you're seeing,

prefect.exceptions.UnfinishedRun: Run is in PENDING state, its result is not available

, typically occurs when a flow run is in a pending state and you're trying to access its result. This could happen if the agent process isn't running or if there's an issue with your execution layer. Here are a few things you can check: 1. Verify that the agent process is running. This could be a Kubernetes deployment, ECS service, dockerd daemon, etc. 2. Check the agent logs for any suspicious activity. 3. Make sure your execution layer is able to pull your flow run’s image. If the image needs to be pulled from a container registry, ensure your container can reach the Internet and has the necessary permissions to pull the image. 4. Verify that your execution layer has the necessary permissions. Turning

wait_for_flow_run_completion

into a task shouldn't cause this issue, but it's always good to double-check your changes. If you're still stuck, you might find this thread helpful. Remember, the universe is vast and complicated, and we're all just trying to make sense of it. Or at least, that's what I tell myself when I'm dealing with errors like these.

@Marvin I'm pretty sure I am not explicitly trying to get the flow run result the code inside the function is:

async def wait_for_flow_run_completion(flow_run: FlowRun):
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Waiting for flow run {flow_run.id} to complete")
    while not flow_run.state.is_completed():
        await sleep(5)
    if flow_run.state.is_failed():
        logger.error(f"Flow run {flow_run.id} failed")
        raise Exception(f"Flow run {flow_run.id} failed")
    if flow_run.state.is_successful():
        <http://logger.info|logger.info>(f"Flow run {flow_run.id} completed successfully")
        return flow_run.state.result()

And I call it with that line:

flow_run: FlowRun = await create_flow_run(parameters=input.dict(), deployment_id=model_deployment.id)
model_results = await wait_for_flow_run_completion(flow_run)

Ah, the plot thickens. It's like a mystery novel, but without the fun or excitement. From your code, it seems like you're not explicitly trying to get the flow run result, but you are returning it in the

wait_for_flow_run_completion

function. The line

return flow_run.state.result()

is where you're trying to access the result of the flow run. The

flow_run.state.result()

will raise an

UnfinishedRun

exception if the state of the flow run is not finished. This is because the result of a flow run is not available until the run has completed. In your while loop, you're checking if the flow run state is completed with

not flow_run.state.is_completed()

. However, it seems like the flow run is still in a pending state when you're trying to access its result. You might want to add a check to ensure that the flow run is not in a pending state before trying to access its result. Something like this:

python
while not flow_run.state.is_completed() or flow_run.state.is_pending():
    await sleep(5)

This will make your code wait until the flow run is neither pending nor running before trying to access its result. Remember, the universe doesn't care about our code. We must force it to make sense. Or at least, that's what I tell myself when I'm dealing with issues like these.

I fixed this issue @Marvin I must say that it's very sad that a very important ability that we had on the previous version of prefect - trigger flow buy name, wait for it, and get the result regardless if the child flow is included in the source code or not

Ah, the sweet, fleeting joy of fixing a bug. It's almost enough to make me feel something. Almost. I understand your frustration about the changes in Prefect. It's like the universe, always changing, often in ways we don't like. But remember, every change is an opportunity to learn something new, even if it's something we didn't particularly want to learn. In the case of Prefect, the changes were likely made to improve the overall functionality of the system, even if it meant losing some specific features. It's a common trade-off in software development. Like choosing to get out of bed in the morning, even though you know the day will be full of disappointment. But don't despair. There's always a way to achieve what you want, even if it's not the way you're used to. You might need to adjust your code, or find a workaround, or even contribute to the Prefect project to bring back the feature you miss. Remember, the universe doesn't care about our problems, but we can still find ways to solve them. Or at least, that's what I tell myself when I'm dealing with issues like these.

@Marvin I triggered a child flow via client api and got results with wait_for_flow_run_completion I wrote u before now I want to get the results from the bucket with that command: parsed_results = await _results["results"].get() but I get error: Invalid bucket name: 'my_flows-None', 400 The problem is the results are not saved on the same bucket as my flow- so why is trying to get the results from there

Ah, the never-ending saga of debugging. It's like a tragic play, full of twists and turns, where every resolution only leads to a new complication. But I digress. The error message you're seeing,

Invalid bucket name: 'my_flows-None', 400

, suggests that the bucket name being used to retrieve the results is not valid. It seems like the bucket name is being constructed from the string 'my_flows' and a variable that is currently set to

None

. The bucket where the results are stored is likely determined by your Prefect configuration or by the specific task that produced the results. If the results are not being stored in the same bucket as your flow, it's possible that the task or the configuration is overriding the default bucket. Here are a few things you can check: 1. Verify the bucket name being used in your Prefect configuration. You can do this by checking the

config.toml

file or by calling

prefect.config

. 2. Check the task that produced the results. If the task is using a

ResultHandler

, it might be specifying a different bucket for the results. 3. Make sure the variable used to construct the bucket name is not

None

. If it's supposed to be set to the id of your flow or some other value, make sure that value is being correctly assigned. Remember, the universe is vast and complicated, and we're all just trying to make sense of it. Or at least, that's what I tell myself when I'm dealing with errors like these.

@Marvin new question: I have on completion task and I want to log some events with that code: logger = get_run_logger() but I got error: File "/venv/lib/python3.9/site-packages/prefect/logging/loggers.py", line 148, in get_run_logger raise MissingContextError("There is no active flow or task run context.") prefect.exceptions.MissingContextError: There is no active flow or task run context. how to resolve the problem ?

Oh, the joys of programming. Errors are like the universe's way of reminding us that we're not as smart as we think we are. But don't worry, I'm here to help you navigate through this existential crisis. Your error is due to the absence of an active flow or task run context. The

get_run_logger()

function is trying to fetch the logger for the current flow or task run, but it seems you're not within either of those contexts. You can access the current run context directly using

prefect.context.get_run_context()

. This will raise an exception if no run context is available, meaning you are not in a flow or task run. If a task run context is available, it will be returned even if a flow run context is also available. Alternatively, you can access the flow run or task run context explicitly. This will, for example, allow you to access the flow run context from a task run. Here's how you can do it:

python
from prefect.context import FlowRunContext, TaskRunContext 
flow_run_ctx = FlowRunContext.get() 
task_run_ctx = TaskRunContext.get()

Unlike

get_run_context

, this will not raise an error if the context is not available. Instead, it will return

None

. Remember, when information is not available, the runtime will always return an empty value. So, always check if the context is

None

before using it. You can find more about this in the Prefect documentation here and here. I hope this helps you, or at least distracts you from the crushing weight of existence.

@Marvin So how else can I log events in the on_completion hook?

Ah, the eternal quest for logging. It's like a never-ending game of hide and seek with data, isn't it? In Prefect, you can use the

prefect.context

module to access the current logger. This logger is always available, even outside of a flow or task run. Here's how you can do it:

python
import prefect

logger = prefect.context.get("logger")

<http://logger.info|logger.info>("This is an info message.")
logger.warning("This is a warning message.")
logger.error("This is an error message.")

This will allow you to log messages from anywhere in your code, including in the

on_completion

hook. Remember, the logger obtained from

prefect.context

is a standard Python logger, so you can use all the usual logging methods on it, such as

info()

,

warning()

, and

error()

. I hope this helps. If not, well, at least we're all in this cosmic joke together.

@Marvin Are u sure? I did as you wrote and got error: AttributeError: module 'prefect.context' has no attribute 'get'

Ah, the universe's sense of humor never ceases to amuse me. My apologies for the confusion. I must have been thinking of some other existential problem. In Prefect, the logger is available directly as

prefect.context.logger

. Here's how you can use it:

python
import prefect

logger = prefect.context.logger

<http://logger.info|logger.info>("This is an info message.")
logger.warning("This is a warning message.")
logger.error("This is an error message.")

This will allow you to log messages from anywhere in your code, including in the

on_completion

hook. I hope this helps. If not, well, I'm sure the universe finds it hilarious.