openrvdas.logger.readers.database_reader

No module-level documentation available.
 1#!/usr/bin/env python3
 2
 3import logging
 4import time
 5
 6from logger.readers.reader import TimestampedReader  # noqa: E402
 7
 8# Don't freak out if we can't find database settings - unless they actually
 9# try to instantiate a DatabaseReader.
10try:
11    from database.settings import DATABASE_ENABLED, Connector
12    from database.settings import DEFAULT_DATABASE, DEFAULT_DATABASE_HOST
13    from database.settings import DEFAULT_DATABASE_USER, DEFAULT_DATABASE_PASSWORD
14    DATABASE_SETTINGS_FOUND = True
15except ModuleNotFoundError:
16    DATABASE_SETTINGS_FOUND = False
17    DEFAULT_DATABASE = DEFAULT_DATABASE_HOST = None
18    DEFAULT_DATABASE_USER = DEFAULT_DATABASE_PASSWORD = None
19
20
21################################################################################
22# Read records from specified table.
23class DatabaseReader(TimestampedReader):
24    """
25    Database records to DASRecords.
26    """
27    ############################
28
29    def __init__(self, fields=None,
30                 database=DEFAULT_DATABASE, host=DEFAULT_DATABASE_HOST,
31                 user=DEFAULT_DATABASE_USER, password=DEFAULT_DATABASE_PASSWORD,
32                 tail=False, sleep_interval=1.0, **kwargs):
33        super().__init__(**kwargs)
34
35        if not DATABASE_SETTINGS_FOUND:
36            raise RuntimeError('File database/settings.py not found. Database '
37                               'functionality is not available. Have you copied '
38                               'over database/settings.py.dist to settings.py?')
39        if not DATABASE_ENABLED:
40            raise RuntimeError('Database not configured in database/settings.py; '
41                               'DatabaseReader unavailable.')
42
43        self.fields = fields
44        self.db = Connector(database=database, host=host, user=user,
45                            password=password, tail=tail)
46        self.sleep_interval = sleep_interval
47        self.next_id = 1
48
49    ############################
50    def read(self, no_block=False):
51        """Read next record in table. Sleep and retry if there's no new
52        record to read, unless no_block is specified, in which case, return
53        whatever we found."""
54        while True:
55            record = self.db.read(self.fields)
56            if record or no_block:
57                return record
58            logging.debug('No new record returned by database read. Sleeping')
59            time.sleep(self.sleep_interval)
60
61    ############################
62    def read_range(self, start=None, stop=None):
63        """Read all records beginning at record number 'stop' and ending
64        *before* record number 'stop'. If stop is None, read all available
65        following records."""
66
67        if stop is not None:
68            num_records = stop - start
69        else:
70            num_records = None
71        return self.db.read(self.fields, start=start, num_records=num_records)
72
73    ############################
74    def read_time_range(self, start_time=None, stop_time=None):
75        """Read the next records from table based on timestamps. If start_time
76        is None, use the timestamp of the last read record. If stop_time is None,
77        read all records since then."""
78        return self.db.read_time(self.fields, start_time=start_time,
79                                 stop_time=stop_time)
class DatabaseReader(logger.readers.reader.TimestampedReader): 
24class DatabaseReader(TimestampedReader):
25    """
26    Database records to DASRecords.
27    """
28    ############################
29
30    def __init__(self, fields=None,
31                 database=DEFAULT_DATABASE, host=DEFAULT_DATABASE_HOST,
32                 user=DEFAULT_DATABASE_USER, password=DEFAULT_DATABASE_PASSWORD,
33                 tail=False, sleep_interval=1.0, **kwargs):
34        super().__init__(**kwargs)
35
36        if not DATABASE_SETTINGS_FOUND:
37            raise RuntimeError('File database/settings.py not found. Database '
38                               'functionality is not available. Have you copied '
39                               'over database/settings.py.dist to settings.py?')
40        if not DATABASE_ENABLED:
41            raise RuntimeError('Database not configured in database/settings.py; '
42                               'DatabaseReader unavailable.')
43
44        self.fields = fields
45        self.db = Connector(database=database, host=host, user=user,
46                            password=password, tail=tail)
47        self.sleep_interval = sleep_interval
48        self.next_id = 1
49
50    ############################
51    def read(self, no_block=False):
52        """Read next record in table. Sleep and retry if there's no new
53        record to read, unless no_block is specified, in which case, return
54        whatever we found."""
55        while True:
56            record = self.db.read(self.fields)
57            if record or no_block:
58                return record
59            logging.debug('No new record returned by database read. Sleeping')
60            time.sleep(self.sleep_interval)
61
62    ############################
63    def read_range(self, start=None, stop=None):
64        """Read all records beginning at record number 'stop' and ending
65        *before* record number 'stop'. If stop is None, read all available
66        following records."""
67
68        if stop is not None:
69            num_records = stop - start
70        else:
71            num_records = None
72        return self.db.read(self.fields, start=start, num_records=num_records)
73
74    ############################
75    def read_time_range(self, start_time=None, stop_time=None):
76        """Read the next records from table based on timestamps. If start_time
77        is None, use the timestamp of the last read record. If stop_time is None,
78        read all records since then."""
79        return self.db.read_time(self.fields, start_time=start_time,
80                                 stop_time=stop_time)

