Add FlatJSONFormatter

Author: marselester

json_log_formatter/__init__.py

@@ -204,3 +204,35 @@ def json_record(self, message, extra, record):
         extra['thread'] = record.thread
         extra['threadName'] = record.threadName
         return super(VerboseJSONFormatter, self).json_record(message, extra, record)
+
+
+class FlatJSONFormatter(JSONFormatter):
+    """Flat JSON log formatter ensures that complex objects are stored as strings.
+
+    Usage example::
+
+        logger.info('Sign up', extra={'request': WSGIRequest({
+            'PATH_INFO': 'bogus',
+            'REQUEST_METHOD': 'bogus',
+            'CONTENT_TYPE': 'text/html; charset=utf8',
+            'wsgi.input': BytesIO(b''),
+        })})
+
+    The log file will contain the following log record (inline)::
+
+        {
+            "message": "Sign up",
+            "time": "2024-10-01T00:59:29.332888+00:00",
+            "request": "<WSGIRequest: BOGUS '/bogus'>"
+        }
+
+    """
+
+    keep = (bool, int, float, complex, str, datetime)
+
+    def json_record(self, message, extra, record):
+        extra = super(FlatJSONFormatter, self).json_record(message, extra, record)
+        return {
+            k: v if v is None or isinstance(v, self.keep) else str(v)
+            for k, v in extra.items()
+        }

tests.py

@@ -17,7 +17,7 @@
 except ImportError:
     from io import StringIO
 
-from json_log_formatter import JSONFormatter, VerboseJSONFormatter
+from json_log_formatter import JSONFormatter, VerboseJSONFormatter, FlatJSONFormatter
 
 log_buffer = StringIO()
 json_handler = logging.StreamHandler(log_buffer)
@@ -336,3 +336,96 @@ def test_stack_info_is_none(self):
         logger.error('An error has occured')
         json_record = json.loads(log_buffer.getvalue())
         self.assertIsNone(json_record['stack_info'])
+
+
+class FlatJSONFormatterTest(TestCase):
+    def setUp(self):
+        json_handler.setFormatter(FlatJSONFormatter())
+
+    def test_given_time_is_used_in_log_record(self):
+        logger.info('Sign up', extra={'time': DATETIME})
+        expected_time = '"time": "2015-09-01T06:09:42.797203"'
+        self.assertIn(expected_time, log_buffer.getvalue())
+
+    def test_current_time_is_used_by_default_in_log_record(self):
+        logger.info('Sign up', extra={'fizz': 'bazz'})
+        self.assertNotIn(DATETIME_ISO, log_buffer.getvalue())
+
+    def test_message_and_time_are_in_json_record_when_extra_is_blank(self):
+        logger.info('Sign up')
+        json_record = json.loads(log_buffer.getvalue())
+        expected_fields = set([
+            'message',
+            'time',
+        ])
+        self.assertTrue(expected_fields.issubset(json_record))
+
+    def test_message_and_time_and_extra_are_in_json_record_when_extra_is_provided(self):
+        logger.info('Sign up', extra={'fizz': 'bazz'})
+        json_record = json.loads(log_buffer.getvalue())
+        expected_fields = set([
+            'message',
+            'time',
+            'fizz',
+        ])
+        self.assertTrue(expected_fields.issubset(json_record))
+
+    def test_exc_info_is_logged(self):
+        try:
+            raise ValueError('something wrong')
+        except ValueError:
+            logger.error('Request failed', exc_info=True)
+        json_record = json.loads(log_buffer.getvalue())
+        self.assertIn(
+            'Traceback (most recent call last)',
+            json_record['exc_info']
+        )
+
+    def test_builtin_types_are_serialized(self):
+        logger.log(level=logging.ERROR, msg='Payment was sent', extra={
+            'first_name': 'bob',
+            'amount': 0.00497265,
+            'context': {
+                'tags': ['fizz', 'bazz'],
+            },
+            'things': ('a', 'b'),
+            'ok': True,
+            'none': None,
+        })
+
+        json_record = json.loads(log_buffer.getvalue())
+        self.assertEqual(json_record['first_name'], 'bob')
+        self.assertEqual(json_record['amount'], 0.00497265)
+        self.assertEqual(json_record['context'], "{'tags': ['fizz', 'bazz']}")
+        self.assertEqual(json_record['things'], "('a', 'b')")
+        self.assertEqual(json_record['ok'], True)
+        self.assertEqual(json_record['none'], None)
+
+    def test_decimal_is_serialized_as_string(self):
+        logger.log(level=logging.ERROR, msg='Payment was sent', extra={
+            'amount': Decimal('0.00497265')
+        })
+        expected_amount = '"amount": "0.00497265"'
+        self.assertIn(expected_amount, log_buffer.getvalue())
+
+    def test_django_wsgi_request_is_serialized_as_dict(self):
+        request = WSGIRequest({
+            'PATH_INFO': 'bogus',
+            'REQUEST_METHOD': 'bogus',
+            'CONTENT_TYPE': 'text/html; charset=utf8',
+            'wsgi.input': BytesIO(b''),
+        })
+
+        logger.log(level=logging.ERROR, msg='Django response error', extra={
+            'status_code': 500,
+            'request': request,
+            'dict': {
+                'request': request,
+            },
+            'list': [request],
+        })
+        json_record = json.loads(log_buffer.getvalue())
+        self.assertEqual(json_record['status_code'], 500)
+        self.assertEqual(json_record['request'], "<WSGIRequest: BOGUS '/bogus'>")
+        self.assertEqual(json_record['dict'], "{'request': <WSGIRequest: BOGUS '/bogus'>}")
+        self.assertEqual(json_record['list'], "[<WSGIRequest: BOGUS '/bogus'>]")