Cattrs 1.7.0 break lineage and papermill

[potiuk]

The Cattrs 1.7.* released by the end of May break lineage (GenConverter used by default):

The error (Python 3.7 and 3.8):

  =================================== FAILURES ===================================
  ___________________________ TestLineage.test_lineage ___________________________
  
  self = <tests.lineage.test_lineage.TestLineage testMethod=test_lineage>
  
      def test_lineage(self):
          dag = DAG(dag_id='test_prepare_lineage', start_date=DEFAULT_DATE)
      
          f1s = "/tmp/does_not_exist_1-{}"
          f2s = "/tmp/does_not_exist_2-{}"
          f3s = "/tmp/does_not_exist_3"
          file1 = File(f1s.format("{{ execution_date }}"))
          file2 = File(f2s.format("{{ execution_date }}"))
          file3 = File(f3s)
      
          with dag:
              op1 = DummyOperator(
                  task_id='leave1',
                  inlets=file1,
                  outlets=[
                      file2,
                  ],
              )
              op2 = DummyOperator(task_id='leave2')
              op3 = DummyOperator(task_id='upstream_level_1', inlets=AUTO, outlets=file3)
              op4 = DummyOperator(task_id='upstream_level_2')
              op5 = DummyOperator(task_id='upstream_level_3', inlets=["leave1", "upstream_level_1"])
      
              op1.set_downstream(op3)
              op2.set_downstream(op3)
              op3.set_downstream(op4)
              op4.set_downstream(op5)
      
          dag.clear()
      
          # execution_date is set in the context in order to avoid creating task instances
          ctx1 = {"ti": TI(task=op1, execution_date=DEFAULT_DATE), "execution_date": DEFAULT_DATE}
          ctx2 = {"ti": TI(task=op2, execution_date=DEFAULT_DATE), "execution_date": DEFAULT_DATE}
          ctx3 = {"ti": TI(task=op3, execution_date=DEFAULT_DATE), "execution_date": DEFAULT_DATE}
          ctx5 = {"ti": TI(task=op5, execution_date=DEFAULT_DATE), "execution_date": DEFAULT_DATE}
      
          # prepare with manual inlets and outlets
          op1.pre_execute(ctx1)
      
          assert len(op1.inlets) == 1
          assert op1.inlets[0].url == f1s.format(DEFAULT_DATE)
      
          assert len(op1.outlets) == 1
          assert op1.outlets[0].url == f2s.format(DEFAULT_DATE)
      
          # post process with no backend
          op1.post_execute(ctx1)
      
          op2.pre_execute(ctx2)
          assert len(op2.inlets) == 0
          op2.post_execute(ctx2)
      
  >       op3.pre_execute(ctx3)
  
  tests/lineage/test_lineage.py:90: 
  _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  airflow/lineage/__init__.py:169: in wrapper
      _get_instance(structure(item, Metadata)) for sublist in _inlets if sublist for item in sublist
  airflow/lineage/__init__.py:169: in <listcomp>
      _get_instance(structure(item, Metadata)) for sublist in _inlets if sublist for item in sublist
  /usr/local/lib/python3.7/site-packages/cattr/converters.py:223: in structure
      return self._structure_func.dispatch(cl)(obj, cl)
  :5: in structure_Metadata
      ???
  :2: in structure_mapping
      ???
  :2: in <dictcomp>
      ???
  _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  
  self = <cattr.converters.GenConverter object at 0x7f6f4cfa0950>
  obj = '/tmp/does_not_exist_2-2016-01-01 00:00:00+00:00', cl = ~VT
  
      def _structure_default(self, obj, cl):
          """This is the fallthrough case. Everything is a subclass of `Any`.
      
          A special condition here handles ``attrs`` classes.
      
          Bare optionals end here too (optionals with arguments are unions.) We
          treat bare optionals as Any.
          """
          if cl is Any or cl is Optional or cl is None:
              return obj
      
          if is_generic(cl):
              fn = make_dict_structure_fn(cl, self)
              self.register_structure_hook(cl, fn)
              return fn(obj)
      
          # We don't know what this is, so we complain loudly.
          msg = (
              "Unsupported type: {0}. Register a structure hook for "
              "it.".format(cl)
          )
  >       raise ValueError(msg)
  E       ValueError: Unsupported type: ~VT. Register a structure hook for it.
  
  /usr/local/lib/python3.7/site-packages/cattr/converters.py:304: ValueError

