severb / graypy

Python logging handler for Graylog that sends messages in GELF (Graylog Extended Log Format).
https://www.graylog.org/
BSD 3-Clause "New" or "Revised" License
258 stars 90 forks source link

BaseGELFHandler._sanitize_to_unicode and NamedTuple #141

Open gjdanis opened 1 year ago

gjdanis commented 1 year ago

It looks like there might be an issue with the implementation of _sanitize_to_unicode for handling NamedTuple:

Python 3.8.10 (default, Feb 10 2022, 14:12:17)
[GCC 4.8.5 20150623 (Red Hat 4.8.5-39)] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> from graypy.handler import BaseGELFHandler
>>> from typing import NamedTuple
>>> class Test(NamedTuple):
...     a: int
...     b: str
...
>>> t = Test(a=1, b="test")
>>> print(BaseGELFHandler._sanitize_to_unicode(t))
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "python3.8/site-packages/graypy/handler.py", line 352, in _sanitize_to_unicode
    return obj.__class__([cls._sanitize_to_unicode(i) for i in obj])
TypeError: __new__() missing 1 required positional argument: 'b'

Checking if something is a NamedTuple isn't super straightforward but can be done like this.

The _asdict() function can then be used to reuse _sanitize_to_unicode with the object as a dictionary.

Here's an example of how this might be handled:

from typing import NamedTuple

data = bytes

def isinstance_namedtuple(obj) -> bool:
    return (
        isinstance(obj, tuple) and
        hasattr(obj, '_asdict') and
        hasattr(obj, '_fields')
    )

def _sanitize_to_unicode(obj):
    if isinstance(obj, dict):
        return dict(
            (_sanitize_to_unicode(k), _sanitize_to_unicode(v))
            for k, v in obj.items()
        )

    ## NEW CODE BEGIN
    if isinstance_namedtuple(obj):
        return _sanitize_to_unicode(obj._asdict())
    ## NEW CODE END

    if isinstance(obj, (list, tuple)):
        return obj.__class__([_sanitize_to_unicode(i) for i in obj])
    if isinstance(obj, data):
        obj = obj.decode("utf-8", errors="replace")
    return obj

class Test(NamedTuple):
    a: int
    b: str

t = Test(a=1, b="test")
print(_sanitize_to_unicode(t))