openrvdas.logger.writers.database_writer

No module-level documentation available.
  1#!/usr/bin/env python3
  2
  3import logging
  4import pprint
  5import time
  6
  7from typing import Union
  8
  9from logger.utils.das_record import DASRecord  # noqa: E402
 10from logger.writers.writer import Writer  # noqa: E402
 11
 12# Don't freak out if we can't find database settings - unless they actually
 13# try to instantiate a DatabaseWriter.
 14try:
 15    from database.settings import DATABASE_ENABLED, Connector  # noqa: E402
 16    from database.settings import DEFAULT_DATABASE, DEFAULT_DATABASE_HOST  # noqa: E402
 17    from database.settings import DEFAULT_DATABASE_USER, DEFAULT_DATABASE_PASSWORD  # noqa: E402
 18    DATABASE_SETTINGS_FOUND = True
 19except ModuleNotFoundError:
 20    DATABASE_SETTINGS_FOUND = False
 21    DEFAULT_DATABASE = DEFAULT_DATABASE_HOST = None
 22    DEFAULT_DATABASE_USER = DEFAULT_DATABASE_PASSWORD = None
 23
 24
 25class DatabaseWriter(Writer):
 26    def __init__(self, database=DEFAULT_DATABASE, host=DEFAULT_DATABASE_HOST,
 27                 user=DEFAULT_DATABASE_USER, password=DEFAULT_DATABASE_PASSWORD,
 28                 save_source=True, **kwargs):
 29        """Write to the passed record to a database table. With connectors
 30        written so far (MySQL and Mongo), writes values in the records as
 31        timestamped field-value pairs. If save_source=True, also save the
 32        source record we are passed.
 33
 34        Expects passed source records to be in one of two formats:
 35
 36        1) DASRecord
 37
 38        2) A dict encoding optionally a source data_id and timestamp and a
 39           mandatory 'fields' key of field_name: value pairs. This is the format
 40           emitted by default by ParseTransform:
 41        ```
 42           {
 43             'data_id': ...,
 44             'timestamp': ...,
 45             'fields': {
 46               field_name: value,    # use default timestamp of 'now'
 47               field_name: value,
 48               ...
 49             }
 50           }
 51        ```
 52        A twist on format (2) is that the values may either be a singleton
 53        (int, float, string, etc) or a list. If the value is a singleton,
 54        it is taken at face value. If it is a list, it is assumed to be a
 55        list of (value, timestamp) tuples, in which case the top-level
 56        timestamp, if any, is ignored.
 57        ```
 58           {
 59             'data_id': ...,
 60             'timestamp': ...,
 61             'fields': {
 62                field_name: [(timestamp, value), (timestamp, value),...],
 63                field_name: [(timestamp, value), (timestamp, value),...],
 64                ...
 65             }
 66           }
 67        ```
 68        """
 69        super().__init__(**kwargs)  # processes 'quiet' and type hints
 70
 71        if not DATABASE_SETTINGS_FOUND:
 72            raise RuntimeError('File database/settings.py not found. Database '
 73                               'functionality is not available. Have you copied '
 74                               'over database/settings.py.dist to settings.py?')
 75        if not DATABASE_ENABLED:
 76            raise RuntimeError('Database not configured in database/settings.py; '
 77                               'DatabaseWriter unavailable.')
 78
 79        self.db = Connector(database=database, host=host,
 80                            user=user, password=password,
 81                            save_source=save_source)
 82
 83    ############################
 84    def _table_exists(self, table_name):
 85        """Does the specified table exist in the database?"""
 86        return self.db.table_exists(table_name)
 87
 88    ############################
 89    def _write_record(self, record: Union[DASRecord, dict]):
 90        """Write record to table. Connectors assume we've got a DASRecord, but
 91        check; if we don't see if it's a suitably-formatted dict that we can
 92        convert into a DASRecord.
 93        """
 94        if not self.can_process_record(record):  # inherited from BaseModule()
 95            self.digest_record(record)  # inherited from BaseModule()
 96            return
 97
 98        if isinstance(record, dict):
 99            try:
