taskiq-python / taskiq-pipelines Goto Github PK

Hey! Thank you for the library.

For some reason, I would use CBOR serializer instead of JSON.

The system could be configured easily enough by overloading Pipeline and AbstractStep dumps() and loads() methods. It works pretty well at least with aio-pika broker thanks to the fact that pika headers could be not only strings but also bytes.

The solution looks a little cumbersome though, it requires overloading more methods than it should be.

As a proper and generic resolution of the problem, I could imagine adding a new abstraction: AsyncBroker.serializer. It should be an instance of the following class:

class TaskiqSerializer(ABC):
    @abstractmethod
    def dumps(self, value: Any) -> bytes: ...
        @abstractmethod
    def loads(self, value: bytes) -> Any: ...
    

The difference between TaskiqSerializer and TaskiqFormatter is that TaskiqFormatter works with TaskiqMessage only but serializer should operate with any primitive types that are supported by underlying serialization system (things like decimals, datetimes, etc). Serializer supersedes formatter ABC and could be used instead with deprecation of TaskiqFormatter class and AsyncBroker.formatter attribute.

Another subtle problem is the type of TaskiqMessage.headers and BrokerMessage.headers. I don't think it is the showstopper; AsyncTaskiqDecoratedTask.labels is Dict[str, Any] already which is the correct type I believe.

What do you think? If you agree with my thoughts I can try to create PR for taskiq itself first and modify taskiq-pipelines later.

Often we need to do different sets of actions based on the result of the execution of some function.

Here we can discuss the API and how it can be implemented.

I'm not really sure how to test it, but we have to.