Database records to DASRecords.

DatabaseReader( fields=None, database=None, host=None, user=None, password=None, tail=False, sleep_interval=1.0, **kwargs) 
30    def __init__(self, fields=None,
31                 database=DEFAULT_DATABASE, host=DEFAULT_DATABASE_HOST,
32                 user=DEFAULT_DATABASE_USER, password=DEFAULT_DATABASE_PASSWORD,
33                 tail=False, sleep_interval=1.0, **kwargs):
34        super().__init__(**kwargs)
35
36        if not DATABASE_SETTINGS_FOUND:
37            raise RuntimeError('File database/settings.py not found. Database '
38                               'functionality is not available. Have you copied '
39                               'over database/settings.py.dist to settings.py?')
40        if not DATABASE_ENABLED:
41            raise RuntimeError('Database not configured in database/settings.py; '
42                               'DatabaseReader unavailable.')
43
44        self.fields = fields
45        self.db = Connector(database=database, host=host, user=user,
46                            password=password, tail=tail)
47        self.sleep_interval = sleep_interval
48        self.next_id = 1

quiet - if type checking should log type errors or operate silently.

Two additional arguments govern how records will be encoded/decoded
from bytes, if desired by the Writer subclass when it calls
_encode_str() or _decode_bytes:

encoding - 'utf-8' by default. If empty or None, do not attempt any
        decoding and return raw bytes. Other possible encodings are
        listed in online documentation here:
        https://docs.python.org/3/library/codecs.html#standard-encodings

encoding_errors - 'ignore' by default. Other error strategies are
        'strict', 'replace', and 'backslashreplace', described here:
        https://docs.python.org/3/howto/unicode.html#encodings

mirror_to - Optional Writer to which all records read or transformed
        by this module (if it is a Reader or Transform) will be
        "mirrored" (copied). Mirroring happens asynchronously via
        a queue and background thread to minimize impact on the
        primary data flow. Writers cannot be mirrored.
fields
db
sleep_interval
next_id
def read(self, no_block=False): 
51    def read(self, no_block=False):
52        """Read next record in table. Sleep and retry if there's no new
53        record to read, unless no_block is specified, in which case, return
54        whatever we found."""
55        while True:
56            record = self.db.read(self.fields)
57            if record or no_block:
58                return record
59            logging.debug('No new record returned by database read. Sleeping')
60            time.sleep(self.sleep_interval)

Read next record in table. Sleep and retry if there's no new record to read, unless no_block is specified, in which case, return whatever we found.

def read_range(self, start=None, stop=None): 
63    def read_range(self, start=None, stop=None):
64        """Read all records beginning at record number 'stop' and ending
65        *before* record number 'stop'. If stop is None, read all available
66        following records."""
67
68        if stop is not None:
69            num_records = stop - start
70        else:
71            num_records = None
72        return self.db.read(self.fields, start=start, num_records=num_records)

Read all records beginning at record number 'stop' and ending before record number 'stop'. If stop is None, read all available following records.

def read_time_range(self, start_time=None, stop_time=None): 
75    def read_time_range(self, start_time=None, stop_time=None):
76        """Read the next records from table based on timestamps. If start_time
77        is None, use the timestamp of the last read record. If stop_time is None,
78        read all records since then."""
79        return self.db.read_time(self.fields, start_time=start_time,
80                                 stop_time=stop_time)

Read the next records from table based on timestamps. If start_time is None, use the timestamp of the last read record. If stop_time is None, read all records since then.