openrvdas.logger.transforms.extract_field_transform

No module-level documentation available.
 1#!/usr/bin/env python3
 2
 3import logging
 4from typing import Union
 5
 6from logger.utils.das_record import DASRecord  # noqa: E402
 7from logger.transforms.transform import Transform  # noqa: E402
 8
 9
10################################################################################
11#
12class ExtractFieldTransform(Transform):
13    """Extract a field from passed DASRecord or dict.
14    """
15
16    def __init__(self, field_name, **kwargs):
17        """Extract the specified field from the passed DASRecord or dict.
18        """
19        super().__init__(**kwargs)  # processes 'quiet' and type hints
20
21        self.field_name = field_name
22
23    ############################
24    def transform(self, record: Union[DASRecord, dict]):
25        """Extract the specified field from the passed DASRecord or dict.
26        """
27        # See if it's something we can process, and if not, try digesting
28        if not self.can_process_record(record):  # BaseModule
29            return self.digest_record(record)  # BaseModule
30
31        if type(record) is DASRecord:
32            return record.fields.get(self.field_name)
33        elif type(record) is dict:
34            fields = record.get('fields')
35            if not fields:
36                return None
37            return fields.get(self.field_name)
38
39        logging.warning('ExtractFieldTransform found no field "%s" in record "%s"',
40                        self.field_name, record)
41        return None
class ExtractFieldTransform(logger.transforms.transform.Transform): 
13class ExtractFieldTransform(Transform):
14    """Extract a field from passed DASRecord or dict.
15    """
16
17    def __init__(self, field_name, **kwargs):
18        """Extract the specified field from the passed DASRecord or dict.
19        """
20        super().__init__(**kwargs)  # processes 'quiet' and type hints
21
22        self.field_name = field_name
23
24    ############################
25    def transform(self, record: Union[DASRecord, dict]):
26        """Extract the specified field from the passed DASRecord or dict.
27        """
28        # See if it's something we can process, and if not, try digesting
29        if not self.can_process_record(record):  # BaseModule
30            return self.digest_record(record)  # BaseModule
31
32        if type(record) is DASRecord:
33            return record.fields.get(self.field_name)
34        elif type(record) is dict:
35            fields = record.get('fields')
36            if not fields:
37                return None
38            return fields.get(self.field_name)
39
40        logging.warning('ExtractFieldTransform found no field "%s" in record "%s"',
41                        self.field_name, record)
42        return None

Extract a field from passed DASRecord or dict.

ExtractFieldTransform(field_name, **kwargs) 
17    def __init__(self, field_name, **kwargs):
18        """Extract the specified field from the passed DASRecord or dict.
19        """
20        super().__init__(**kwargs)  # processes 'quiet' and type hints
21
22        self.field_name = field_name

Extract the specified field from the passed DASRecord or dict.

field_name
def transform(self, record: Union[logger.utils.das_record.DASRecord, dict]): 
25    def transform(self, record: Union[DASRecord, dict]):
26        """Extract the specified field from the passed DASRecord or dict.
27        """
28        # See if it's something we can process, and if not, try digesting
29        if not self.can_process_record(record):  # BaseModule
30            return self.digest_record(record)  # BaseModule
31
32        if type(record) is DASRecord:
33            return record.fields.get(self.field_name)
34        elif type(record) is dict:
35            fields = record.get('fields')
36            if not fields:
37                return None
38            return fields.get(self.field_name)
39
40        logging.warning('ExtractFieldTransform found no field "%s" in record "%s"',
41                        self.field_name, record)
42        return None

Extract the specified field from the passed DASRecord or dict.