@@ -187,7 +187,6 @@ def __call__(self, request: Union[beam.Row, list[beam.Row]], *args, **kwargs):
187187 if isinstance (request , list ):
188188 responses = []
189189 requests_map : dict [Any , Any ] = {}
190- batch_size = len (request )
191190 if self .fields and len (self .fields ) > 0 :
192191 unique_values = set ()
193192 field_name = self .fields [0 ]
@@ -196,10 +195,9 @@ def __call__(self, request: Union[beam.Row, list[beam.Row]], *args, **kwargs):
196195 unique_values .add (req_dict [field_name ])
197196 if unique_values :
198197 conditions = [f"{ field_name } = '{ val } '" for val in unique_values ]
199- raw_query = "SELECT %s FROM %s WHERE %s" % (
200- self .select_fields ,
201- self .table_name ,
202- " OR " .join (conditions ))
198+ raw_query = "SELECT %s FROM %s WHERE %s" % (self .select_fields ,
199+ self .table_name ,
200+ " OR " .join (conditions ))
203201 else :
204202 raw_query = self .query_template
205203 else :
@@ -231,7 +229,6 @@ def __call__(self, request: Union[beam.Row, list[beam.Row]], *args, **kwargs):
231229 responses .append ((req , beam .Row ()))
232230 return responses
233231 else :
234- request_dict = request ._asdict ()
235232 if self .query_fn :
236233 # if a query_fn is provided then it return a list of values
237234 # that should be populated into the query template string.
0 commit comments