21 lines
883 B
Python
21 lines
883 B
Python
from typing import Union
|
|
|
|
from pyinfra.parser.parser_composer import EitherParserComposer
|
|
from pyinfra.parser.parsers.identity import IdentityBlobParser
|
|
from pyinfra.parser.parsers.json import JsonBlobParser
|
|
from pyinfra.parser.parsers.string import StringBlobParser
|
|
from pyinfra.visitor.strategies.parsing.parsing import ParsingStrategy
|
|
|
|
|
|
# TODO: Each analysis service should specify a custom parsing strategy for the type of data it expects to be found
|
|
# on the storage. This class is only a temporary trial-and-error->fallback type of solution.
|
|
class DynamicParsingStrategy(ParsingStrategy):
|
|
def __init__(self):
|
|
self.parser = EitherParserComposer(JsonBlobParser(), StringBlobParser(), IdentityBlobParser())
|
|
|
|
def parse(self, data: bytes) -> Union[bytes, dict]:
|
|
return self.parser(data)
|
|
|
|
def parse_and_wrap(self, data):
|
|
return self.parse(data)
|