Upgrade mypy to version 1.13.0 (apache#32978)

* Upgrade mypy to version 1.13.0

* formatting, yaml io fix

sdks/python/apache_beam/coders/coder_impl.py

@@ -1975,7 +1975,7 @@ class DecimalCoderImpl(StreamCoderImpl):
 
   def encode_to_stream(self, value, out, nested):
     # type: (decimal.Decimal, create_OutputStream, bool) -> None
-    scale = -value.as_tuple().exponent
+    scale = -value.as_tuple().exponent  # type: ignore[operator]
     int_value = int(value.scaleb(scale))
     out.write_var_int64(scale)
     self.BIG_INT_CODER_IMPL.encode_to_stream(int_value, out, nested)

sdks/python/apache_beam/coders/coders_property_based_test.py

@@ -141,7 +141,7 @@ def test_row_coder(self, data: st.DataObject):
     coders_registry.register_coder(RowType, RowCoder)
 
     # TODO(https://github.com/apache/beam/issues/23002): Apply nulls for these
-    row = RowType(  # type: ignore
+    row = RowType(
         **{
             name: data.draw(SCHEMA_TYPES_TO_STRATEGY[type_])
             for name,

sdks/python/apache_beam/coders/coders_test_common.py

@@ -53,7 +53,7 @@
 except ImportError:
   dataclasses = None  # type: ignore
 
-MyNamedTuple = collections.namedtuple('A', ['x', 'y'])
+MyNamedTuple = collections.namedtuple('A', ['x', 'y'])  # type: ignore[name-match]
 MyTypedNamedTuple = NamedTuple('MyTypedNamedTuple', [('f1', int), ('f2', str)])
 
 

sdks/python/apache_beam/metrics/metric.py

@@ -140,7 +140,7 @@ class DelegatingCounter(Counter):
     def __init__(
         self, metric_name: MetricName, process_wide: bool = False) -> None:
       super().__init__(metric_name)
-      self.inc = MetricUpdater(  # type: ignore[assignment]
+      self.inc = MetricUpdater(  # type: ignore[method-assign]
           cells.CounterCell,
           metric_name,
           default_value=1,
@@ -150,19 +150,19 @@ class DelegatingDistribution(Distribution):
     """Metrics Distribution Delegates functionality to MetricsEnvironment."""
     def __init__(self, metric_name: MetricName) -> None:
       super().__init__(metric_name)
-      self.update = MetricUpdater(cells.DistributionCell, metric_name)  # type: ignore[assignment]
+      self.update = MetricUpdater(cells.DistributionCell, metric_name)  # type: ignore[method-assign]
 
   class DelegatingGauge(Gauge):
     """Metrics Gauge that Delegates functionality to MetricsEnvironment."""
     def __init__(self, metric_name: MetricName) -> None:
       super().__init__(metric_name)
-      self.set = MetricUpdater(cells.GaugeCell, metric_name)  # type: ignore[assignment]
+      self.set = MetricUpdater(cells.GaugeCell, metric_name)  # type: ignore[method-assign]
 
   class DelegatingStringSet(StringSet):
     """Metrics StringSet that Delegates functionality to MetricsEnvironment."""
     def __init__(self, metric_name: MetricName) -> None:
       super().__init__(metric_name)
-      self.add = MetricUpdater(cells.StringSetCell, metric_name)  # type: ignore[assignment]
+      self.add = MetricUpdater(cells.StringSetCell, metric_name)  # type: ignore[method-assign]
 
 
 class MetricResults(object):

sdks/python/apache_beam/ml/inference/xgboost_inference_test.py

@@ -53,7 +53,7 @@ def _compare_prediction_result(a: PredictionResult, b: PredictionResult):
     example_equal = numpy.array_equal(a.example.todense(), b.example.todense())
 
   else:
-    example_equal = numpy.array_equal(a.example, b.example)
+    example_equal = numpy.array_equal(a.example, b.example)  # type: ignore[arg-type]
   if isinstance(a.inference, dict):
     return all(
         x == y for x, y in zip(a.inference.values(),

sdks/python/apache_beam/ml/transforms/base_test.py

@@ -345,7 +345,7 @@ def __init__(self, columns, **kwargs):
     super().__init__(columns=columns, **kwargs)
 
   def get_model_handler(self) -> ModelHandler:
-    FakeModelHandler.__repr__ = lambda x: 'FakeEmbeddingsManager'  # type: ignore[assignment]
+    FakeModelHandler.__repr__ = lambda x: 'FakeEmbeddingsManager'  # type: ignore[method-assign]
     return FakeModelHandler()
 
   def get_ptransform_for_processing(self, **kwargs) -> beam.PTransform:
@@ -532,7 +532,7 @@ def __init__(self, columns, **kwargs):
     super().__init__(columns=columns, **kwargs)
 
   def get_model_handler(self) -> ModelHandler:
-    FakeModelHandler.__repr__ = lambda x: 'FakeImageEmbeddingsManager'  # type: ignore[assignment]
+    FakeModelHandler.__repr__ = lambda x: 'FakeImageEmbeddingsManager'  # type: ignore[method-assign]
     return FakeImageModelHandler()
 
   def get_ptransform_for_processing(self, **kwargs) -> beam.PTransform:

sdks/python/apache_beam/runners/dataflow/internal/clients/dataflow/__init__.py

@@ -30,4 +30,4 @@
   pass
 # pylint: enable=wrong-import-order, wrong-import-position
 
-__path__ = pkgutil.extend_path(__path__, __name__)  # type: ignore
+__path__ = pkgutil.extend_path(__path__, __name__)

sdks/python/apache_beam/runners/pipeline_context.py

@@ -306,7 +306,7 @@ def get_or_create_environment_with_resource_hints(
       #  "Message"; expected "Environment"  [arg-type]
       # Here, Environment is a subclass of Message but mypy still
       # throws an error.
-      cloned_env.CopyFrom(template_env)  # type: ignore[arg-type]
+      cloned_env.CopyFrom(template_env)
       cloned_env.resource_hints.clear()
       cloned_env.resource_hints.update(resource_hints)
 

sdks/python/apache_beam/runners/portability/fn_api_runner/translations.py

@@ -256,7 +256,7 @@ def has_as_main_input(self, pcoll):
             transform.spec.payload, beam_runner_api_pb2.ParDoPayload)
         local_side_inputs = payload.side_inputs
       else:
-        local_side_inputs = {}  # type: ignore[assignment]
+        local_side_inputs = {}
       for local_id, pipeline_id in transform.inputs.items():
         if pcoll == pipeline_id and local_id not in local_side_inputs:
           return True

sdks/python/apache_beam/runners/portability/fn_api_runner/worker_handlers.py

@@ -1071,7 +1071,7 @@ def get_raw(self,
 
     if state_key.WhichOneof('type') not in self._SUPPORTED_STATE_TYPES:
       raise NotImplementedError(
-          'Unknown state type: ' + state_key.WhichOneof('type'))
+          'Unknown state type: ' + state_key.WhichOneof('type'))  # type: ignore[operator]
 
     with self._lock:
       if not continuation_token:

sdks/python/apache_beam/runners/portability/local_job_service.py

@@ -35,7 +35,7 @@
 import grpc
 from google.protobuf import json_format
 from google.protobuf import struct_pb2
-from google.protobuf import text_format  # type: ignore # not in typeshed
+from google.protobuf import text_format
 
 from apache_beam import pipeline
 from apache_beam.metrics import monitoring_infos

sdks/python/apache_beam/runners/render.py

@@ -64,7 +64,7 @@
 import urllib.parse
 
 from google.protobuf import json_format
-from google.protobuf import text_format  # type: ignore
+from google.protobuf import text_format
 
 from apache_beam.options import pipeline_options
 from apache_beam.portability.api import beam_runner_api_pb2

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

@@ -125,7 +125,7 @@ def emit(self, record: logging.LogRecord) -> None:
       log_entry.message = (
           "Failed to format '%s' with args '%s' during logging." %
           (str(record.msg), record.args))
-    log_entry.thread = record.threadName
+    log_entry.thread = record.threadName  # type: ignore[assignment]
     log_entry.log_location = '%s:%s' % (
         record.pathname or record.module, record.lineno or record.funcName)
     (fraction, seconds) = math.modf(record.created)

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

@@ -259,7 +259,7 @@ def do_sample(self, windowed_value):
     self.type_check(windowed_value.value)
 
     size, observables = (
-        self.coder_impl.get_estimated_size_and_observables(windowed_value))
+        self.coder_impl.get_estimated_size_and_observables(windowed_value))  # type: ignore[union-attr]
     if not observables:
       self.current_size = size
     else:

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

@@ -1335,8 +1335,8 @@ def _get_cache_token(self, state_key):
         return self._context.user_state_cache_token
       else:
         return self._context.bundle_cache_token
-    elif state_key.WhichOneof('type').endswith('_side_input'):
-      side_input = getattr(state_key, state_key.WhichOneof('type'))
+    elif state_key.WhichOneof('type').endswith('_side_input'):  # type: ignore[union-attr]
+      side_input = getattr(state_key, state_key.WhichOneof('type'))  # type: ignore[arg-type]
       return self._context.side_input_cache_tokens.get(
           (side_input.transform_id, side_input.side_input_id),
           self._context.bundle_cache_token)

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

@@ -27,7 +27,7 @@
 import sys
 import traceback
 
-from google.protobuf import text_format  # type: ignore # not in typeshed
+from google.protobuf import text_format
 
 from apache_beam.internal import pickler
 from apache_beam.io import filesystems

sdks/python/apache_beam/testing/benchmarks/chicago_taxi/tfdv_analyze_and_validate.py

@@ -29,7 +29,7 @@
 from apache_beam.testing.load_tests.load_test_metrics_utils import MeasureTime
 from apache_beam.testing.load_tests.load_test_metrics_utils import MetricsReader
 
-from google.protobuf import text_format  # type: ignore  # typeshed out of date
+from google.protobuf import text_format
 from trainer import taxi
 
 

sdks/python/apache_beam/testing/benchmarks/chicago_taxi/trainer/taxi.py

@@ -18,7 +18,7 @@
 from tensorflow_transform import coders as tft_coders
 from tensorflow_transform.tf_metadata import schema_utils
 
-from google.protobuf import text_format  # type: ignore  # typeshed out of date
+from google.protobuf import text_format
 from tensorflow.python.lib.io import file_io
 from tensorflow_metadata.proto.v0 import schema_pb2
 

sdks/python/apache_beam/testing/load_tests/sideinput_test.py

@@ -111,7 +111,7 @@ class SequenceSideInputTestDoFn(beam.DoFn):
       def __init__(self, first_n: int):
         self._first_n = first_n
 
-      def process(  # type: ignore[override]
+      def process(
           self, element: Any, side_input: Iterable[Tuple[bytes,
                                                          bytes]]) -> None:
         i = 0
@@ -129,7 +129,7 @@ class MappingSideInputTestDoFn(beam.DoFn):
       def __init__(self, first_n: int):
         self._first_n = first_n
 
-      def process(  # type: ignore[override]
+      def process(
           self, element: Any, dict_side_input: Dict[bytes, bytes]) -> None:
         i = 0
         for key in dict_side_input:
@@ -146,7 +146,7 @@ def __init__(self):
         # Avoid having to use save_main_session
         self.window = window
 
-      def process(self, element: int) -> Iterable[window.TimestampedValue]:  # type: ignore[override]
+      def process(self, element: int) -> Iterable[window.TimestampedValue]:
         yield self.window.TimestampedValue(element, element)
 
     class GetSyntheticSDFOptions(beam.DoFn):
@@ -156,7 +156,7 @@ def __init__(
         self.key_size = key_size
         self.value_size = value_size
 
-      def process(self, element: Any) -> Iterable[Dict[str, Union[int, str]]]:  # type: ignore[override]
+      def process(self, element: Any) -> Iterable[Dict[str, Union[int, str]]]:
         yield {
             'num_records': self.elements_per_record,
             'key_size': self.key_size,

sdks/python/apache_beam/testing/test_stream_service_test.py

@@ -30,9 +30,9 @@
 # Nose automatically detects tests if they match a regex. Here, it mistakens
 # these protos as tests. For more info see the Nose docs at:
 # https://nose.readthedocs.io/en/latest/writing_tests.html
-beam_runner_api_pb2.TestStreamPayload.__test__ = False  # type: ignore[attr-defined]
-beam_interactive_api_pb2.TestStreamFileHeader.__test__ = False  # type: ignore[attr-defined]
-beam_interactive_api_pb2.TestStreamFileRecord.__test__ = False  # type: ignore[attr-defined]
+beam_runner_api_pb2.TestStreamPayload.__test__ = False
+beam_interactive_api_pb2.TestStreamFileHeader.__test__ = False
+beam_interactive_api_pb2.TestStreamFileRecord.__test__ = False
 
 
 class EventsReader:

sdks/python/apache_beam/transforms/core.py

@@ -1415,7 +1415,7 @@ class PartitionFn(WithTypeHints):
   def default_label(self):
     return self.__class__.__name__
 
-  def partition_for(self, element, num_partitions, *args, **kwargs):
+  def partition_for(self, element, num_partitions, *args, **kwargs):  # type: ignore[empty-body]
     # type: (T, int, *typing.Any, **typing.Any) -> int
 
     """Specify which partition will receive this element.
@@ -3451,7 +3451,7 @@ def _dynamic_named_tuple(type_name, field_names):
         type_name, field_names)
     # typing: can't override a method. also, self type is unknown and can't
     # be cast to tuple
-    result.__reduce__ = lambda self: (  # type: ignore[assignment]
+    result.__reduce__ = lambda self: (  # type: ignore[method-assign]
         _unpickle_dynamic_named_tuple, (type_name, field_names, tuple(self)))  # type: ignore[arg-type]
   return result
 

sdks/python/apache_beam/transforms/display.py

@@ -173,7 +173,7 @@ def create_payload(dd) -> Optional[beam_runner_api_pb2.LabelledPayload]:
       elif isinstance(value, (float, complex)):
         return beam_runner_api_pb2.LabelledPayload(
             label=label,
-            double_value=value,
+            double_value=value,  # type: ignore[arg-type]
             key=display_data_dict['key'],
             namespace=display_data_dict.get('namespace', ''))
       else:

sdks/python/apache_beam/transforms/external.py

@@ -740,7 +740,7 @@ def expand(self, pvalueish):
     components = context.to_runner_api()
     request = beam_expansion_api_pb2.ExpansionRequest(
         components=components,
-        namespace=self._external_namespace,
+        namespace=self._external_namespace,  # type: ignore[arg-type]
         transform=transform_proto,
         output_coder_requests=output_coders,
         pipeline_options=pipeline._options.to_runner_api())

sdks/python/apache_beam/transforms/ptransform.py

@@ -748,7 +748,7 @@ def to_runner_api(self, context, has_parts=False, **extra_kwargs):
     # type: (PipelineContext, bool, Any) -> beam_runner_api_pb2.FunctionSpec
     from apache_beam.portability.api import beam_runner_api_pb2
     # typing: only ParDo supports extra_kwargs
-    urn, typed_param = self.to_runner_api_parameter(context, **extra_kwargs)  # type: ignore[call-arg]
+    urn, typed_param = self.to_runner_api_parameter(context, **extra_kwargs)
     if urn == python_urns.GENERIC_COMPOSITE_TRANSFORM and not has_parts:
       # TODO(https://github.com/apache/beam/issues/18713): Remove this fallback.
       urn, typed_param = self.to_runner_api_pickled(context)

sdks/python/apache_beam/transforms/stats.py

@@ -919,7 +919,7 @@ def _offset(self, new_weight):
 
   # TODO(https://github.com/apache/beam/issues/19737): Signature incompatible
   # with supertype
-  def create_accumulator(self):  # type: ignore[override]
+  def create_accumulator(self):
     # type: () -> _QuantileState
     self._qs = _QuantileState(
         unbuffered_elements=[],

sdks/python/apache_beam/transforms/userstate.py

@@ -299,7 +299,7 @@ def validate_stateful_dofn(dofn: 'DoFn') -> None:
           'callback: %s.') % (dofn, timer_spec))
     method_name = timer_spec._attached_callback.__name__
     if (timer_spec._attached_callback != getattr(dofn, method_name,
-                                                 None).__func__):
+                                                 None).__func__):  # type: ignore[union-attr]
       raise ValueError((
           'The on_timer callback for %s is not the specified .%s method '
           'for DoFn %r (perhaps it was overwritten?).') %
@@ -314,7 +314,7 @@ def set(self, timestamp: Timestamp, dynamic_timer_tag: str = '') -> None:
     raise NotImplementedError
 
 
-_TimerTuple = collections.namedtuple('timer_tuple', ('cleared', 'timestamp'))
+_TimerTuple = collections.namedtuple('timer_tuple', ('cleared', 'timestamp'))  # type: ignore[name-match]
 
 
 class RuntimeTimer(BaseTimer):

sdks/python/apache_beam/utils/profiler.py

@@ -104,7 +104,9 @@ def __exit__(self, *args):
         self.profile.create_stats()
         self.profile_output = self._upload_profile_data(
             # typing: seems stats attr is missing from typeshed
-            self.profile_location, 'cpu_profile', self.profile.stats)  # type: ignore[attr-defined]
+            self.profile_location,
+            'cpu_profile',
+            self.profile.stats)
 
       if self.enable_memory_profiling:
         if not self.hpy:

sdks/python/apache_beam/utils/proto_utils.py

@@ -46,7 +46,7 @@ def pack_Any(msg: message.Message) -> any_pb2.Any:
 
 
 @overload
-def pack_Any(msg: None) -> None:
+def pack_Any(msg: None) -> None:  # type: ignore[overload-cannot-match]
   pass
 
 

sdks/python/mypy.ini

@@ -28,11 +28,17 @@ files = apache_beam
 color_output = true
 # uncomment this to see how close we are to being complete
 # check_untyped_defs = true
-disable_error_code = var-annotated
+disable_error_code = var-annotated, import-untyped, valid-type, truthy-function, attr-defined, annotation-unchecked
+
+[tool.mypy]
+ignore_missing_imports = true
 
 [mypy-apache_beam.coders.proto2_coder_test_messages_pb2]
 ignore_errors = true
 
+[mypy-apache_beam.dataframe.*]
+ignore_errors = true
+
 [mypy-apache_beam.examples.*]
 ignore_errors = true
 

sdks/python/tox.ini

@@ -149,7 +149,7 @@ commands =
 
 [testenv:mypy]
 deps =
-  mypy==0.790
+  mypy==1.13.0
   dask==2022.01.0
   distributed==2022.01.0
 # make extras available in case any of these libs are typed