openrvdas.logger.utils.stderr_logging

No module-level documentation available.
 1#!/usr/bin/env python3
 2
 3import json
 4import logging
 5import time
 6
 7from logger.utils.timestamp import LOGGING_TIME_FORMAT  # noqa: E402
 8
 9DEFAULT_LOGGING_FORMAT = ' '.join([
10    '%(asctime)-15sZ',
11    '%(levelno)s',
12    '%(levelname)s',
13    '%(filename)s:%(lineno)d',
14    '%(message)s',
15])
16
17STDERR_FORMATTER = logging.Formatter(fmt=DEFAULT_LOGGING_FORMAT,
18                                     datefmt=LOGGING_TIME_FORMAT)
19STDERR_FORMATTER.converter = time.gmtime
20
21
22class StdErrLoggingHandler(logging.Handler):
23    """Write Python logging.* messages to whatever writer we're passed. To
24    use, run
25
26      logging.getLogger().addHandler(StdErrLoggingHandler(my_writer))
27    """
28
29    def __init__(self, writers, parse_to_json=False):
30        """
31        writers - either a Writer object or a list of Writer objects
32
33        parse_to_json - if true, expect to receive output as
34            a string in DEFAULT_LOGGING_FORMAT, and parse it into a dict of
35            the respective values.
36        """
37        super().__init__()
38        self.writers = writers
39        self.parse_to_json = parse_to_json
40
41    def emit(self, record):
42        # Temporarily push the logging level up as high as it can go to
43        # effectively disable recursion induced by logging that occurs inside
44        # whatever writer we're using.
45        log_level = logging.root.getEffectiveLevel()
46        logging.root.setLevel(logging.CRITICAL)
47
48        message = STDERR_FORMATTER.format(record)
49
50        # If we're supposed to parse string into a dict
51        if self.parse_to_json:
52            try:
53                (asctime, levelno, levelname, mesg) = message.split(' ', maxsplit=3)
54                levelno = int(levelno)
55                fields = {'asctime': asctime, 'levelno': levelno,
56                          'levelname': levelname, 'message': mesg}
57            except ValueError:
58                fields = {'message': message}
59            message = json.dumps(fields)
60
61        # Write message out to each writer
62        if isinstance(self.writers, list):
63            [writer.write(message) for writer in self.writers if writer]
64        else:
65            self.writers.write(message)
66        logging.root.setLevel(log_level)
DEFAULT_LOGGING_FORMAT = '%(asctime)-15sZ %(levelno)s %(levelname)s %(filename)s:%(lineno)d %(message)s'
STDERR_FORMATTER = <logging.Formatter object>
class StdErrLoggingHandler(logging.Handler): 
23class StdErrLoggingHandler(logging.Handler):
24    """Write Python logging.* messages to whatever writer we're passed. To
25    use, run
26
27      logging.getLogger().addHandler(StdErrLoggingHandler(my_writer))
28    """
29
30    def __init__(self, writers, parse_to_json=False):
31        """
32        writers - either a Writer object or a list of Writer objects
33
34        parse_to_json - if true, expect to receive output as
35            a string in DEFAULT_LOGGING_FORMAT, and parse it into a dict of
36            the respective values.
37        """
38        super().__init__()
39        self.writers = writers
40        self.parse_to_json = parse_to_json
41
42    def emit(self, record):
43        # Temporarily push the logging level up as high as it can go to
44        # effectively disable recursion induced by logging that occurs inside
45        # whatever writer we're using.
46        log_level = logging.root.getEffectiveLevel()
47        logging.root.setLevel(logging.CRITICAL)
48
49        message = STDERR_FORMATTER.format(record)
50
51        # If we're supposed to parse string into a dict
52        if self.parse_to_json:
53            try:
54                (asctime, levelno, levelname, mesg) = message.split(' ', maxsplit=3)
55                levelno = int(levelno)
56                fields = {'asctime': asctime, 'levelno': levelno,
57                          'levelname': levelname, 'message': mesg}
58            except ValueError:
59                fields = {'message': message}
60            message = json.dumps(fields)
61
62        # Write message out to each writer
63        if isinstance(self.writers, list):
64            [writer.write(message) for writer in self.writers if writer]
65        else:
66            self.writers.write(message)
67        logging.root.setLevel(log_level)

Write Python logging.* messages to whatever writer we're passed. To use, run

logging.getLogger().addHandler(StdErrLoggingHandler(my_writer))

StdErrLoggingHandler(writers, parse_to_json=False) 
30    def __init__(self, writers, parse_to_json=False):
31        """
32        writers - either a Writer object or a list of Writer objects
33
34        parse_to_json - if true, expect to receive output as
35            a string in DEFAULT_LOGGING_FORMAT, and parse it into a dict of
36            the respective values.
37        """
38        super().__init__()
39        self.writers = writers
40        self.parse_to_json = parse_to_json

writers - either a Writer object or a list of Writer objects

parse_to_json - if true, expect to receive output as a string in DEFAULT_LOGGING_FORMAT, and parse it into a dict of the respective values.

writers
parse_to_json
def emit(self, record): 
42    def emit(self, record):
43        # Temporarily push the logging level up as high as it can go to
44        # effectively disable recursion induced by logging that occurs inside
45        # whatever writer we're using.
46        log_level = logging.root.getEffectiveLevel()
47        logging.root.setLevel(logging.CRITICAL)
48
49        message = STDERR_FORMATTER.format(record)
50
51        # If we're supposed to parse string into a dict
52        if self.parse_to_json:
53            try:
54                (asctime, levelno, levelname, mesg) = message.split(' ', maxsplit=3)
55                levelno = int(levelno)
56                fields = {'asctime': asctime, 'levelno': levelno,
57                          'levelname': levelname, 'message': mesg}
58            except ValueError:
59                fields = {'message': message}
60            message = json.dumps(fields)
61
62        # Write message out to each writer
63        if isinstance(self.writers, list):
64            [writer.write(message) for writer in self.writers if writer]
65        else:
66            self.writers.write(message)
67        logging.root.setLevel(log_level)

Do whatever it takes to actually log the specified logging record.

This version is intended to be implemented by subclasses and so raises a NotImplementedError.