Pass tuple of exceptions as task label value

[sector119]

Hello,

How can I pass tuple of exceptions as task label value?
Like @broker.task("billing:sync-payments", throws=(ZeroDivisionError,))
In my middleware throws = message.labels.get("throws", self.default_throws) returns str not the original exception

Thanks

[s3rius]

Hi, I see simplest solution here is to pickle the exception.

That's the script that I used to test my theory. It works, but with a small constratins.

import asyncio
import pickle
from typing import Any

from taskiq_redis import ListQueueBroker

from taskiq import TaskiqMiddleware
from taskiq.message import TaskiqMessage
from taskiq.result import TaskiqResult

broker = ListQueueBroker("redis://localhost:6379/0")


class TestMWare(TaskiqMiddleware):

    def _print(self, hook: str, message: TaskiqMessage) -> None:
        try:
            print(hook, pickle.loads(message.labels.get("throws"))[0])
        except:
            print(message.labels.get("throws"))
            print(hook, "No luck")

    def pre_execute(self, message: TaskiqMessage) -> TaskiqMessage:
        self._print("PREEX", message)
        return super().pre_execute(message)

    def post_execute(
        self,
        message: TaskiqMessage,
        result: TaskiqResult[Any],
    ) -> None:
        self._print("POSTEX", message)
        return super().post_execute(message, result)

    def pre_send(self, message: TaskiqMessage) -> TaskiqMessage:
        self._print("PRESEND", message)
        return super().pre_send(message)

    def post_send(self, message: TaskiqMessage) -> None:
        self._print("POSTSEND", message)
        return super().post_send(message)


broker.add_middlewares(TestMWare())


@broker.task(throws=pickle.dumps((ValueError,)))
async def meme():
    pass


async def main():
    await broker.startup()
    await meme.kiq()
    await broker.shutdown()


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

As you can see, I implemented all hooks to see what value it has in lables on each step. Before the message is sent it's serialized, and if you try send bytes in lables it detects it and encodes it in base64 value so it can be decoded back on the other side.

That's what you'll see when you'll try to run it. Client side:

Worker side:

As you can see, on the worker side the label was automatically decoded into an actual value, but on client side you see exact message that is going to be sent over the broker.

[s3rius]

Hope that this answer helped you.

[sector119]

Thank you, after looking at labels.py, I did it the same way ) Would it be possible to serialise ANY type with pickle like
return base64.b64encode(pickle.dumps(label_value)).decode(), LabelType.ANY.value and set LabelType.ANY to lambda x: pickle.loads(base64.b64decode(x)) ?

[s3rius]

It is possible, but it's really unsecure. It means that you can send actual code in labels.

[s3rius]

Maybe we can add an option that enables this behavior.