Add support for using callable objects as tasks

[jmg-duarte]

Fixes #7196

Example

The following example should now work:

from typing import Any

import prefect

class CallableObj:
    def __init__(self) -> None:
        self.x = 10

    def __call__(self, *args: Any, **kwds: Any) -> Any:
        print(self.x)


@prefect.flow
def test_flow():
    t = prefect.Task(fn=CallableObj())
    t()

Checklist

• This pull request references any related issue by including "closes <link to issue>"
  • If no issue exists and your change is not a small fix, please create an issue first.
• This pull request includes tests or only affects documentation.
• This pull request includes a label categorizing the change e.g. fix, feature, enhancement

[netlify]

✅ Deploy Preview for prefect-orion ready!

Built without sensitive environment variables

Name	Link
🔨 Latest commit	7ee264d
🔍 Latest deploy log	https://app.netlify.com/sites/prefect-orion/deploys/634ed419d4d54c00080827ee
😎 Deploy Preview	https://deploy-preview-7217--prefect-orion.netlify.app/api-ref/prefect/tasks
📱 Preview on mobile	Toggle QR Code... Use your smartphone camera to open QR code link.

---

To edit notification comments on pull requests, go to your Netlify site settings.

[zanieb]

Can we update this in tasks/flows.py to call to_qualified_name(type(obj)) instead? to_qualified_name should throw an error for instances because we will not be able to recreate it in from_qualified_name.

[jmg-duarte]

Hmmm, I don't think that is a good idea because it will break existing behavior (?)

In [1]: def f():
   ...:     pass
   ...: 

In [2]: type(f).__qualname__
Out[2]: 'function'

[zanieb]

Oh we'd still want to include some if logic, e.g. if not inpect.isfunction(obj): ...

[jmg-duarte]

Ah, moving the logic to Task.__init__, something like:

if not hasattr(obj, "__qualname__"):
    self.task_key = to_qualified_name(type(obj))
else:
    self.task_key = to_qualified_name(obj)

[zanieb]

Yep!

[zanieb]

@jmg-duarte The following test fails

    def test_task_run_from_callable_object(self):
        class Foo:
            message = "hello"

            def __call__(self, prefix: str, suffix: str) -> Any:
                return prefix + self.message + suffix

        obj = Foo()
        foo = Task(fn=obj)

        @flow
        def bar():
            return foo("a", suffix="b")

        assert bar() == "ahellob"

E AttributeError: 'Foo' object has no attribute '__name__'

I may have to revert this change. Do you have time to look at a fix?

[jmg-duarte]

@madkinsz as a quick fix: adding name=... to the Task call works.
As for an actual fix, I suggest using a similar strategy to the one for the task_key. If that's ok with you, I can submit a fix using that

[zanieb]

Ah I see I didn't even notice that it was in the __init__ block. Yeah can you remove the name= from your existing test and also include this test that it runs okay? I'm worried we'll inadvertently break something by accessing something that's not present at runtime as well.

[jmg-duarte]

also include this test that it runs okay?

Sorry, I don't understand what you mean by this and by extension, which path forward to take: revert or apply the suggestion I gave

[zanieb]

Sorry! Can you:

• Apply the suggestion you gave in a new pull request
• Remove the name=... from the test you added in this pull request so the fix is covered by the test suite
• Include the test I wrote here to ensure that the task can be used in a flow in creation