______________________ TestPapermillOperator.test_execute ______________________
  
  self = <tests.providers.papermill.operators.test_papermill.TestPapermillOperator testMethod=test_execute>
  mock_papermill = <MagicMock name='pm' id='140003323639184'>
  
      @patch('airflow.providers.papermill.operators.papermill.pm')
      def test_execute(self, mock_papermill):
          in_nb = "/tmp/does_not_exist"
          out_nb = "/tmp/will_not_exist"
          parameters = {"msg": "hello_world", "train": 1}
      
          op = PapermillOperator(
              input_nb=in_nb,
              output_nb=out_nb,
              parameters=parameters,
              task_id="papermill_operator_test",
              dag=None,
          )
      
  >       op.pre_execute(context={})  # make sure to have the inlets
  
  tests/providers/papermill/operators/test_papermill.py:39: 
  _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  airflow/lineage/__init__.py:186: in wrapper
      self.inlets = [_render_object(i, context) for i in self.inlets if attr.has(i)]
  airflow/lineage/__init__.py:186: in <listcomp>
      self.inlets = [_render_object(i, context) for i in self.inlets if attr.has(i)]
  airflow/lineage/__init__.py:80: in _render_object
      type(obj),
  /usr/local/lib/python3.7/site-packages/cattr/converters.py:223: in structure
      return self._structure_func.dispatch(cl)(obj, cl)
  :8: in structure_NoteBook
      ???
  /usr/local/lib/python3.7/site-packages/cattr/converters.py:439: in _structure_union
      return self._structure_func.dispatch(other)(obj, other)
  :2: in structure_mapping
      ???
  :2: in <dictcomp>
      ???
  _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  
  self = <cattr.converters.GenConverter object at 0x7f555ae21b90>
  obj = 'hello_world', cl = ~VT
  
      def _structure_default(self, obj, cl):
          """This is the fallthrough case. Everything is a subclass of `Any`.
      
          A special condition here handles ``attrs`` classes.
      
          Bare optionals end here too (optionals with arguments are unions.) We
          treat bare optionals as Any.
          """
          if cl is Any or cl is Optional or cl is None:
              return obj
      
          if is_generic(cl):
              fn = make_dict_structure_fn(cl, self)
              self.register_structure_hook(cl, fn)
              return fn(obj)
      
          # We don't know what this is, so we complain loudly.
          msg = (
              "Unsupported type: {0}. Register a structure hook for "
              "it.".format(cl)
          )
  >       raise ValueError(msg)
  E       ValueError: Unsupported type: ~VT. Register a structure hook for it.
  
  /usr/local/lib/python3.7/site-packages/cattr/converters.py:304: ValueError

[ValBerthe]

Downgrading cattrs doesn't seem to fix this issue.
Tried:

• 1.7.0
• 1.4.0
• 1.1.2
• 1.0.0

Stacktrace :

Traceback (most recent call last):
  File "/usr/local/lib/python3.8/dist-packages/airflow/models/taskinstance.py", line 1137, in _run_raw_task
    self._prepare_and_execute_task_with_callbacks(context, task)
  File "/usr/local/lib/python3.8/dist-packages/airflow/models/taskinstance.py", line 1288, in _prepare_and_execute_task_with_callbacks
    task_copy.pre_execute(context=context)
  File "/usr/local/lib/python3.8/dist-packages/airflow/lineage/__init__.py", line 186, in wrapper
    self.inlets = [_render_object(i, context) for i in self.inlets if attr.has(i)]
  File "/usr/local/lib/python3.8/dist-packages/airflow/lineage/__init__.py", line 186, in <listcomp>
    self.inlets = [_render_object(i, context) for i in self.inlets if attr.has(i)]
  File "/usr/local/lib/python3.8/dist-packages/airflow/lineage/__init__.py", line 74, in _render_object
    return structure(
  File "/usr/local/lib/python3.8/dist-packages/cattr/converters.py", line 223, in structure
    # Classes to Python primitives.
  File "", line 8, in structure_NoteBook
  File "/usr/local/lib/python3.8/dist-packages/cattr/converters.py", line 439, in _structure_union
    # optional with more than one parameter.
  File "/usr/local/lib/python3.8/dist-packages/cattr/dispatch.py", line 47, in _dispatch
    return self._function_dispatch.dispatch(cl)
  File "/usr/local/lib/python3.8/dist-packages/cattr/dispatch.py", line 121, in dispatch
    return handler(typ)
  File "/usr/local/lib/python3.8/dist-packages/cattr/converters.py", line 711, in gen_structure_mapping
  File "/usr/local/lib/python3.8/dist-packages/cattr/gen.py", line 353, in make_mapping_structure_fn
    val_handler = converter._structure_func.dispatch(val_type)
ValueError: not enough values to unpack (expected 1, got 0)

This is not broken with Airflow 2.0.2.

[potiuk]

This is a different error I believe. The issue is closed (we have cattrs<1.7.0 and the tests work fine on CI - this was the main reason the issue was created). If you have other error/reproducible scenario, can you please open another issue for it @ValBerthe ?