[DO NOT MERGE] Prototype yapf 0.43.0 migration

.pre-commit-config.yaml

@@ -11,10 +11,10 @@
 # limitations under the License.
 
 repos:
-  - repo: https://github.com/pre-commit/mirrors-yapf
+  - repo: https://github.com/google/yapf
     # this rev is a release tag in the repo above and corresponds with a yapf
     # version. make sure this matches the version of yapf in tox.ini.
-    rev: v0.29.0
+    rev: v0.43.0
     hooks:
       - id: yapf
         files: ^sdks/python/apache_beam/

CHANGES.md

@@ -80,6 +80,7 @@
 ## Breaking Changes
 
 * X behavior was changed ([#X](https://github.com/apache/beam/issues/X)).
+* Yapf version upgraded to 0.43.0 for formatting (Python) ([#34801](https://github.com/apache/beam/pull/34801/)).
 
 ## Deprecations
 

sdks/python/apache_beam/coders/coder_impl.py

@@ -679,8 +679,7 @@ def __init__(
       self,
       key_coder,  # type: CoderImpl
       value_coder,  # type: CoderImpl
-      is_deterministic = False
-  ):
+      is_deterministic=False):
     self._key_coder = key_coder
     self._value_coder = value_coder
     self._is_deterministic = is_deterministic
@@ -1061,8 +1060,8 @@ def decode_from_stream(self, in_stream, nested):
     # type: (create_InputStream, bool) -> Any
     return self._construct_from_components([
         c.decode_from_stream(
-            in_stream, nested or i + 1 < len(self._coder_impls)) for i,
-        c in enumerate(self._coder_impls)
+            in_stream, nested or i + 1 < len(self._coder_impls))
+        for i, c in enumerate(self._coder_impls)
     ])
 
   def estimate_size(self, value, nested=False):
@@ -1890,8 +1889,7 @@ def _row_column_encoders(self, columns):
         RowColumnEncoder.create(
             self.schema.fields[i].type.atomic_type,
             self.components[i],
-            columns[name]) for i,
-        name in enumerate(self.field_names)
+            columns[name]) for i, name in enumerate(self.field_names)
     ]
 
   def encode_batch_to_stream(self, columns: Dict[str, np.ndarray], out):

sdks/python/apache_beam/coders/coders.py

