replaced trial and error logic for parsing blobs in visitor with parser composer instance

pyinfra/visitor.py

@@ -13,6 +13,10 @@ from more_itertools import peekable
 
 from pyinfra.config import CONFIG, parse_disjunction_string
 from pyinfra.exceptions import DataLoadingFailure
+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.server.packing import string_to_bytes
 from pyinfra.storage.storage import Storage
 
@@ -173,29 +177,12 @@ class ParsingStrategy(abc.ABC):
 
 
 class DynamicParsingStrategy(ParsingStrategy):
-    @staticmethod
-    def attempt_parse_json(data):
-        data = json.loads(data)
-        if "data" in data:
-            data["data"] = string_to_bytes(data["data"])
-        return data
 
-    @staticmethod
-    def fallback(data):
-        return data
+    def __init__(self):
+        self.parser = EitherParserComposer(JsonBlobParser(), StringBlobParser(), IdentityBlobParser())
 
     def parse(self, data: bytes) -> Union[bytes, dict]:
-        try:
-            data = data.decode()
-        except UnicodeDecodeError:
-            return self.fallback(data)
-
-        try:
-            return self.attempt_parse_json(data)
-        except json.JSONDecodeError:
-            pass
-
-        return self.fallback(data)
+        return self.parser(data)
 
     def parse_and_wrap(self, data):
         """Storage items can be a blob or a blob with metadata. Standardizes to the latter.
@@ -255,7 +242,7 @@ class QueueVisitor:
 
         return data
 
-    def standardize(self, data: bytes, queue_item_body) -> Dict:
+    def standardize(self, data: bytes) -> Dict:
         assert isinstance(data, bytes)
         data = self.parsing_strategy(data)
         return data
@@ -266,7 +253,7 @@ class QueueVisitor:
         data = self.download(object_descriptor)
         logging.debug(f"Downloaded {object_descriptor}.")
         data = gzip.decompress(data)
-        data = self.standardize(data, queue_item_body)
+        data = self.standardize(data)
         return data
 
     def process_storage_item(self, data_metadata_pack):

test/parser/parser_test.py

@@ -29,6 +29,6 @@ def test_either_parser_composer():
 
     a = "a"
     assert parser(a.encode()) == a
-    
+
     a = 1
     assert parser(a) == a