aws lamdba compatible logtail handler in python

logger.py

import logging
import os
import queue
from logging import Logger
from typing import Optional
from logtail import LogtailHandler, handler, flusher, uploader

_logger: Optional[Logger] = None


class CustomLogtailHandler(LogtailHandler):
    """
    logtail handler implementation compatible with AWS Lambda functions,
    using queue.Queue instead of multiprocessing.queue since the latter uses multiprocessing.SemLock
    which is incompatible with aws lambdas
    Note: use logtail-python==0.1.4 only, since constructor is being overriden completely
    """
    def __init__(self, source_token, level=logging.NOTSET):
        # init grandparent but not parent to avoid instantiating the multiprocessing.queue
        super(LogtailHandler, self).__init__(level=level)

        # override parent initialization
        self.source_token = source_token
        self.host = handler.DEFAULT_HOST
        self.context = handler.DEFAULT_CONTEXT
        self.pipe = queue.Queue(maxsize=handler.DEFAULT_BUFFER_CAPACITY)
        self.uploader = uploader.Uploader(self.source_token, self.host)
        self.drop_extra_events = handler.DEFAULT_DROP_EXTRA_EVENTS
        self.include_extra_attributes = handler.DEFAULT_INCLUDE_EXTRA_ATTRIBUTES
        self.buffer_capacity = handler.DEFAULT_BUFFER_CAPACITY
        self.flush_interval = handler.DEFAULT_FLUSH_INTERVAL
        self.raise_exceptions = handler.DEFAULT_RAISE_EXCEPTIONS
        self.dropcount = 0
        self.flush_thread = flusher.FlushWorker(
            self.uploader,
            self.pipe,
            self.buffer_capacity,
            self.flush_interval
        )
        if self._is_main_process():
            self.flush_thread.start()


def get_logger(name: Optional[str] = None) -> Logger:
    """
    Creates a lambda specific logger with a standardized log format
    """
    global _logger
    if _logger: 
        return _logger
        
    _logger = logging.getLogger(name)
    _logger.setLevel(logging.DEBUG)

    # create logging formatter
    formatter = logging.Formatter('%(asctime)s :: %(name)s :: %(levelname)s :: %(message)s')

    # add console handler
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.DEBUG)
    console_handler.setFormatter(formatter)
    _logger.addHandler(console_handler)

    # add logtail handler
    logtail_source_token = "YOUR_KEY"
    if logtail_source_token:
        _logger.debug('logtail source token found! Adding custom logtail handler')
        
        logtail_handler = CustomLogtailHandler(source_token=logtail_source_token)
        logtail_handler.setLevel(logging.DEBUG)
        logtail_handler.setFormatter(formatter)
        _logger.addHandler(logtail_handler)
        
        _logger.debug('custom logtail handler initiated')
    else:
        _logger.debug('logtail source token not found in secrets. Only console stream will be added')

    return _logger

Doker

---