@@ -382,9 +382,8 @@ def register_structured_urn(urn, cls):
     """
     setattr(
         cls,
-        'to_runner_api_parameter',
-        lambda self,
-        unused_context: (urn, None, self._get_component_coders()))
+        'to_runner_api_parameter', lambda self, unused_context:
+        (urn, None, self._get_component_coders()))
 
     # pylint: disable=unused-variable
     @Coder.register_urn(urn, None)

sdks/python/apache_beam/coders/coders_property_based_test.py

@@ -144,9 +144,7 @@ def test_row_coder(self, data: st.DataObject):
     row = RowType(
         **{
             name: data.draw(SCHEMA_TYPES_TO_STRATEGY[type_])
-            for name,
-            type_,
-            nullable in schema
+            for name, type_, nullable in schema
         })
 
     coder = RowCoder(typing_to_runner_api(RowType).row_type.schema)

sdks/python/apache_beam/coders/coders_test_common.py

@@ -163,7 +163,7 @@ def tearDownClass(cls):
         coders.BigEndianShortCoder,
         coders.SinglePrecisionFloatCoder,
         coders.ToBytesCoder,
-        coders.BigIntegerCoder, # tested in DecimalCoder
+        coders.BigIntegerCoder,  # tested in DecimalCoder
         coders.TimestampPrefixingOpaqueWindowCoder,
     ])
     cls.seen_nested -= set(
@@ -686,9 +686,14 @@ def test_nullable_coder(self):
 
   def test_map_coder(self):
     values = [
-        {1: "one", 300: "three hundred"}, # force yapf to be nice
+        {
+            1: "one", 300: "three hundred"
+        },  # force yapf to be nice
         {},
-        {i: str(i) for i in range(5000)}
+        {
+            i: str(i)
+            for i in range(5000)
+        }
     ]
     map_coder = coders.MapCoder(coders.VarIntCoder(), coders.StrUtf8Coder())
     self.check_coder(map_coder, *values)

sdks/python/apache_beam/coders/row_coder_test.py

@@ -446,8 +446,7 @@ def test_batch_encode_decode(self):
     for size in [len(self.PEOPLE) - 1, len(self.PEOPLE), len(self.PEOPLE) + 1]:
       dest = {
           field: np.ndarray((size, ), dtype=a.dtype)
-          for field,
-          a in columnar.items()
+          for field, a in columnar.items()
       }
       n = min(size, len(self.PEOPLE))
       self.assertEqual(

sdks/python/apache_beam/coders/standard_coders_test.py

@@ -139,65 +139,48 @@ class StandardCodersTest(unittest.TestCase):
       'beam:coder:bool:v1': lambda x: x,
       'beam:coder:string_utf8:v1': lambda x: x,
       'beam:coder:varint:v1': lambda x: x,
-      'beam:coder:kv:v1': lambda x,
-      key_parser,
-      value_parser: (key_parser(x['key']), value_parser(x['value'])),
+      'beam:coder:kv:v1': lambda x, key_parser, value_parser:
+      (key_parser(x['key']), value_parser(x['value'])),
       'beam:coder:interval_window:v1': lambda x: IntervalWindow(
-          start=Timestamp(micros=(x['end'] - x['span']) * 1000),
-          end=Timestamp(micros=x['end'] * 1000)),
-      'beam:coder:iterable:v1': lambda x,
-      parser: list(map(parser, x)),
-      'beam:coder:state_backed_iterable:v1': lambda x,
-      parser: list(map(parser, x)),
+          start=Timestamp(micros=(x['end'] - x['span']) * 1000), end=Timestamp(
+              micros=x['end'] * 1000)),
+      'beam:coder:iterable:v1': lambda x, parser: list(map(parser, x)),
+      'beam:coder:state_backed_iterable:v1': lambda x, parser: list(
+          map(parser, x)),
       'beam:coder:global_window:v1': lambda x: window.GlobalWindow(),
-      'beam:coder:windowed_value:v1': lambda x,
-      value_parser,
+      'beam:coder:windowed_value:v1': lambda x, value_parser, window_parser:
+      windowed_value.create(
+          value_parser(x['value']), x['timestamp'] * 1000, tuple(
+              window_parser(w) for w in x['windows'])),
+      'beam:coder:param_windowed_value:v1': lambda x, value_parser,
       window_parser: windowed_value.create(
-          value_parser(x['value']),
-          x['timestamp'] * 1000,
-          tuple(window_parser(w) for w in x['windows'])),
-      'beam:coder:param_windowed_value:v1': lambda x,
-      value_parser,
-      window_parser: windowed_value.create(
-          value_parser(x['value']),
-          x['timestamp'] * 1000,
-          tuple(window_parser(w) for w in x['windows']),
-          PaneInfo(
-              x['pane']['is_first'],
-              x['pane']['is_last'],
-              PaneInfoTiming.from_string(x['pane']['timing']),
-              x['pane']['index'],
-              x['pane']['on_time_index'])),
-      'beam:coder:timer:v1': lambda x,
-      value_parser,
-      window_parser: userstate.Timer(
-          user_key=value_parser(x['userKey']),
-          dynamic_timer_tag=x['dynamicTimerTag'],
-          clear_bit=x['clearBit'],
-          windows=tuple(window_parser(w) for w in x['windows']),
-          fire_timestamp=None,
-          hold_timestamp=None,
-          paneinfo=None) if x['clearBit'] else userstate.Timer(
-              user_key=value_parser(x['userKey']),
-              dynamic_timer_tag=x['dynamicTimerTag'],
-              clear_bit=x['clearBit'],
-              fire_timestamp=Timestamp(micros=x['fireTimestamp'] * 1000),
-              hold_timestamp=Timestamp(micros=x['holdTimestamp'] * 1000),
-              windows=tuple(window_parser(w) for w in x['windows']),
-              paneinfo=PaneInfo(
-                  x['pane']['is_first'],
-                  x['pane']['is_last'],
-                  PaneInfoTiming.from_string(x['pane']['timing']),
-                  x['pane']['index'],
-                  x['pane']['on_time_index'])),
+          value_parser(x['value']), x['timestamp'] * 1000, tuple(
+              window_parser(w) for w in x['windows']), PaneInfo(
+                  x['pane']['is_first'], x['pane']['is_last'], PaneInfoTiming.
+                  from_string(x['pane']['timing']), x['pane']['index'], x[
+                      'pane']['on_time_index'])),
+      'beam:coder:timer:v1': lambda x, value_parser, window_parser: userstate.
+      Timer(
+          user_key=value_parser(x['userKey']), dynamic_timer_tag=x[
+              'dynamicTimerTag'], clear_bit=x['clearBit'], windows=tuple(
+                  window_parser(w) for w in x['windows']), fire_timestamp=None,
+          hold_timestamp=None, paneinfo=None)
+      if x['clearBit'] else userstate.Timer(
+          user_key=value_parser(x['userKey']), dynamic_timer_tag=x[
+              'dynamicTimerTag'], clear_bit=x['clearBit'], fire_timestamp=
+          Timestamp(micros=x['fireTimestamp'] * 1000), hold_timestamp=Timestamp(
+              micros=x['holdTimestamp'] * 1000), windows=tuple(
+                  window_parser(w) for w in x['windows']), paneinfo=PaneInfo(
+                      x['pane']['is_first'], x['pane']['is_last'],
+                      PaneInfoTiming.from_string(x['pane']['timing']), x[
+                          'pane']['index'], x['pane']['on_time_index'])),
       'beam:coder:double:v1': parse_float,
-      'beam:coder:sharded_key:v1': lambda x,
-      value_parser: ShardedKey(
+      'beam:coder:sharded_key:v1': lambda x, value_parser: ShardedKey(
           key=value_parser(x['key']), shard_id=x['shardId'].encode('utf-8')),
-      'beam:coder:custom_window:v1': lambda x,
-      window_parser: window_parser(x['window']),
-      'beam:coder:nullable:v1': lambda x,
-      value_parser: x.encode('utf-8') if x else None
+      'beam:coder:custom_window:v1': lambda x, window_parser: window_parser(
+          x['window']),
+      'beam:coder:nullable:v1': lambda x, value_parser: x.encode('utf-8')
+      if x else None
   }
 
   def test_standard_coders(self):

sdks/python/apache_beam/dataframe/convert.py

@@ -257,14 +257,14 @@ def extract_input(placeholder):
         {ix: df._expr
          for (ix, df) in enumerate(new_dataframes)})
 
-    TO_PCOLLECTION_CACHE.update(
-        {new_dataframes[ix]._expr._id: pc
-         for ix, pc in new_results.items()})
+    TO_PCOLLECTION_CACHE.update({
+        new_dataframes[ix]._expr._id: pc
+        for ix, pc in new_results.items()
+    })
 
   raw_results = {
       ix: TO_PCOLLECTION_CACHE[df._expr._id]
-      for ix,
-      df in enumerate(dataframes)
+      for ix, df in enumerate(dataframes)
   }
 
   if yield_elements == "schemas":

sdks/python/apache_beam/dataframe/doctests.py

@@ -188,8 +188,7 @@ def compute_using_session(self, to_compute):
     session = expressions.PartitioningSession(self._env._inputs)
     return {
         name: session.evaluate(frame._expr)
-        for name,
-        frame in to_compute.items()
+        for name, frame in to_compute.items()
     }
 
   def compute_using_beam(self, to_compute):
@@ -198,13 +197,13 @@ def compute_using_beam(self, to_compute):
         input_pcolls = {
             placeholder: p
             | 'Create%s' % placeholder >> beam.Create([input[::2], input[1::2]])
-            for placeholder,
-            input in self._env._inputs.items()
+            for placeholder, input in self._env._inputs.items()
         }
         output_pcolls = (
-            input_pcolls | transforms._DataframeExpressionsTransform(
-                {name: frame._expr
-                 for name, frame in to_compute.items()}))
+            input_pcolls | transforms._DataframeExpressionsTransform({
+                name: frame._expr
+                for name, frame in to_compute.items()
+            }))
         for name, output_pcoll in output_pcolls.items():
           _ = output_pcoll | 'Record%s' % name >> beam.FlatMap(
               recorder.record_fn(name))
@@ -365,18 +364,15 @@ def to_callable(cond):
 
     self._wont_implement_ok = {
         test: [to_callable(cond) for cond in examples]
-        for test,
-        examples in (wont_implement_ok or {}).items()
+        for test, examples in (wont_implement_ok or {}).items()
     }
     self._not_implemented_ok = {
         test: [to_callable(cond) for cond in examples]
-        for test,
-        examples in (not_implemented_ok or {}).items()
+        for test, examples in (not_implemented_ok or {}).items()
     }
     self._skip = {
         test: [to_callable(cond) for cond in examples]
-        for test,
-        examples in (skip or {}).items()
+        for test, examples in (skip or {}).items()
     }
     super().__init__(
         checker=_DeferrredDataframeOutputChecker(self._test_env, use_beam),
@@ -541,9 +537,9 @@ def is_example_line(line):
   IMPORT_PANDAS = 'import pandas as pd'
 
   example_srcs = []
-  lines = iter([(lineno, line.rstrip()) for lineno,
-                line in enumerate(rst.split('\n')) if is_example_line(line)] +
-               [(None, 'END')])
+  lines = iter([(lineno, line.rstrip())
+                for lineno, line in enumerate(rst.split('\n'))
+                if is_example_line(line)] + [(None, 'END')])
 
   # https://ipython.readthedocs.io/en/stable/sphinxext.html
   lineno, line = next(lines)
@@ -696,12 +692,8 @@ def _run_patched(func, *args, **kwargs):
     # Unfortunately the runner is not injectable.
     original_doc_test_runner = doctest.DocTestRunner
     doctest.DocTestRunner = lambda **kwargs: BeamDataframeDoctestRunner(
-        env,
-        use_beam=use_beam,
-        wont_implement_ok=wont_implement_ok,
-        not_implemented_ok=not_implemented_ok,
-        skip=skip,
-        **kwargs)
+        env, use_beam=use_beam, wont_implement_ok=wont_implement_ok,
+        not_implemented_ok=not_implemented_ok, skip=skip, **kwargs)
     with expressions.allow_non_parallel_operations():
       return func(
           *args, extraglobs=extraglobs, optionflags=optionflags, **kwargs)