openrvdas.logger.writers.writer
The biggest thing that the abstract parent class Writer now does is help with (optional) type checking of the child class' inputs. In the past, an explicit, but very awkward, form of type checking was used, with child classes passing the Transform class a list of input_format and output_format specifications.
That is now deprecated in favor of using Python's type hints. Type hints should be specified for the child class' write() method. Then the write() method can call self.can_process_record(record) to see whether it's one of the input types it can handle. If not, it can "return self.digest_record(record) to have the parent class try to deal with it:
E.g.: def write(self, record: Union[int, str, float]): if not self.can_process_record(record): # inherited from BaseModule() self.digest_record(record) # inherited from BaseModule() return [do normal writing here...
If no type hints are specified, can_process_record() will return True for all records except those of type "None" or "list". The logic is that digest_record() will return a None when given a None, and when given a list, will iteratively apply the write() to every element of the list in order.
Note that the child class can explicitly call super().__init__(quiet=True) or such to initialize the type checking and set its debugging level. If it is not explicitly initialized, it will be done implicitly the first time can_process_record() or digest_record() are called, but with the default of quiet=False.
1#!/usr/bin/env python3 2""" 3The biggest thing that the abstract parent class Writer now does is help 4with (optional) type checking of the child class' inputs. In the past, an 5explicit, but very awkward, form of type checking was used, with child classes 6passing the Transform class a list of input_format and output_format specifications. 7 8That is now deprecated in favor of using Python's type hints. Type hints should be 9specified for the child class' write() method. Then the write() method can 10call self.can_process_record(record) to see whether it's one of the input types it 11can handle. If not, it can "return self.digest_record(record) to have the parent 12class try to deal with it: 13 14E.g.: 15 def write(self, record: Union[int, str, float]): 16 if not self.can_process_record(record): # inherited from BaseModule() 17 self.digest_record(record) # inherited from BaseModule() 18 return 19 [do normal writing here... 20 21If no type hints are specified, can_process_record() will return True for all 22records *except* those of type "None" or "list". The logic is that digest_record() 23will return a None when given a None, and when given a list, will iteratively 24apply the write() to every element of the list in order. 25 26Note that the child class can explicitly call super().__init__(quiet=True) or such 27to initialize the type checking and set its debugging level. If it is not explicitly 28initialized, it will be done implicitly the first time can_process_record() or 29digest_record() are called, but with the default of quiet=False. 30""" 31 32from logger.utils.base_module import BaseModule # noqa: E402 33 34 35class Writer(BaseModule): 36 """ 37 Base class Writer about which we know nothing else. By default the 38 input format is Unknown unless overridden. 39 40 Passes arguments quiet, encoding and encoding_errors up to BaseModule 41 """ 42 ############################ 43 44 def __init__(self, **kwargs): 45 """Abstract base class for data Writers. 46 """ 47 super().__init__(**kwargs) 48 self._initialize_type_hints() 49 50 ############################ 51 def _initialize_type_hints(self): 52 """ Retrieve any type hints for child write() method so we can 53 check whether the type of record we've received can be parsed 54 natively or not.""" 55 super()._initialize_type_hints(module_type='write', 56 module_method=self.__class__.write) 57 58 ############################ 59 def write(self, record): 60 """Core method - write a record that we've been passed.""" 61 raise NotImplementedError('Class %s (subclass of Writer is missing ' 62 'implementation of write () method.' 63 % self.__class__.__name__) 64 65 66################################################################################ 67class TimestampedWriter(Writer): 68 """ 69 A TimestampedWriter is a special case of a Writer where we 70 can write out the timestamp associated with a record. 71 """ 72 73 ############################ 74 def __init__(self, **kwargs): 75 super().__init__(**kwargs) 76 77 ############################ 78 def write_timestamp(self, record, timestamp=None): 79 raise NotImplementedError('Abstract base class TimestampedWriter has no ' 80 'implementation of write_timestamp() method.')
36class Writer(BaseModule): 37 """ 38 Base class Writer about which we know nothing else. By default the 39 input format is Unknown unless overridden. 40 41 Passes arguments quiet, encoding and encoding_errors up to BaseModule 42 """ 43 ############################ 44 45 def __init__(self, **kwargs): 46 """Abstract base class for data Writers. 47 """ 48 super().__init__(**kwargs) 49 self._initialize_type_hints() 50 51 ############################ 52 def _initialize_type_hints(self): 53 """ Retrieve any type hints for child write() method so we can 54 check whether the type of record we've received can be parsed 55 natively or not.""" 56 super()._initialize_type_hints(module_type='write', 57 module_method=self.__class__.write) 58 59 ############################ 60 def write(self, record): 61 """Core method - write a record that we've been passed.""" 62 raise NotImplementedError('Class %s (subclass of Writer is missing ' 63 'implementation of write () method.' 64 % self.__class__.__name__)
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
45 def __init__(self, **kwargs): 46 """Abstract base class for data Writers. 47 """ 48 super().__init__(**kwargs) 49 self._initialize_type_hints()
Abstract base class for data Writers.
60 def write(self, record): 61 """Core method - write a record that we've been passed.""" 62 raise NotImplementedError('Class %s (subclass of Writer is missing ' 63 'implementation of write () method.' 64 % self.__class__.__name__)
Core method - write a record that we've been passed.
68class TimestampedWriter(Writer): 69 """ 70 A TimestampedWriter is a special case of a Writer where we 71 can write out the timestamp associated with a record. 72 """ 73 74 ############################ 75 def __init__(self, **kwargs): 76 super().__init__(**kwargs) 77 78 ############################ 79 def write_timestamp(self, record, timestamp=None): 80 raise NotImplementedError('Abstract base class TimestampedWriter has no ' 81 'implementation of write_timestamp() method.')
A TimestampedWriter is a special case of a Writer where we can write out the timestamp associated with a record.