DataTypeError raised when listener's return type is not JSON

[roblight]

Firstly, awesome project! I appreciate it being developed and shared.

Tested with v0.7.1 and Python 3.10.4 on Ubuntu 22.04.1 LTS.

I believe the solution is to add data_type argument to self.publish_func() since the default value is "json" for the publish methods:

panini/panini/managers/nats_client.py

Line 394 in b60b5b2

await self.publish_func(reply_to, response)

Test code and output below:

import json
import traceback

from panini import app as panini_app

app = panini_app.App(
    service_name="quickstart-app",
    host="127.0.0.1",
    port=4222,
)

message = {
    "key4": "value1",
    "key7": 2,
    "key3": 3.024444412342342342,
    "key1": [1, 2, 3, 4],
    "key6": {"1": 1, "2": 2, "3": 3, "4": 4, "5": 5},
    "key5": {"subkey1": "1", "subkey2": 2, "3": 3, "4": 4, "5": 5},
    "key2": None,
}

@app.task()
async def publish_string():
    some_string = json.dumps(message, sort_keys=True)

    try:
        print(f"*** sending request for json")
        # default data_type is "json"
        response = await app.request(
            subject="some.publish.subject.json",
            message=message,
            timeout=5
        )
        print(f"response expecting json: type: {type(response)}")
    except BaseException as exc:
        print(f"****** got exception: {exc}")

    try:
        print(f"*** sending request for str")
        response = await app.request(
            subject="some.publish.subject.str",
            message=some_string,
            data_type=str,
            timeout=5
        )
        print(f"response expecting str: type: {type(response)}")
    except BaseException as exc:
        print(f"****** got exception: {exc}")

    try:
        print(f"*** sending request for bytes")
        response = await app.request(
            subject="some.publish.subject.bytes",
            message=some_string.encode(),
            data_type=bytes,
            timeout=5
        )
        print(f"response expecting bytes: type: {type(response)}")
    except BaseException as exc:
        print(f"****** got exception: {exc}")

# default data_type is "json"
@app.listen("some.publish.subject.json")
async def receive_dict(msg):
    print(f"request expecting json: type: {type(msg.data)}")
    return {"some": "response"}

@app.listen("some.publish.subject.str", data_type=str)
async def receive_string(msg):
    print(f"request expecting str: type: {type(msg.data)}")
    return '{"some": "response"}'

@app.listen("some.publish.subject.bytes", data_type=bytes)
async def receive_bytes(msg):
    print(f"request expecting bytes: type: {type(msg.data)}")
    return b'{"some": "response"}'

if __name__ == "__main__":
    app.start()

*** sending request for json
request expecting json: type: <class 'dict'>
response expecting json: type: <class 'dict'>
*** sending request for str
request expecting str: type: <class 'str'>
****** got exception: nats: timeout
*** sending request for bytes
request expecting bytes: type: <class 'bytes'>
****** got exception: nats: timeout
^C
...
panini.exceptions.DataTypeError: Expected dict or list but got <class 'str'>
...
panini.exceptions.DataTypeError: Expected dict or list but got <class 'bytes'>

[artas728]

Hello Rob
Thank you for your feedback!
We will fix the data type problem with the next release

[roblight]

Hello Rob Thank you for your feedback! We will fix the data type problem with the next release

Do we have a next release date in mind? Thank you!

[artas728]

Oh, sorry @roblight I missed your last message.
I'm going to make an alpha release in a few hours. There will be significant changes with data_types/serialization logic. I will write here when it is ready

[artas728]

@roblight, Panini v0.8.0a1 available via pip. This will still take a few weeks before the release of v0.8.0. In the meantime, I suggest you try to solve your problem with v0.8.0a1.

I think the solution is a bit more global than you would expect. So, update with the "data_type" parameter that is associated with this issue:

1. Parameter data_type "json" has been removed. Actually, "json" here meant jsonble python object. Perhaps it was not the most explicit data type name. I believe it was confusing many developers, we decided to remove it from v0.8.0.
2. For "publish" and "request" methods no need to declare "data_type" anymore, Panini detects a type of your message by itself.
3. New parameter for “request” method - “response_data_type”. For example:


subject = "a.b.c"
message = {"param1": "value1"}


response: bytes = app.request(subject=subject, message=message, response_data_type=bytes)


Let's update your code according to the above:

import json
import traceback

from panini import app as panini_app

