RotatingStreamHandler: A version of logging's RotatingFileHandler that can take a Gzip file handler

RotatingStreamHandler.py

import logging
from logging.handlers import RotatingFileHandler
import gzip
import sys
import os


class RotatingStreamHandler(RotatingFileHandler):
   def __init__(self, filename, open_function=open, mode="a", **kwargs):
      """
      Like RotatingFileHandler, but allowing you to pass your own streaming file handler

      Use the original init from RotatingFileHandler, but afterwards replace the stream
      with our own file handler
      """
      self.open_function = open_function
      super().__init__(filename, **kwargs)
      self.mode = mode
      self.stream = self.open_function(
         self.baseFilename, mode=self.mode, encoding=self.encoding, errors=self.errors
      )

   def doRollover(self):
      """
      Use RotatingFileHandler's doRollover, but again, replace the stream with our own version
      afterwards
      """
      
      super().doRollover()
      self.stream = self.open_function(
         self.baseFilename, mode=self.mode, encoding=self.encoding, errors=self.errors
      )

   def shouldRollover(self, record):
      """
      A version of RotatingFileHandler's shouldRollover that does the comparison using the
      gzipped string instead of the full string
      """

      if self.stream is None:                 # delay was set...
         self.stream = self._open()
      if self.maxBytes > 0:                   # are we rolling over?
         pos = self.stream.tell()
         if not pos:
               # gh-116263: Never rollover an empty file
               return False
         msg = "%s\n" % self.format(record)
         
         # Get Gzip compressd representation of new message
         msg_bytes = gzip.compress(bytes(msg, 'utf-8'))
         if pos + self._gzip_size(msg) >= self.maxBytes:
               # See bpo-45401: Never rollover anything other than regular files
               if os.path.exists(self.baseFilename) and not os.path.isfile(self.baseFilename):
                  return False
               return True
      return False

   def _gzip_size(self, msg):
      """
      Get the size in bytes of a Gzipped representation of a string
      """

      msg_bytes = gzip.compress(bytes(msg, 'utf-8'))
      return sys.getsizeof(msg_bytes)


# Usage
# ===
raw_logger = logger = logging.getLogger('raw')
raw_handler = RotatingStreamHandler(
   filename="raw.log.gz", open_function=gzip.open, mode="wt", maxBytes=1024, backupCount=10
)
raw_logger.addHandler(raw_handler)


# Demo
# ===
raw_logger.warning("qui officia deserunt mollit anim id est laborum.")
raw_logger.warning("emque laudantium, totam rem aperiam, eaque ipsa quae ab illo inventore veritatis")
raw_logger.warning(" et quasi architecto beatae vitae dicta sunt explicabo. Nemo enim ipsam voluptat")
raw_logger.warning("em quia voluptas sit aspernatur aut odit aut fugit, sed quia consequuntur magni ")
raw_logger.warning("dolores eos qui ratione voluptatem sequi nesciunt. Neque porro quisquam est, qui")
raw_logger.warning(" dolorem ipsum quia dolor sit amet, consectetur, adipisci velit, sed quia non nu")
raw_logger.warning("mquam eius modi tempora incidunt ut labore et dolore magnam aliquam quaerat volu")
raw_logger.warning("ptatem. Ut enim ad minima veniam, quis nostrum exercitationem ullam corporis sus")
raw_logger.warning("cipit laboriosam, nisi ut aliquid ex ea commodi consequatur? Quis autem vel eum ")
raw_logger.warning("iure reprehenderit qui in ea voluptate velit esse quam nihil molestiae consequat")
raw_logger.warning("At vero eos et accusamus et iusto odio dignissimos ducimus qui blanditiis praese")
raw_logger.warning("ntium voluptatum deleniti atque corrupti quos dolores et quas molestias exceptur")
raw_logger.warning("i sint occaecati cupiditate non provident, similique sunt in culpa qui officia d")
raw_logger.warning("eserunt mollitia animi, id est laborum et dolorum fuga. Et harum quidem rerum fa")
raw_logger.warning("cilis est et expedita distinctio. Nam libero tempore, cum soluta nobis est elige")
raw_logger.warning("ndi optio cumque nihil impedit quo minus id quod maxime placeat facere possimus,")
raw_logger.warning(" omnis voluptas assumenda est, omnis dolor repellendus. Temporibus autem quibusd")
raw_logger.warning("am et aut officiis debitis aut rerum necessitatibus saepe eveniet ut et voluptat")
raw_logger.warning("es repudiandae sint et molestiae non recusandae. Itaque earum rerum hic tenetur ")
raw_logger.warning("a sapiente delectus, ut aut reiciendis voluptatibus maiores alias consequatur au")
raw_logger.warning("But I must explain to you how all this mistaken idea of denouncing pleasure and ")
raw_logger.warning("praising pain was born and I will give you a complete account of the system, and")
raw_logger.warning(" expound the actual teachings of the great explorer of the truth, the master-bui")
raw_logger.warning("lder of human happiness. No one rejects, dislikes, or avoids pleasure itself, be")
raw_logger.warning("cause it is pleasure, but because those who do not know how to pursue pleasure r")
raw_logger.warning("ationally encounter consequences that are extremely painful. Nor again is there ")
raw_logger.warning("anyone who loves or pursues or desires to obtain pain of itself, because it is p")
raw_logger.warning("ain, but because occasionally circumstances occur in which toil and pain can pro")
raw_logger.warning("cure him some great pleasure. To take a trivial example, which of us ever undert")
raw_logger.warning("akes laborious physical exercise, except to obtain some advantage from it? But w")
raw_logger.warning("ho has any right to find fault with a man who chooses to enjoy a pleasure that h")
raw_logger.warning("as no annoying consequences, or one who avoids a pain that produces no resultant")
raw_logger.warning("On the other hand, we denounce with righteous indignation and dislike men who ar")
raw_logger.warning("e so beguiled and demoralized by the charms of pleasure of the moment, so blinde")
raw_logger.warning("d by desire, that they cannot foresee the pain and trouble that are bound to ens")
raw_logger.warning("ue; and equal blame belongs to those who fail in their duty through weakness of ")
raw_logger.warning("will, which is the same as saying through shrinking from toil and pain. These ca")
raw_logger.warning("ses are perfectly simple and easy to distinguish. In a free hour, when our power")
raw_logger.warning(" of choice is untrammelled and when nothing prevents our being able to do what w")
raw_logger.warning("e like best, every pleasure is to be welcomed and every pain avoided. But in cer")
raw_logger.warning("tain circumstances and owing to the claims of duty or the obligations of busines")
raw_logger.warning("s it will frequently occur that pleasures have to be repudiated and annoyances a")
raw_logger.warning("ccepted. The wise man therefore always holds in these matters to this principle ")
raw_logger.warning("of selection: he rejects pleasures to secure other greater pleasures, or else he")
raw_logger.warning("Lorem ipsum dolor sit amet, consectetur adipiscing elit,")
raw_logger.warning("sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.")
raw_logger.warning("Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi")
raw_logger.warning("ut aliquip ex ea commodo consequat. Duis aute irure dolor in reprehenderit")
raw_logger.warning("in voluptate velit esse cillum dolore eu fugiat nulla pariatur. Excepteur sint")
raw_logger.warning("occaecat cupidatat non proident, sunt in culpa")