100                data_id = record.get('data_id', 'no_data_id')
101                timestamp = record.get('timestamp')
102                fields = record['fields']
103                record = DASRecord(data_id=data_id, timestamp=timestamp, fields=fields)
104            except KeyError:
105                logging.error('Unable to create DASRecord from dict: %s',
106                              pprint.pformat(record))
107        self.db.write_record(record)
108
109    ############################
110    def _delete_table(self, table_name):
111        """Delete a table."""
112        self.db.delete_table(table_name)
113
114    ############################
115    def write(self, record: Union[DASRecord, dict]):
116        """Write out record. Connectors assume we've got a DASRecord, so check
117        what we've got and convert as necessary.
118        """
119        # See if it's something we can process, and if not, try digesting
120        if not self.can_process_record(record):  # inherited from BaseModule()
121            self.digest_record(record)  # inherited from BaseModule()
122            return
123
124        # If we've been passed a DASRecord, things are easy: write it and return.
125        if isinstance(record, DASRecord):
126            self._write_record(record)
127            return
128
129        if not isinstance(record, dict):
130            if not self.quiet:
131                logging.error('Record passed to DatabaseWriter is not of type '
132                              '"DASRecord" or dict; is type "%s"', type(record))
133            return
134
135        # If here, our record is a dict, figure out whether it is a top-level
136        # field dict or not.
137        data_id = record.get('data_id')
138        timestamp = record.get('timestamp', time.time())
139        fields = record.get('fields')
140        if fields is None:
141            if not self.quiet:
142                logging.error('Dict record passed to DatabaseWriter has no "fields" '
143                              'key, which either means it\'s not a dict you should be '
144                              'passing, or it is in the old "field_dict" format that '
145                              'assumes key:value pairs are at the top level.')
146                logging.error('The record in question: %s', str(record))
147            return
148
149        # Now check whether our 'values' are singletons (in which case
150        # we've got a single record) or lists of tuples. Shortcut by
151        # checking only the first value in our 'fields' dict.
152        try:
153            first_key, first_value = next(iter(fields.items()))
154        except StopIteration:
155            # Empty fields
156            logging.debug('Empty "fields" dict in record: %s', str(record))
157            return
158
159        # If we've got a singleton, it's a single record. Convert to
160        # DASRecord and write it out.
161        if not isinstance(first_value, list):
162            das_record = DASRecord(data_id=data_id, timestamp=timestamp, fields=fields)
163            self._write_record(das_record)
164            return
165
166        # If we're here, our values (or at least our first one) are lists
167        # of (value, timestamp) pairs. First thing we do is
168        # reformat the data into a map of
169        #
170        #        {timestamp: {field:value, field:value],...}}
171        values_by_timestamp = {}
172        try:
173            for field, ts_value_list in fields.items():
174                for (timestamp, value) in ts_value_list:
175                    if timestamp not in values_by_timestamp:
176                        values_by_timestamp[timestamp] = {}
177                    values_by_timestamp[timestamp][field] = value
178        except ValueError:
179            if not self.quiet:
180                logging.error('Badly-structured field dictionary: %s: %s',
181                              field, pprint.pformat(ts_value_list))
182
183        # Now go through each timestamp, generate a DASRecord from its
184        # values, and write them.
185        for timestamp in sorted(values_by_timestamp):
186            das_record = DASRecord(data_id=data_id, timestamp=timestamp,
187                                   fields=values_by_timestamp[timestamp])
188            self._write_record(das_record)
class DatabaseWriter(logger.writers.writer.Writer): 
 26class DatabaseWriter(Writer):
 27    def __init__(self, database=DEFAULT_DATABASE, host=DEFAULT_DATABASE_HOST,
 28                 user=DEFAULT_DATABASE_USER, password=DEFAULT_DATABASE_PASSWORD,
 29                 save_source=True, **kwargs):
 30        """Write to the passed record to a database table. With connectors
 31        written so far (MySQL and Mongo), writes values in the records as
 32        timestamped field-value pairs. If save_source=True, also save the
 33        source record we are passed.
 34
 35        Expects passed source records to be in one of two formats:
 36
 37        1) DASRecord
 38
 39        2) A dict encoding optionally a source data_id and timestamp and a
 40           mandatory 'fields' key of field_name: value pairs. This is the format
 41           emitted by default by ParseTransform:
 42        ```
 43           {
 44             'data_id': ...,
 45             'timestamp': ...,
 46             'fields': {
 47               field_name: value,    # use default timestamp of 'now'
 48               field_name: value,
 49               ...
 50             }
 51           }
 52        ```
 53        A twist on format (2) is that the values may either be a singleton
 54        (int, float, string, etc) or a list. If the value is a singleton,
 55        it is taken at face value. If it is a list, it is assumed to be a
 56        list of (value, timestamp) tuples, in which case the top-level
 57        timestamp, if any, is ignored.
 58        ```
 59           {
 60             'data_id': ...,
 61             'timestamp': ...,
 62             'fields': {
 63                field_name: [(timestamp, value), (timestamp, value),...],
 64                field_name: [(timestamp, value), (timestamp, value),...],
 65                ...
 66             }
 67           }
 68        ```
 69        """
 70        super().__init__(**kwargs)  # processes 'quiet' and type hints
 71
 72        if not DATABASE_SETTINGS_FOUND:
 73            raise RuntimeError('File database/settings.py not found. Database '
 74                               'functionality is not available. Have you copied '
 75                               'over database/settings.py.dist to settings.py?')
 76        if not DATABASE_ENABLED:
 77            raise RuntimeError('Database not configured in database/settings.py; '
 78                               'DatabaseWriter unavailable.')
 79
 80        self.db = Connector(database=database, host=host,
 81                            user=user, password=password,
 82                            save_source=save_source)
 83
 84    ############################
 85    def _table_exists(self, table_name):
 86        """Does the specified table exist in the database?"""
 87        return self.db.table_exists(table_name)
 88
 89    ############################
 90    def _write_record(self, record: Union[DASRecord, dict]):
 91        """Write record to table. Connectors assume we've got a DASRecord, but
 92        check; if we don't see if it's a suitably-formatted dict that we can
 93        convert into a DASRecord.
 94        """
 95        if not self.can_process_record(record):  # inherited from BaseModule()
 96            self.digest_record(record)  # inherited from BaseModule()
 97            return
 98
 99        if isinstance(record, dict):
