https://prefect.io logo
#prefect-community
Title
# prefect-community
i

Ilhom Hayot o'g'li

06/08/2022, 2:20 PM
Hi all! I am trying to retreive time taken for @task and Flow( ) How to retreive/monitor start and end times of flow in the code. Is there any built in functions?
k

Kevin Kho

06/08/2022, 2:33 PM
No the UI calculates the duration on the fly (for Prefect 1.0) so you need to query the GraphQL API and then to
end_time
-
start_time
with some further processing
i

Ilhom Hayot o'g'li

06/08/2022, 2:41 PM
what I wanted to do was to create function and use it with tasks anotated @time . Is that recommended?
k

Kevin Kho

06/08/2022, 2:47 PM
@task
returns a class. Can you try switching the order of the decorators?
i

Ilhom Hayot o'g'li

06/08/2022, 2:53 PM
I have module not callable error? Is that about @Task?
k

Kevin Kho

06/08/2022, 3:22 PM
@Nate could you share your alternative to the double decorator here?
n

Nate

06/08/2022, 4:00 PM
Copy code
from prefect import Flow
from prefect.tasks.core.function import FunctionTask
from time import sleep, time
from typing import Any, Callable

class timed_task(FunctionTask):
    def __init__(
        self,
        fn: Callable = None,
        name: str = None,
        **kwargs: Any
    ):
        def timed_fn(**kwargs) -> Callable:
            t1 = time()
            result = fn(**kwargs)
            t2 = time()
            print(f'Function {fn.__name__!r} executed in {(t2-t1):.4f}s')
            return result
        
        super().__init__(fn=timed_fn, name=name, **kwargs)

@timed_task
def my_func() -> None:
    sleep(10)
 
with Flow('My Timed Flow') as flow:
    my_func()

if __name__ == "__main__":
    flow.run()
5 Views