[BEAM-12000] Update deprecated collections to use collections.abc in python sdk

.pre-commit-config.yaml

@@ -42,3 +42,8 @@ repos:
         args: ["--rcfile=sdks/python/.pylintrc"]
         files: ^sdks/python/apache_beam/
         exclude: *exclude
+  - repo: https://github.com/pycqa/isort
+    rev: 5.8.0
+    hooks:
+      - id: isort
+        name: isort (python)

sdks/python/.coveragerc

@@ -16,4 +16,23 @@
 #
 
 [run]
-omit = target/*
+omit = target/*
+
+[report]
+exclude_lines =
+  # Have to re-enable the standard pragma
+  pragma: no cover
+  abc.abstractmethod
+
+  # Overload stubs should never be executed.
+  @overload
+
+  # Don't complain about missing debug-only code:
+  def __repr__
+  if self\.debug
+
+  # Don't complain if tests don't hit defensive assertion code:
+  raise NotImplementedError
+
+  # Don't complain if non-runnable code isn't run:
+  if __name__ == .__main__.:

sdks/python/apache_beam/runners/direct/transform_evaluator.py

@@ -21,6 +21,7 @@
 
 import atexit
 import collections
+import collections.abc
 import logging
 import random
 import time
@@ -936,7 +937,7 @@ def process_element(self, element):
     assert not self.global_state.get_state(
         None, _GroupByKeyOnlyEvaluator.COMPLETION_TAG)
     if (isinstance(element, WindowedValue) and
-        isinstance(element.value, collections.Iterable) and
+        isinstance(element.value, collections.abc.Iterable) and
         len(element.value) == 2):
       k, v = element.value
       encoded_k = self.key_coder.encode(k)
@@ -1025,7 +1026,7 @@ def start_bundle(self):
 
   def process_element(self, element):
     if (isinstance(element, WindowedValue) and
-        isinstance(element.value, collections.Iterable) and
+        isinstance(element.value, collections.abc.Iterable) and
         len(element.value) == 2):
       k, v = element.value
       self.gbk_items[self.key_coder.encode(k)].append(v)

sdks/python/apache_beam/runners/worker/sideinputs.py

@@ -20,6 +20,7 @@
 # pytype: skip-file
 
 import collections
+import collections.abc
 import logging
 import queue
 import threading
@@ -207,7 +208,7 @@ def _inner():
   return _inner
 
 
-class EmulatedIterable(collections.Iterable):
+class EmulatedIterable(collections.abc.Iterable):
   """Emulates an iterable for a side input."""
   def __init__(self, iterator_fn):
     self.iterator_fn = iterator_fn

sdks/python/apache_beam/transforms/trigger.py

@@ -23,6 +23,7 @@
 # pytype: skip-file
 
 import collections
+import collections.abc
 import copy
 import logging
 import numbers
@@ -1259,7 +1260,7 @@ def __reduce__(self):
     return list, (list(self), )
 
   def __eq__(self, other):
-    if isinstance(other, collections.Iterable):
+    if isinstance(other, collections.abc.Iterable):
       return all(
           a == b for a, b in zip_longest(self, other, fillvalue=object()))
     else:

sdks/python/apache_beam/typehints/typecheck.py

@@ -23,6 +23,7 @@
 # pytype: skip-file
 
 import collections
+import collections.abc
 import inspect
 import sys
 import types
@@ -105,7 +106,7 @@ def _check_type(output):
           'Returning a %s from a ParDo or FlatMap is '
           'discouraged. Please use list("%s") if you really '
           'want this behavior.' % (object_type, output))
-    elif not isinstance(output, collections.Iterable):
+    elif not isinstance(output, collections.abc.Iterable):
       raise TypeCheckError(
           'FlatMap and ParDo must return an '
           'iterable. %s was returned instead.' % type(output))

sdks/python/apache_beam/typehints/typehints.py

@@ -66,6 +66,7 @@
 # pytype: skip-file
 
 import collections
+import collections.abc
 import copy
 import logging
 import typing
@@ -532,7 +533,7 @@ def type_check(self, instance):
               error_msg))
 
   def __getitem__(self, type_params):
-    if not isinstance(type_params, (collections.Sequence, set)):
+    if not isinstance(type_params, (collections.abc.Sequence, set)):
       raise TypeError('Cannot create Union without a sequence of types.')
 
     # Flatten nested Union's and duplicated repeated type hints.
@@ -573,7 +574,7 @@ class OptionalHint(UnionHint):
   """
   def __getitem__(self, py_type):
     # A single type must have been passed.
-    if isinstance(py_type, collections.Sequence):
+    if isinstance(py_type, collections.abc.Sequence):
       raise TypeError(
           'An Option type-hint only accepts a single type '
           'parameter.')
@@ -693,7 +694,7 @@ def bind_type_variables(self, bindings):
   def __getitem__(self, type_params):
     ellipsis = False
 
-    if not isinstance(type_params, collections.Iterable):
+    if not isinstance(type_params, collections.abc.Iterable):
       # Special case for hinting tuples with arity-1.
       type_params = (type_params, )
 
@@ -962,7 +963,7 @@ class IterableHint(CompositeTypeHint):
   class IterableTypeConstraint(SequenceTypeConstraint):
     def __init__(self, iter_type):
       super(IterableHint.IterableTypeConstraint,
-            self).__init__(iter_type, collections.Iterable)
+            self).__init__(iter_type, collections.abc.Iterable)
 
     def __repr__(self):
       return 'Iterable[%s]' % _unified_repr(self.inner_type)

sdks/python/apache_beam/utils/proto_utils.py

@@ -40,13 +40,13 @@
 @overload
 def pack_Any(msg):
   # type: (message.Message) -> any_pb2.Any
-  pass
+  ...
 
 
 @overload
 def pack_Any(msg):
   # type: (None) -> None
-  pass
+  ...
 
 
 def pack_Any(msg):
@@ -65,13 +65,13 @@ def pack_Any(msg):
 @overload
 def unpack_Any(any_msg, msg_class):
   # type: (any_pb2.Any, Type[MessageT]) -> MessageT
-  pass
+  ...
 
 
 @overload
 def unpack_Any(any_msg, msg_class):
   # type: (any_pb2.Any, None) -> None
-  pass
+  ...
 
 
 def unpack_Any(any_msg, msg_class):
@@ -89,13 +89,13 @@ def unpack_Any(any_msg, msg_class):
 @overload
 def parse_Bytes(serialized_bytes, msg_class):
   # type: (bytes, Type[MessageT]) -> MessageT
-  pass
+  ...
 
 
 @overload
 def parse_Bytes(serialized_bytes, msg_class):
   # type: (bytes, Union[Type[bytes], None]) -> bytes
-  pass
+  ...
 
 
 def parse_Bytes(serialized_bytes, msg_class):

sdks/python/apache_beam/utils/proto_utils_test.py

@@ -0,0 +1,25 @@
+import unittest
+from google.protobuf import timestamp_pb2
+
+from apache_beam.utils.proto_utils import pack_Any
+from apache_beam.utils.proto_utils import to_Timestamp
+from apache_beam.utils.proto_utils import unpack_Any
+
+
+class ProtoUtilsTest(unittest.TestCase):
+  def make_proto_timestamp(self):
+    # type: () -> timestamp_pb2.Timestamp
+    return to_Timestamp(0)
+
+  def test_none_pack(self):
+    packed_none = pack_Any(None)
+    assert packed_none is None
+
+  def test_date_pack(self):
+    # type: () -> None
+    proto_timestamp = self.make_proto_timestamp()
+    packed_msg = pack_Any(proto_timestamp)
+    orig_msg = unpack_Any(packed_msg, timestamp_pb2.Timestamp)
+    none_msg = unpack_Any(packed_msg, None)
+    assert proto_timestamp == orig_msg
+    assert none_msg is None

sdks/python/setup.cfg

@@ -39,6 +39,9 @@ exclude_lines =
   pragma: no cover
   abc.abstractmethod
 
+  # Overload stubs should never be executed.
+  @overload
+
   # Don't complain about missing debug-only code:
   def __repr__
   if self\.debug
@@ -53,7 +56,11 @@ exclude_lines =
 output = target/site/cobertura/coverage.xml
 
 [isort]
+line_length = 120
 known_standard_library = dataclasses
+force_single_line = True
+combine_star = True
+order_by_type = True
 
 [yapf]
 indent_width = 2