File tree 1 file changed +11
-9
lines changed
1 file changed +11
-9
lines changed Original file line number Diff line number Diff line change @@ -611,7 +611,7 @@ def _after_process_record(self, context: dict) -> None:
611
611
612
612
# SDK developer overrides:
613
613
614
- def preprocess_record (self , record : dict , context : dict ) -> dict : # noqa: ARG002, PLR6301
614
+ def preprocess_record (self , record : dict , context : dict ) -> dict : # noqa: PLR6301, ARG002
615
615
"""Process incoming record and return a modified result.
616
616
617
617
Args:
@@ -743,12 +743,15 @@ def process_batch_files(
743
743
tail ,
744
744
mode = "rb" ,
745
745
) as file :
746
- context_file = (
747
- gzip_open (file ) if encoding .compression == "gzip" else file
748
- )
749
- context = {
750
- "records" : [deserialize_json (line ) for line in context_file ] # type: ignore[attr-defined]
751
- }
746
+ if encoding .compression == "gzip" :
747
+ with gzip_open (file ) as context_file :
748
+ context = {
749
+ "records" : [
750
+ deserialize_json (line ) for line in context_file
751
+ ]
752
+ }
753
+ else :
754
+ context = {"records" : [deserialize_json (line ) for line in file ]}
752
755
self .process_batch (context )
753
756
elif (
754
757
importlib .util .find_spec ("pyarrow" )
@@ -760,8 +763,7 @@ def process_batch_files(
760
763
tail ,
761
764
mode = "rb" ,
762
765
) as file :
763
- context_file = file
764
- table = pq .read_table (context_file )
766
+ table = pq .read_table (file )
765
767
context = {"records" : table .to_pylist ()}
766
768
self .process_batch (context )
767
769
else :
You can’t perform that action at this time.
0 commit comments