Simple auto flushing, thread safe logger for Elastic consumer (Singleton)

log.py

from __future__ import annotations
from typing import Union
import os
import uuid
import json
from datetime import datetime
from threading import Lock


class Log( object ):
  __instance__: Union[ Log, None ] = None
  __lock: Lock = Lock()

  def __init__ ( self ):
    if Log.__instance__ is None:
      self.app_name = os.environ.get( 'app_name', 'default_application' )
      self.app_id = os.environ.get( 'app_id', str( uuid.uuid4() ) )
      Log.__instance__ = self

  @staticmethod
  def get_instance () -> Log:
    if not Log.__instance__:
      Log()
    return Log.__instance__

  def write ( self, message, took: float = None, additional: dict = None ):
    date = datetime.now()
    body = {
      'id'      : str( uuid.uuid4() ),
      'app'     : self.app_name,
      'node'    : self.app_id,
      'message' : message,
      'date'    : str( date.isoformat() ),
      'dateUnix': str( date.strftime( '%s' ) ),
      'took'    : 0.0
    }

    if took is not None:
      body[ 'took' ] = took

    if additional is not None:
      for key in additional:
        body[ key ] = additional[ key ]

    with self.__lock:
      print( json.dumps( body ), flush=True )