Handle bytea[] types on p2p

config.yaml

@@ -53,11 +53,12 @@ jobs:
     source:
       ref: PG
       query_string: |
-        SELECT 
-          1 AS number, 
+        SELECT
+          1 AS number,
           '\\x1234'::bytea AS my_bytes,
-          '{"key": "value", "array": [1, 2, 3], "dict": {}}'::json AS my_json,
-          '[ [{"x": 1}, {"y": 2}], null, [{"z": 3}] ]'::json AS list_dict
+          ARRAY['\\x1234'::bytea, '\\x5678'::bytea] AS array_bytea,
+          '{"key": "value", "array": [1, 2, 3], "dict": {}}'::json AS my_json
+          -- '[ [{"x": 1}, {"y": 2}], null, [{"z": 3}] ]'::json AS list_dict
 
     destination:
       ref: PG

src/sources/postgres.py

@@ -24,7 +24,7 @@ def _convert_dict_to_json(df: DataFrame) -> DataFrame:
         non_null_values = df[column][df[column].notna()]
         if len(non_null_values) > 0:
             first_value = non_null_values.iloc[0]
-            if isinstance(first_value, dict | list):
+            if isinstance(first_value, dict):
                 df[column] = df[column].apply(
                     lambda x: json.dumps(x) if x is not None else None
                 )
@@ -57,6 +57,10 @@ def _convert_bytea_to_hex(df: DataFrame) -> DataFrame:
     for column in df.columns:
         if isinstance(df[column].iloc[0], memoryview):
             df[column] = df[column].apply(lambda x: f"0x{x.tobytes().hex()}")
+        # if isinstance(df[column].iloc[0], list):
+        #     # Check if the list contains memoryview objects
+        #     if all(isinstance(item, memoryview) for item in df[column].iloc[0]):
+        #         df[column] = df[column].apply(lambda lst: [f"0x{item.tobytes().hex()}" for item in lst])
     return df
 
 
@@ -140,9 +144,8 @@ async def fetch(self) -> TypedDataFrame:
         df = await loop.run_in_executor(
             None, lambda: pd.read_sql_query(self.query_string, con=self.engine)
         )
-
-        df = _convert_dict_to_json(df)
         df = _convert_bytea_to_hex(df)
+        df = _convert_dict_to_json(df)
         # TODO include types.
         return TypedDataFrame(dataframe=df, types={})