openrvdas.logger.writers.redis_writer

 1#!/usr/bin/env python3
 2
 3import logging
 4
 5# Don't barf if they don't have redis installed. Only complain if
 6# they actually try to use it, below
 7try:
 8    import redis
 9    REDIS_ENABLED = True
10except ModuleNotFoundError:
11    REDIS_ENABLED = False
12
13from logger.writers.writer import Writer  # noqa: E402
14
15DEFAULT_HOST = 'localhost'
16DEFAULT_PORT = '6379'
17
18
19class RedisWriter(Writer):
20    """Write to redis server pubsub channel."""
21
22    def __init__(self, channel, password=None, **kwargs):
23        """
24        Write text records to a Redis pubsub server channel.
25        ```
26        channel      Redis channel to write to, format channel[@hostname[:port]]
27        ```
28        """
29        super().__init__(**kwargs)  # processes 'quiet' and type hints
30
31        if not REDIS_ENABLED:
32            raise ModuleNotFoundError('RedisReader(): Redis is not installed. Please '
33                                      'try "pip3 install redis" prior to use.')
34        self.channel = channel
35        self.hostname = DEFAULT_HOST
36        self.port = DEFAULT_PORT
37
38        if channel.find('@') > 0:
39            (self.channel, self.hostname) = channel.split(sep='@', maxsplit=1)
40        if self.hostname.find(':') > 0:
41            (self.hostname, self.port) = self.hostname.split(sep=':', maxsplit=1)
42        self.port = int(self.port)
43
44        # Connect to the specified server
45        try:
46            self.redis = redis.StrictRedis(host=self.hostname, port=self.port,
47                                           password=password, decode_responses=True)
48            self.redis.ping()
49            self.pubsub = self.redis.pubsub()
50        except redis.exceptions.ConnectionError as e:
51            logging.error('Unable to connect to server at %s:%d',
52                          self.hostname, self.port)
53            raise e
54
55    ############################
56    def write(self, record):
57        """Write the record to the pubsub channel."""
58
59        # See if it's something we can process, and if not, try digesting
60        if not self.can_process_record(record):  # inherited from BaseModule()
61            self.digest_record(record)  # inherited from BaseModule()
62            return
63
64        # If record is not a string, try converting to JSON. If we don't know
65        # how, throw a hail Mary and force it into str format
66        # if not type(record) is str:
67        #  if type(record) in [int, float, bool, list, dict]:
68        #    record = json.dumps(record)
69        #  else:
70        #    record = str(record)
71
72        try:
73            self.redis.publish(self.channel, record)
74        except redis.exceptions.ConnectionError as e:
75            logging.error('Unable to connect to server at %s:%d',
76                          self.hostname, self.port)
77            raise e