100            try:
101                data_id = record.get('data_id', 'no_data_id')
102                timestamp = record.get('timestamp')
103                fields = record['fields']
104                record = DASRecord(data_id=data_id, timestamp=timestamp, fields=fields)
105            except KeyError:
106                logging.error('Unable to create DASRecord from dict: %s',
107                              pprint.pformat(record))
108        self.db.write_record(record)
109
110    ############################
111    def _delete_table(self, table_name):
112        """Delete a table."""
113        self.db.delete_table(table_name)
114
115    ############################
116    def write(self, record: Union[DASRecord, dict]):
117        """Write out record. Connectors assume we've got a DASRecord, so check
118        what we've got and convert as necessary.
119        """
120        # See if it's something we can process, and if not, try digesting
121        if not self.can_process_record(record):  # inherited from BaseModule()
122            self.digest_record(record)  # inherited from BaseModule()
123            return
124
125        # If we've been passed a DASRecord, things are easy: write it and return.
126        if isinstance(record, DASRecord):
127            self._write_record(record)
128            return
129
130        if not isinstance(record, dict):
131            if not self.quiet:
132                logging.error('Record passed to DatabaseWriter is not of type '
133                              '"DASRecord" or dict; is type "%s"', type(record))
134            return
135
136        # If here, our record is a dict, figure out whether it is a top-level
137        # field dict or not.
138        data_id = record.get('data_id')
139        timestamp = record.get('timestamp', time.time())
140        fields = record.get('fields')
141        if fields is None:
142            if not self.quiet:
143                logging.error('Dict record passed to DatabaseWriter has no "fields" '
144                              'key, which either means it\'s not a dict you should be '
145                              'passing, or it is in the old "field_dict" format that '
146                              'assumes key:value pairs are at the top level.')
147                logging.error('The record in question: %s', str(record))
148            return
149
150        # Now check whether our 'values' are singletons (in which case
151        # we've got a single record) or lists of tuples. Shortcut by
152        # checking only the first value in our 'fields' dict.
153        try:
154            first_key, first_value = next(iter(fields.items()))
155        except StopIteration:
156            # Empty fields
157            logging.debug('Empty "fields" dict in record: %s', str(record))
158            return
159
160        # If we've got a singleton, it's a single record. Convert to
161        # DASRecord and write it out.
162        if not isinstance(first_value, list):
163            das_record = DASRecord(data_id=data_id, timestamp=timestamp, fields=fields)
164            self._write_record(das_record)
165            return
166
167        # If we're here, our values (or at least our first one) are lists
168        # of (value, timestamp) pairs. First thing we do is
169        # reformat the data into a map of
170        #
171        #        {timestamp: {field:value, field:value],...}}
172        values_by_timestamp = {}
173        try:
174            for field, ts_value_list in fields.items():
175                for (timestamp, value) in ts_value_list:
176                    if timestamp not in values_by_timestamp:
177                        values_by_timestamp[timestamp] = {}
178                    values_by_timestamp[timestamp][field] = value
179        except ValueError:
180            if not self.quiet:
181                logging.error('Badly-structured field dictionary: %s: %s',
182                              field, pprint.pformat(ts_value_list))
183
184        # Now go through each timestamp, generate a DASRecord from its
185        # values, and write them.
186        for timestamp in sorted(values_by_timestamp):
187            das_record = DASRecord(data_id=data_id, timestamp=timestamp,
188                                   fields=values_by_timestamp[timestamp])
189            self._write_record(das_record)

