openrvdas.logger.transforms.prefix_transform

 1#!/usr/bin/env python3
 2
 3import logging
 4import re
 5
 6from logger.transforms.transform import Transform  # noqa: E402
 7
 8
 9################################################################################
10class PrefixTransform(Transform):
11    """Prepend a prefix to a text record."""
12
13    def __init__(self, prefix, sep=' ', **kwargs):
14        """Prepend the specified prefix to the record, using space as the default
15        separator. If prefix is a <regex>:prefix map, go through in order and use
16        the prefix of the first regex that matches. If no prefix matches, raise a
17        warning (if quiet != True) and return None.
18
19        Note: order of map evaluation is supposed to be guaranteed as of Python 3.7,
20        but trust at your own risk.
21
22        Note: the empty string '' as a regex will match all non-empty strings, so
23        if you trust Python ordering, it can be used as a backstop default match for
24        all records that don't match other strings.
25
26        prefix    Either a simple string prefix or a dict mapping. If a string, it
27                  will be used as the prefix. If a dict, it will be interpreted as
28                  a <regex>:prefix mapping, and the prefix corresponding to the
29                  first matching regex will be used. If no regex matches, None will
30                  be returned.
31
32        sep       A separator to be used between prefix and record.
33
34        quiet     If true, do not log a warning if no regex matches.
35        """
36        super().__init__(**kwargs)  # processes 'quiet' and type hints
37
38        if type(prefix) is str:
39            self.prefix = prefix + sep
40        elif type(prefix) is dict:
41            self.prefix = {re.compile(regex): pre + sep for regex, pre in prefix.items()}
42        else:
43            raise TypeError(f'prefix argument in PrefixTransform must be either a str or '
44                            f'dict. Received type "{type(prefix)}": {prefix}')
45
46    ############################
47    def transform(self, record: str) -> str:
48        """Prepend a prefix."""
49
50        # See if it's something we can process, and if not, try digesting
51        if not self.can_process_record(record):  # inherited from BaseModule()
52            return self.digest_record(record)  # inherited from BaseModule()
53
54        if type(self.prefix) is str:
55            return self.prefix + record
56
57        # If here, we've got a dict
58        for regex, pre in self.prefix.items():
59            if regex.search(record) is not None:
60                return pre + record
61
62        # If here, nothing matched
63        if not self.quiet:
64            logging.warning(f'PrefixTransform: no prefix pattern matched record "{record}"')
65        return None