openrvdas.logger.transforms.from_json_transform

No module-level documentation available.
 1#!/usr/bin/env python3
 2
 3import json
 4import logging
 5from typing import Union
 6
 7from logger.utils.das_record import DASRecord  # noqa: E402
 8from logger.transforms.transform import Transform  # noqa: E402
 9
10
11################################################################################
12#
13class FromJSONTransform(Transform):
14    """Convert passed JSON to either a DASRecord or a dict.
15    """
16
17    def __init__(self, das_record=False, **kwargs):
18        """Parse the received JSON and convert to appropriate data
19        structure. If das_record == True, assume we've been passed a dict
20        of field:value pairs, and try to embed them into a DASRecord.
21        """
22        super().__init__(**kwargs)  # processes 'quiet' and type hints
23        self.das_record = das_record
24
25    ############################
26    def transform(self, record: str) -> Union[dict, DASRecord]:
27        """Parse JSON record to Python data struct or DASRecord."""
28
29        # See if it's something we can process, and if not, try digesting
30        if not self.can_process_record(record):  # inherited from BaseModule()
31            return self.digest_record(record)  # inherited from BaseModule()
32
33        try:
34            data = json.loads(record)
35        except json.decoder.JSONDecodeError:
36            logging.warning('Failed to parse JSON string: "%s"', record)
37            return None
38
39        if not self.das_record:
40            return data
41
42        if not type(data) is dict:
43            logging.warning('FromJSON asked to create DASRecord from non-dict '
44                            'data: "%s"', type(data))
45            return None
46
47        return DASRecord(fields=data)
class FromJSONTransform(logger.transforms.transform.Transform): 
14class FromJSONTransform(Transform):
15    """Convert passed JSON to either a DASRecord or a dict.
16    """
17
18    def __init__(self, das_record=False, **kwargs):
19        """Parse the received JSON and convert to appropriate data
20        structure. If das_record == True, assume we've been passed a dict
21        of field:value pairs, and try to embed them into a DASRecord.
22        """
23        super().__init__(**kwargs)  # processes 'quiet' and type hints
24        self.das_record = das_record
25
26    ############################
27    def transform(self, record: str) -> Union[dict, DASRecord]:
28        """Parse JSON record to Python data struct or DASRecord."""
29
30        # See if it's something we can process, and if not, try digesting
31        if not self.can_process_record(record):  # inherited from BaseModule()
32            return self.digest_record(record)  # inherited from BaseModule()
33
34        try:
35            data = json.loads(record)
36        except json.decoder.JSONDecodeError:
37            logging.warning('Failed to parse JSON string: "%s"', record)
38            return None
39
40        if not self.das_record:
41            return data
42
43        if not type(data) is dict:
44            logging.warning('FromJSON asked to create DASRecord from non-dict '
45                            'data: "%s"', type(data))
46            return None
47
48        return DASRecord(fields=data)

Convert passed JSON to either a DASRecord or a dict.

FromJSONTransform(das_record=False, **kwargs) 
18    def __init__(self, das_record=False, **kwargs):
19        """Parse the received JSON and convert to appropriate data
20        structure. If das_record == True, assume we've been passed a dict
21        of field:value pairs, and try to embed them into a DASRecord.
22        """
23        super().__init__(**kwargs)  # processes 'quiet' and type hints
24        self.das_record = das_record

Parse the received JSON and convert to appropriate data structure. If das_record == True, assume we've been passed a dict of field:value pairs, and try to embed them into a DASRecord.

das_record
def transform(self, record: str) -> Union[dict, logger.utils.das_record.DASRecord]: 
27    def transform(self, record: str) -> Union[dict, DASRecord]:
28        """Parse JSON record to Python data struct or DASRecord."""
29
30        # See if it's something we can process, and if not, try digesting
31        if not self.can_process_record(record):  # inherited from BaseModule()
32            return self.digest_record(record)  # inherited from BaseModule()
33
34        try:
35            data = json.loads(record)
36        except json.decoder.JSONDecodeError:
37            logging.warning('Failed to parse JSON string: "%s"', record)
38            return None
39
40        if not self.das_record:
41            return data
42
43        if not type(data) is dict:
44            logging.warning('FromJSON asked to create DASRecord from non-dict '
45                            'data: "%s"', type(data))
46            return None
47
48        return DASRecord(fields=data)

Parse JSON record to Python data struct or DASRecord.