Based on what I heard, Prefect only works with functions and not methods.
Isn’t a method just a function where self
is the first argument?
When I create a decorator, it works for both functions and methods, so I am confused why it wouldn’t for @task or @flow.
import time
import types
# Decorator to calculate time taken by a method and store it in a dictionary
def timeit(method):
def wrapper(*args, **kwargs):
start_time = time.time()
result = method(*args, **kwargs)
end_time = time.time()
elapsed_time = end_time - start_time
method_name = method.__name__
print(f"Time taken by {method_name}: {elapsed_time}")
return result
return wrapper
@timeit
def test_function():
time.sleep(1)
return "hi"
print(test_function())
class TestObject():
@timeit
def test_method(self):
time.sleep(0.5)
return "hello"
a = TestObject()
print(a.test_method())
Time taken by test_function: 1.0047569274902344
hi
Time taken by test_method: 0.5026259422302246
hello
I just looked at the documentation for version 3.0.0 and this seems supported.
from prefect import task, flow
import time
class MyClass:
@task
def test_class_method(self): # (I should have used instance instead of class)
time.sleep(1)
return "hi"
@flow
def my_flow():
# Instantiate the class
my_instance = MyClass()
# Call the class methods decorated as tasks
result = my_instance.test_class_method()
print(f"Result: {result}")
my_flow()
01:49:44.549 | INFO | prefect.engine - Created flow run 'cooperative-tamarin' for flow 'my-flow'
01:49:44.584 | INFO | Task run 'test_class_method-351' - Created task run 'test_class_method-351' for task 'test_class_method'
01:49:45.623 | INFO | Task run 'test_class_method-351' - Finished in state Completed()
Result: hi
01:49:45.665 | INFO | Flow run 'cooperative-tamarin' - Finished in state Completed()
However, I get a typing error/warning:
No overloads for "__call__" match the provided arguments
Argument types: ()PylancereportCallIssue
(property) test_class_method: Task[(self: MyClass), str]
To remove this warning, I need to specify self my_instance.test_class_method(self=my_instance)
, but then I get a runtime error:
TypeError: Task.__call__() got multiple values for argument 'self'