Sunday, 2 July 2023

Python: format string output to JSON

I'm playing around with FastAPI and Structlog and wanted to test and convert log format from plain text/string to JSON format for better readability and processing by the log aggregator platforms. Facing a case where certain log output are available in JSON but rest in plain string.

Current Output

INFO:     127.0.0.1:62154 - "GET /api/preface HTTP/1.1" 200 OK
INFO:     127.0.0.1:62154 - "GET /loader.json HTTP/1.1" 200 OK
INFO:     127.0.0.1:62155 - "GET /hello_world HTTP/1.1" 200 OK
{"key":"test_key","message":"Push to NFS Success","event":"Testing Fast API..","logger":"test_my_api","filename":"main.py","func_name":"Hello_World","process":23760,"module":"docker","thread":23140,"pathname":"D:\\my_work\\fast_api\\main.py","process_name":"SpawnProcess-1","level":"info","time-iso":"2023-06-30T15:25:03.113400Z"}

Expected Output:

{ "level" : "INFO" , IP : 127.0.0.1:62154 , method : GET, endpoint: /api/preface, protocol : HTTP/1.1, status_code : 200, status : OK }
{ "level" : "INFO" , IP : 127.0.0.1:62154 , method : GET, endpoint: /loader.json, protocol : HTTP/1.1, status_code : 200, status : OK }
{ "level" : "INFO" , IP : 127.0.0.1:62154 , method : GET, endpoint: /hello_world, protocol : HTTP/1.1, status_code : 200, status : OK }
{"key":"test_key","message":"Push to NFS Success","event":"Testing Fast API..","logger":"test_my_api","filename":"main.py","func_name":"Hello_World","process":23760,"module":"docker","thread":23140,"pathname":"D:\\my_work\\fast_api\\main.py","process_name":"SpawnProcess-1","level":"info","time-iso":"2023-06-30T15:25:03.113400Z"}

What am I missing here ? thanks !

struct.py

import orjson
import structlog
import logging

## Added only the necessary context.
class StructLogTest:
    def __init__(self, logging_level=logging.DEBUG, logger_name="test"):
        self.logging_level = logging_level
        self.logger_name = logger_name
        StructLogTest.logger_name_var = self.logger_name
        self.configure_structlog(self.logging_level, self.logger_name)

    def logger_name(_, __, event_dict):
        event_dict["test_log"] = StructLogTest.logger_name_var
        return event_dict


    @staticmethod
    def configure_structlog(logging_level, logger_name):
        structlog.configure(
            processors=[
                StructLogTest.logger_name,
                structlog.threadlocal.merge_threadlocal,
                structlog.processors.CallsiteParameterAdder(),
                structlog.processors.add_log_level,
                structlog.stdlib.PositionalArgumentsFormatter(),
                structlog.processors.StackInfoRenderer(),
                structlog.processors.format_exc_info,
                structlog.processors.TimeStamper(fmt="iso", utc=True, key="time-iso"),
                structlog.processors.JSONRenderer(serializer=orjson.dumps),
            ],
            wrapper_class=structlog.make_filtering_bound_logger(logging_level),
            context_class=dict,
            logger_factory=structlog.BytesLoggerFactory(),
        )
        return structlog

    def define_Logger(self, *args, **kwargs):
        return structlog.get_logger(*args, **kwargs)

    def info(self, message, *args, **kwargs):
        return structlog.get_logger().info(message, *args, **kwargs)
    
    and other methods so on..


    

main.py

from struct import StructLogTest
from fastapi import APIRouter
import requests
from requests.auth import HTTPBasicAuth
from requests import Response

log = StructLogTest(logger_name="test_my_api")
log = log.get_Logger()

@router.get("/hello_world")
def Hello_World():
    logg = log.bind(key=test_key)
    logg.info(
        "Testing Fast API..",
        message=some_other_meaningful_function.dump(),
    )
    return {" Hello World !! "}


from Python: format string output to JSON

No comments:

Post a Comment