@@ -2775,6 +2775,20 @@ def expand(self, input):
27752775 failed_rows = failed_rows ,
27762776 failed_rows_with_errors = failed_rows_with_errors )
27772777
2778+ class ConvertToBeamRowsSetupSchema :
2779+ def __init__ (self , schema ):
2780+ self ._value = schema
2781+
2782+ def __enter__ (self ):
2783+ if not isinstance (self ._value ,
2784+ (bigquery .TableSchema , bigquery .TableFieldSchema )):
2785+ return bigquery_tools .get_bq_tableschema (self ._value )
2786+
2787+ return self ._value
2788+
2789+ def __exit__ (self , * args ):
2790+ pass
2791+
27782792 class ConvertToBeamRows (PTransform ):
27792793 def __init__ (self , schema , dynamic_destinations ):
27802794 self .schema = schema
@@ -2785,18 +2799,22 @@ def expand(self, input_dicts):
27852799 return (
27862800 input_dicts
27872801 | "Convert dict to Beam Row" >> beam .Map (
2788- lambda row : beam .Row (
2789- ** {
2790- StorageWriteToBigQuery .DESTINATION : row [
2791- 0 ], StorageWriteToBigQuery .RECORD : bigquery_tools .
2792- beam_row_from_dict (row [1 ], self .schema )
2793- })))
2802+ lambda row , schema = DoFn .SetupContextParam (
2803+ StorageWriteToBigQuery .ConvertToBeamRowsSetupSchema , args =
2804+ [self .schema ]): beam .Row (
2805+ ** {
2806+ StorageWriteToBigQuery .DESTINATION : row [0 ],
2807+ StorageWriteToBigQuery .RECORD : bigquery_tools .
2808+ beam_row_from_dict (row [1 ], schema )
2809+ })))
27942810 else :
27952811 return (
27962812 input_dicts
27972813 | "Convert dict to Beam Row" >> beam .Map (
2798- lambda row : bigquery_tools .beam_row_from_dict (row , self .schema ))
2799- )
2814+ lambda row , schema = DoFn .SetupContextParam (
2815+ StorageWriteToBigQuery .ConvertToBeamRowsSetupSchema , args = [
2816+ self .schema
2817+ ]): bigquery_tools .beam_row_from_dict (row , schema )))
28002818
28012819 def with_output_types (self ):
28022820 row_type_hints = bigquery_tools .get_beam_typehints_from_tableschema (
0 commit comments