No longer able to register methods in TaskIQ 0.11.16

[TheTechromancer]

After updating to 0.11.16, registering a method as a task produces an AttributeError:

bbot_server/applets/_base.py:234: in register_watchdog_tasks
    task = broker.register_task(method, **kwargs)
../../../.cache/pypoetry/virtualenvs/bbot-server-DmHQwTaK-py3.13/lib/python3.13/site-packages/taskiq/abc/broker.py:377: in register_task
    return self.task(task_name=task_name, **labels)(func)
../../../.cache/pypoetry/virtualenvs/bbot-server-DmHQwTaK-py3.13/lib/python3.13/site-packages/taskiq/abc/broker.py:331: in inner
    self.decorator_class(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = AsyncTaskiqDecoratedTask(bbot_server.applets.events:archive_events_task), broker = <taskiq_nats.broker.NatsBroker object at 0x70a16466cec0>
task_name = 'bbot_server.applets.events:archive_events_task'
original_func = <bound method EventsApplet.archive_events_task of <bbot_server.applets.events.EventsApplet object at 0x70a16609c050>>, labels = {'schedule': [{'cron': '0 0 * * *'}]}

    def __init__(
        self,
        broker: "AsyncBroker",
        task_name: str,
        original_func: Callable[_FuncParams, _ReturnType],
        labels: Dict[str, Any],
    ) -> None:
        self.broker = broker
        self.task_name = task_name
        self.original_func = original_func
        self.labels = labels
    
        # This is a hack to make ProcessPoolExecutor work
        # with decorated functions.
        #
        # The problem is that when we decorate a function
        # it becomes a new class. This class has the same
        # name as the original function.
        #
        # When receiver sends original function to another
        # process, it will have the same name as the decorated
        # class. This will cause an error, because ProcessPoolExecutor
        # uses `__name__` and `__qualname__` attributes to
        # import functions from other processes and then it verifies
        # that the function is the same as the original one.
        #
        # This hack renames the original function and injects
        # it back to the module where it was defined.
        # This way ProcessPoolExecutor will be able to import
        # the function by it's name and verify its correctness.
        new_name = f"{original_func.__name__}__taskiq_original"
>       self.original_func.__name__ = new_name
E       AttributeError: 'method' object has no attribute '__name__' and no __dict__ for setting new attributes. Did you mean: '__ne__'?

../../../.cache/pypoetry/virtualenvs/bbot-server-DmHQwTaK-py3.13/lib/python3.13/site-packages/taskiq/decor.py:79: AttributeError

[s3rius]

Can you pelase show how you declare tasks?

[s3rius]

It seems like you mark your class methods as tasks. I tried this and it worked.

import asyncio
import time

from taskiq_redis import RedisStreamBroker

broker = RedisStreamBroker("redis://localhost")


class TaskClass:
    class Inner:
        @broker.task()
        @staticmethod
        def extensive_cpu_task(n):
            time.sleep(2)
            s = sum(i for i in range(n))
            print("Done")
            return s


async def main():
    await broker.startup()

    task = await TaskClass.Inner.extensive_cpu_task.kiq(1000000)
    await task.wait_result()

    await broker.shutdown()


if __name__ == "__main__":
    asyncio.run(main())

I wonder how do you declare tasks?

[TheTechromancer]

Sure, here is the relevant code:

class BaseApplet:
    async def register_watchdog_tasks(self, broker):
        # register watchdog tasks
        methods = {name: member for name, member in inspect.getmembers(self) if callable(member)}
        for method_name, method in methods.items():
            # handle case where tasks have already been registered
            method = getattr(method, "original_func", method)

            _watchdog_task = getattr(method, "_watchdog_task", None)
            if _watchdog_task is None:
                continue
            kwargs = getattr(method, "_kwargs", {})
            # crontab handling
            cron_default = kwargs.pop("cron", None)
            cron_config_key = kwargs.pop("cron_config_key", None)
            if cron_config_key is not None:
                if cron_default is None:
                    raise ValueError(
                        f"{self.name}.{method_name}: When specifying a crontab config value, you must also give a default crontab value (kwarg: 'cron')"
                    )
                cron = OmegaConf.select(self.config, cron_config_key, default=cron_default)
                kwargs["schedule"] = [{"cron": cron}]
            elif cron_default is not None:
                kwargs["schedule"] = [{"cron": cron_default}]
            self.log.debug(f"Registering task: {method_name} {kwargs}")
            task = broker.register_task(method, **kwargs)
            # overwrite the original method with the decorated TaskIQ task
            setattr(self, method_name, task)

Basically, we are iterating over our class, and registering certain methods with the TaskIQ broker.

[s3rius]

Can you please show me how you use it? Because it seems like something is purging out __name__ attribute of methods.

[s3rius]

I tried this snippet and seems like all methods have __name__ by default.

    for name, met in inspect.getmembers(TaskClass):
        if not callable(met):
            continue
        print(name)
        print(met.__name__)

Which led me to an idea that something is removing __name__ attribute of a function. Also, what got my attention is message that some method doesn't have __name__. But by default methods are typed as functions.

[TheTechromancer]

That is really strange. I'm not sure how they're losing their __name__ attribute. Yeah wtf. I'll look into this some more tomorrow.

[TheTechromancer]

Okay, I tried printing method.__name__ before it's passed into broker.register_task(), and at least at that point, it's still accessible. It seems to lose it somewhere between broker.register_task() and self.decorator_class().

[lindycoder]

Seems related, introduced in 0.11.15: I'm not able to add a callable object as task anymore

import asyncio
from dataclasses import dataclass

from taskiq_redis import RedisStreamBroker

broker = RedisStreamBroker("redis://localhost")

@dataclass(frozen=True, kw_only=True)
class MyTask:
    async def __call__(self, *args, **kwargs):
        await asyncio.sleep(2)


async def main():
    broker.task(task_name="somthing")(MyTask())

if __name__ == "__main__":
    asyncio.run(main())

This works with 0.11.14
starting 0.11.15, produces the same error AttributeError: 'MyTask' object has no attribute '__name__'. Did you mean: '__ne__'?

[s3rius]

I don't think it should have worked in the first place. The tasks were intended to be functions, and currently, classes and objects are not officially supported.

We can add support for them, but it requires some considerations, such as how to handle self. Should we be able to mark methods as tasks?

Also, it's unclear to me how this code works because, on the worker side, the task isn't declared.

[s3rius]

@lindycoder, can you please share with me MRE of the codebase that works on previous version and doesn't work since the 0.11.16.

Because the code you sent doesn't work on verison 0.11.15.

Here's what I got.

[2025-04-01 20:33:21,189][taskiq.receiver.receiver][WARNING][worker-0] task "somthing" is not found. Maybe you forgot to import it?

[scott-boost]

@s3rius maybe you misread

he wrote his code snippet should work upto and including version 0.11.14

[s3rius]

@scott-boost, yeah, thanks for noticing. My bad. I was running it on 0.11.14. It doesn't work on 0.11.15 because of __name__ and doesn't work on 0.11.14 because tasks are not available to the worker.

[lindycoder]

@s3rius This is all part of a bigger framework we have, I just wanted to share that we currently have a use case where we pass in a "callable", even though it's not a function. I didn't expect my example to work at all, just to not raise.

[s3rius]

@lindycoder, okay, clear point. Can you describe the framework you have a little bit so I can re-create something similar to test out my solutions?