Base class Writer about which we know nothing else. By default the input format is Unknown unless overridden.

Passes arguments quiet, encoding and encoding_errors up to BaseModule

DatabaseWriter( database=None, host=None, user=None, password=None, save_source=True, **kwargs) 
27    def __init__(self, database=DEFAULT_DATABASE, host=DEFAULT_DATABASE_HOST,
28                 user=DEFAULT_DATABASE_USER, password=DEFAULT_DATABASE_PASSWORD,
29                 save_source=True, **kwargs):
30        """Write to the passed record to a database table. With connectors
31        written so far (MySQL and Mongo), writes values in the records as
32        timestamped field-value pairs. If save_source=True, also save the
33        source record we are passed.
34
35        Expects passed source records to be in one of two formats:
36
37        1) DASRecord
38
39        2) A dict encoding optionally a source data_id and timestamp and a
40           mandatory 'fields' key of field_name: value pairs. This is the format
41           emitted by default by ParseTransform:
42        ```
43           {
44             'data_id': ...,
45             'timestamp': ...,
46             'fields': {
47               field_name: value,    # use default timestamp of 'now'
48               field_name: value,
49               ...
50             }
51           }
52        ```
53        A twist on format (2) is that the values may either be a singleton
54        (int, float, string, etc) or a list. If the value is a singleton,
55        it is taken at face value. If it is a list, it is assumed to be a
56        list of (value, timestamp) tuples, in which case the top-level
57        timestamp, if any, is ignored.
58        ```
59           {
60             'data_id': ...,
61             'timestamp': ...,
62             'fields': {
63                field_name: [(timestamp, value), (timestamp, value),...],
64                field_name: [(timestamp, value), (timestamp, value),...],
65                ...
66             }
67           }
68        ```
69        """
70        super().__init__(**kwargs)  # processes 'quiet' and type hints
71
72        if not DATABASE_SETTINGS_FOUND:
73            raise RuntimeError('File database/settings.py not found. Database '
74                               'functionality is not available. Have you copied '
75                               'over database/settings.py.dist to settings.py?')
76        if not DATABASE_ENABLED:
77            raise RuntimeError('Database not configured in database/settings.py; '
78                               'DatabaseWriter unavailable.')
79
80        self.db = Connector(database=database, host=host,
81                            user=user, password=password,
82                            save_source=save_source)

