[BEAM-7746] Fix a typing issue where SourceBase was assumed to have a coder attribute

sdks/python/apache_beam/io/iobase.py

@@ -899,10 +899,13 @@ def get_windowing(self, unused_inputs):
 
   def _infer_output_coder(self, input_type=None, input_coder=None):
     # type: (...) -> Optional[coders.Coder]
+    from apache_beam.runners.dataflow.native_io import iobase as dataflow_io
     if isinstance(self.source, BoundedSource):
       return self.source.default_output_coder()
-    else:
+    elif isinstance(self.source, dataflow_io.NativeSource):
       return self.source.coder
+    else:
+      return None
 
   def display_data(self):
     return {'source': DisplayDataItem(self.source.__class__,

sdks/python/apache_beam/runners/dataflow/native_io/iobase.py

@@ -26,12 +26,17 @@
 
 import logging
 from builtins import object
+from typing import TYPE_CHECKING
+from typing import Optional
 
 from apache_beam import pvalue
 from apache_beam.io import iobase
 from apache_beam.transforms import ptransform
 from apache_beam.transforms.display import HasDisplayData
 
+if TYPE_CHECKING:
+  from apache_beam import coders
+
 _LOGGER = logging.getLogger(__name__)
 
 
@@ -58,6 +63,7 @@ class NativeSource(iobase.SourceBase):
 
   This class is deprecated and should not be used to define new sources.
   """
+  coder = None  # type: Optional[coders.Coder]
 
   def reader(self):
     """Returns a NativeSourceReader instance associated with this source."""