def test_bad_types(self):
    p = df.Pipeline('DirectPipelineRunner', argv=sys.argv)

    # [START type_hints_missing_define_numbers]
    numbers = p | df.Create(['1', '2', '3'])
    # [END type_hints_missing_define_numbers]

    # Consider the following code.
    # [START type_hints_missing_apply]
    evens = numbers | df.Filter(lambda x: x % 2 == 0)
    # [END type_hints_missing_apply]

    # Now suppose numers was defined as [snippet above].
    # When running this pipeline, you'd get a runtime error,
    # possibly on a remote machine, possibly very late.

    with self.assertRaises(TypeError):
      p.run()

    # To catch this early, we can assert what types we expect.
    with self.assertRaises(typehints.TypeCheckError):
      # [START type_hints_takes]
      p.options.view_as(TypeOptions).pipeline_type_check = True
      evens = numbers | df.Filter(lambda x: x % 2 == 0).with_input_types(int)
      # [END type_hints_takes]

    # Type hints can be declared on DoFns and callables as well, rather
    # than where they're used, to be more self contained.
    with self.assertRaises(typehints.TypeCheckError):
      # [START type_hints_do_fn]
      @df.typehints.with_input_types(int)
      class FilterEvensDoFn(df.DoFn):
        def process(self, context):
          if context.element % 2 == 0:
            yield context.element
      evens = numbers | df.ParDo(FilterEvensDoFn())
      # [END type_hints_do_fn]

    words = p | df.Create('words', ['a', 'bb', 'c'])
    # One can assert outputs and apply them to transforms as well.
    # Helps document the contract and checks it at pipeline construction time.
    # [START type_hints_transform]
    T = df.typehints.TypeVariable('T')
    @df.typehints.with_input_types(T)
    @df.typehints.with_output_types(df.typehints.Tuple[int, T])
    class MyTransform(df.PTransform):
      def apply(self, pcoll):
        return pcoll | df.Map(lambda x: (len(x), x))

    words_with_lens = words | MyTransform()
    # [END type_hints_transform]

    with self.assertRaises(typehints.TypeCheckError):
      words_with_lens | df.Map(lambda x: x).with_input_types(
          df.typehints.Tuple[int, int])
dataset/ETHPy150Open GoogleCloudPlatform/DataflowPythonSDK/google/cloud/dataflow/examples/snippets/snippets_test.py/TypeHintsTest.test_bad_types/original