Write to the passed record to a database table. With connectors written so far (MySQL and Mongo), writes values in the records as timestamped field-value pairs. If save_source=True, also save the source record we are passed.

Expects passed source records to be in one of two formats:

1) DASRecord

2) A dict encoding optionally a source data_id and timestamp and a mandatory 'fields' key of field_name: value pairs. This is the format emitted by default by ParseTransform:

   {
     'data_id': ...,
     'timestamp': ...,
     'fields': {
       field_name: value,    # use default timestamp of 'now'
       field_name: value,
       ...
     }
   }

A twist on format (2) is that the values may either be a singleton (int, float, string, etc) or a list. If the value is a singleton, it is taken at face value. If it is a list, it is assumed to be a list of (value, timestamp) tuples, in which case the top-level timestamp, if any, is ignored.

   {
     'data_id': ...,
     'timestamp': ...,
     'fields': {
        field_name: [(timestamp, value), (timestamp, value),...],
        field_name: [(timestamp, value), (timestamp, value),...],
        ...
     }
   }
db
def write(self, record: Union[logger.utils.das_record.DASRecord, dict]): 
116    def write(self, record: Union[DASRecord, dict]):
117        """Write out record. Connectors assume we've got a DASRecord, so check
118        what we've got and convert as necessary.
119        """
120        # See if it's something we can process, and if not, try digesting
121        if not self.can_process_record(record):  # inherited from BaseModule()
122            self.digest_record(record)  # inherited from BaseModule()
123            return
124
125        # If we've been passed a DASRecord, things are easy: write it and return.
126        if isinstance(record, DASRecord):
127            self._write_record(record)
128            return
129
130        if not isinstance(record, dict):
131            if not self.quiet:
132                logging.error('Record passed to DatabaseWriter is not of type '
133                              '"DASRecord" or dict; is type "%s"', type(record))
134            return
135
136        # If here, our record is a dict, figure out whether it is a top-level
137        # field dict or not.
138        data_id = record.get('data_id')
139        timestamp = record.get('timestamp', time.time())
140        fields = record.get('fields')
141        if fields is None:
142            if not self.quiet:
143                logging.error('Dict record passed to DatabaseWriter has no "fields" '
144                              'key, which either means it\'s not a dict you should be '
145                              'passing, or it is in the old "field_dict" format that '
146                              'assumes key:value pairs are at the top level.')
147                logging.error('The record in question: %s', str(record))
148            return
149
150        # Now check whether our 'values' are singletons (in which case
151        # we've got a single record) or lists of tuples. Shortcut by
152        # checking only the first value in our 'fields' dict.
153        try:
154            first_key, first_value = next(iter(fields.items()))
155        except StopIteration:
156            # Empty fields
157            logging.debug('Empty "fields" dict in record: %s', str(record))
158            return
159
160        # If we've got a singleton, it's a single record. Convert to
161        # DASRecord and write it out.
162        if not isinstance(first_value, list):
163            das_record = DASRecord(data_id=data_id, timestamp=timestamp, fields=fields)
164            self._write_record(das_record)
165            return
166
167        # If we're here, our values (or at least our first one) are lists
168        # of (value, timestamp) pairs. First thing we do is
169        # reformat the data into a map of
170        #
171        #        {timestamp: {field:value, field:value],...}}
172        values_by_timestamp = {}
173        try:
174            for field, ts_value_list in fields.items():
175                for (timestamp, value) in ts_value_list:
176                    if timestamp not in values_by_timestamp:
177                        values_by_timestamp[timestamp] = {}
178                    values_by_timestamp[timestamp][field] = value
179        except ValueError:
180            if not self.quiet:
181                logging.error('Badly-structured field dictionary: %s: %s',
182                              field, pprint.pformat(ts_value_list))
183
184        # Now go through each timestamp, generate a DASRecord from its
185        # values, and write them.
186        for timestamp in sorted(values_by_timestamp):
187            das_record = DASRecord(data_id=data_id, timestamp=timestamp,
188                                   fields=values_by_timestamp[timestamp])
189            self._write_record(das_record)

Write out record. Connectors assume we've got a DASRecord, so check what we've got and convert as necessary.