app = panini_app.App(
    service_name="quickstart-app",
    host="127.0.0.1",
    port=4222,
)

message = {
    "key4": "value1",
    "key7": 2,
    "key3": 3.024444412342342342,
    "key1": [1, 2, 3, 4],
    "key6": {"1": 1, "2": 2, "3": 3, "4": 4, "5": 5},
    "key5": {"subkey1": "1", "subkey2": 2, "3": 3, "4": 4, "5": 5},
    "key2": None,
}

@app.task()
async def publish_string():
    some_string = json.dumps(message, sort_keys=True)

    try:
        print(f"*** sending request for json")
        # default data_type is "json"
        response = await app.request(
            subject="some.publish.subject.json",
            message=message,
            timeout=5
        )
        print(f"response expecting json: type: {type(response)} (it should be dict)")
    except BaseException as exc:
        print(f"****** got exception: {exc}")

    try:
        print(f"*** sending request for str")
        response = await app.request(
            subject="some.publish.subject.str",
            message=some_string,
            response_data_type=str,
            timeout=5
        )
        print(f"response expecting str: type: {type(response)}")
    except BaseException as exc:
        print(f"****** got exception: {exc}")

    try:
        print(f"*** sending request for bytes")
        response = await app.request(
            subject="some.publish.subject.bytes",
            message=some_string.encode(),
            response_data_type=bytes,
            timeout=5
        )
        print(f"response expecting bytes: type: {type(response)}")
    except BaseException as exc:
        print(f"****** got exception: {exc}")

# default data_type is "json"
@app.listen("some.publish.subject.json")
async def receive_dict(msg):
    print(f"request expecting json: type: {type(msg.data)} (it should be dict)")
    return {"some": "response"}

@app.listen("some.publish.subject.str", data_type=str)
async def receive_string(msg):
    print(f"request expecting str: type: {type(msg.data)}")
    return '{"some": "response"}'

@app.listen("some.publish.subject.bytes", data_type=bytes)
async def receive_bytes(msg):
    print(f"request expecting bytes: type: {type(msg.data)}")
    return b'{"some": "response"}'

if __name__ == "__main__":
    app.start()

*** sending request for json
request expecting json: type: <class 'dict'> (it should be dict)
response expecting json: type: <class 'dict'> (it should be dict)
*** sending request for str
request expecting str: type: <class 'str'>
response expecting str: type: <class 'str'>
*** sending request for bytes
request expecting bytes: type: <class 'bytes'>
response expecting bytes: type: <class 'bytes'>

Let me know if you have any questions about it.

[artas728]

Also, other updates that are not directly related to the issue, but may be useful for solving the problem with "data_type":

1. Panini supports dataclass as data_type now. You may use it to serialize or validate your messages. I’ve tested it mostly with pydantic dataclasses and also with mashumaro. Example of usage is here: https://github.com/lwinterface/panini/blob/develop/examples/simple_examples/dataclass_msg.py

2. Panini supports any Callable object as data_type for custom processing. An example of usage is here:

from panini.exceptions import MessageSchemaError
from panini import app as panini_app

app = panini_app.App(
    service_name="test_serializer_callable",
    host="127.0.0.1",
    port=4222,
)


def callable_validator(**message):
    if type(message) is not dict:
        raise MessageSchemaError("type(data) is not dict")
    if "data" not in message:
        raise MessageSchemaError("'data' not in message")
    if type(message["data"]) is not int:
        raise MessageSchemaError("type(message['data']) is not int")
    if message["data"] < 0:
        raise MessageSchemaError(f"Value of field 'data' is {message['data']} that negative")
    message["data"] += 1
    return message


@app.listen("test_validator.foo", data_type=callable_validator)
async def publish(msg):
    return {"success": True}


@app.listen("test_validator.foo-with-error-cb", data_type=callable_validator)
async def publish(msg):
    return {"success": True}


@app.listen("test_validator.check")
async def check(msg):
    try:
        message = callable_validator(**msg.data)
    except MessageSchemaError:
        return {"success": False}

    return {"success": True}


if __name__ == "__main__":
    app.start()

3. Panini validator has been removed. If you need incoming message validation, we recommend to use dataclasses. Example of usage for Pydantic: https://pydantic-docs.helpmanual.io/usage/validators/

[roblight]

Hi @artas728 thank you so much! I plan to review the changes and give feedback ASAP.