Fix Mypy errors in airflow/dag_processing #20470

josh-fell · 2021-12-22T19:11:06Z

Related: #19891

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code change, Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in UPDATING.md.

josh-fell · 2021-12-22T19:12:44Z

airflow/dag_processing/processor.py

+            with redirect_stdout(StreamLogWriter(log, logging.INFO)), (  # type: ignore[type-var]
+                redirect_stderr(StreamLogWriter(log, logging.WARN))  # type: ignore[type-var]


Errors here:

airflow/dag_processing/processor.py:152: error: Value of type variable "_T_io" of "redirect_stdout" cannot be "StreamLogWriter" with redirect_stdout(StreamLogWriter(log, logging.INFO)), ... ^ airflow/dag_processing/processor.py:152: error: Value of type variable "_T_io" of "redirect_stderr" cannot be "StreamLogWriter"

Typing for args of both redirect_stdout() and redirect_stderr() is handled in an external
contextlib stub file and defined as ContextManager[TypeVar("_T_io", bound=Optional[IO[str]])]
so ignoring the Mypy errors here. But any feedback on a proper handling of this is welcomed of course.

We can implement the IO protocol on StreamLogWriter instead. It already has most of the important things implemented, the rest are mostly trivial.

I pushed a fix (which is really a workaround). I could not find a way to make it works but It seems that unless we want to reimplement all the methods manually, the IOBase and IO type hierarchies are incompatible and other than ignoring the MyPy errors there is not much we can do about it

https://giters.com/python/typeshed/issues/6077

At least I pushed it down as much as I could. The ignores now simply ignore those errors:

airflow/utils/log/logging_mixin.py:83: error: Definition of "__next__" in base class "IOBase" is incompatible with definition in base class "IO" class StreamLogWriter(IOBase, IO[str]): ^ airflow/utils/log/logging_mixin.py:83: error: Definition of "__next__" in base class "IOBase" is incompatible with definition in base class "Iterator" class StreamLogWriter(IOBase, IO[str]): ^ airflow/utils/log/logging_mixin.py:83: error: Definition of "readlines" in base class "IOBase" is incompatible with definition in base class "IO" class StreamLogWriter(IOBase, IO[str]): ^ airflow/utils/log/logging_mixin.py:83: error: Definition of "__iter__" in base class "IOBase" is incompatible with definition in base class "IO" class StreamLogWriter(IOBase, IO[str]): ^ airflow/utils/log/logging_mixin.py:83: error: Definition of "__iter__" in base class "IOBase" is incompatible with definition in base class "Iterator" class StreamLogWriter(IOBase, IO[str]): ^ airflow/utils/log/logging_mixin.py:83: error: Definition of "__iter__" in base class "IOBase" is incompatible with definition in base class "Iterable" class StreamLogWriter(IOBase, IO[str]): ^ airflow/utils/log/logging_mixin.py:83: error: Definition of "readline" in base class "IOBase" is incompatible with definition in base class "IO" class StreamLogWriter(IOBase, IO[str]): ^ airflow/utils/log/logging_mixin.py:83: error: Definition of "writelines" in base class "IOBase" is incompatible with definition in base class "IO" class StreamLogWriter(IOBase, IO[str]): ^ airflow/utils/log/logging_mixin.py:83: error: Definition of "__enter__" in base class "IOBase" is incompatible with definition in base class "IO" class StreamLogWriter(IOBase, IO[str]): ^

Or maybe there is an IO Protocol you mentioned @uranusjr which I simply could not find ? The only "IO" I see is the Generic (which I now implemented) - but being Generic it does not have any implementation. The only "base" implementation I am aware of that implements IO is BaseIO family (but then it's incompatible with the Generic).

I am also not 100% sure if removing encoding = None is a good thing and whether it does not have any side effects :(.

josh-fell · 2021-12-22T19:21:35Z

airflow/dag_processing/processor.py

                    self._execute_task_callbacks(dagbag, request)
                elif isinstance(request, SlaCallbackRequest):
-                    self.manage_slas(dagbag.dags.get(request.dag_id))
+                    self.manage_slas(dagbag.dags[request.dag_id])


Error here:

airflow/dag_processing/processor.py:577: error: Argument 1 to "manage_slas" of "DagFileProcessor" has incompatible type "Optional[DAG]"; expected "DAG" self.manage_slas(dagbag.dags.get(request.dag_id))

Not entirely sure if there is a case when an SLA callback would be executed but not have the corresponding DAG in the DagBag. If it's possible, I can certainly modify manage_slas() typing and raise an exception for a missing DAG if this is a possible scenario.

SLA doesn’t make sense without a DAG, so this is better IMO. It’s good to fail here if things go wrong.

Yeah. They basically can't go wrong here or we are in a really deep mess.

Yup, the change should be good

airflow/models/taskinstance.py

uranusjr

Also

      def test_encoding(self):
          logger = mock.MagicMock()
          logger.log = mock.MagicMock()
      
          log = StreamLogWriter(logger, 1)
  >       assert log.encoding is None
  E       AttributeError: 'StreamLogWriter' object has no attribute 'encoding'

Although I’m not sure why we need encoding exactly; maybe we could remove this test entirely?

uranusjr · 2021-12-24T08:12:44Z

airflow/dag_processing/processor.py

            # redirect stdout/stderr to log
-            with redirect_stdout(StreamLogWriter(log, logging.INFO)), redirect_stderr(
-                StreamLogWriter(log, logging.WARN)
+            with redirect_stdout(StreamLogWriter(log, logging.INFO)), (
+                redirect_stderr(StreamLogWriter(log, logging.WARN))
            ), Stats.timer() as timer:


This does not work because you can’t use parentheses in a with statement like anywhere else in the language, until 3.10 😞

A popular workaround is

import contextlib with contextlib.ExitStack() as stack: stack.enter_context(redirect_stdout(StreamLogWriter(log, logging.INFO))) stack.enter_context(redirect_stderr(StreamLogWriter(log, logging.WARN))) timer = stack.enter_context(Stats.timer())

That is...unfortunate. This change was made to make flake8 happy for line-length validation. Now that the Mypy ignore comments are gone, we could revert this change. WDYT?

Although I’m not sure why we need encoding exactly; maybe we could remove this test entirely?

I looked at the history and the reason is that some of the internals of subprocess use stderr.encoding and stdout.enconding and we effectvely replace stderr and stdout with StreamLogWriter (which has other bad consequences like infinte redirect when someone wants to also log stufff to stderr/stdout - buit this is a different topic - #20500 )

This is an excerpt from subprcess.py where it woudl fail:

# Translate newlines, if requested. # This also turns bytes into strings. if self.text_mode: if stdout is not None: stdout = self._translate_newlines(stdout, self.stdout.encoding, self.stdout.errors) if stderr is not None: stderr = self._translate_newlines(stderr, self.stderr.encoding, self.stderr.errors) return (stdout, stderr)

On the other hand - we should not really IMHO set it to None (though it results in "utf-8" being used in decode of bytes) but likely with the sys.getdefaultencoding()

(and BTW -yeah, we should rewrite big part of our logging implementation to make it sane).

potiuk · 2021-12-31T13:17:48Z

Needs rebase :(

josh-fell · 2021-12-31T15:48:59Z

Needs rebase :(

Will do. What do we want to do about encoding? Add encoding back, remove the failing test, something else?

kaxil · 2021-12-31T15:51:31Z

airflow/stats.py


    @classmethod
-    def timing(cls, stat: str, dt: Union[float, datetime.timedelta]) -> None:
+    def timing(cls, stat: str, dt: Optional[Union[float, datetime.timedelta]]) -> None:


this is not correct

Instead you should have:

diff --git a/airflow/dag_processing/manager.py b/airflow/dag_processing/manager.py index 408bb48aa..4483c6a58 100644 --- a/airflow/dag_processing/manager.py +++ b/airflow/dag_processing/manager.py @@ -70,8 +70,8 @@ class DagFileStat(NamedTuple): num_dags: int import_errors: int - last_finish_time: Optional[datetime] - last_duration: Optional[float] + last_finish_time: datetime + last_duration: float run_count: int

The Optional typing was to handle this instance. Should we try something else instead?
https://github.com/apache/airflow/blob/main/airflow/dag_processing/manager.py#L1043-L1047

In that case the following should be ok:

diff --git a/airflow/dag_processing/manager.py b/airflow/dag_processing/manager.py index 408bb48aa..25c6ec70f 100644 --- a/airflow/dag_processing/manager.py +++ b/airflow/dag_processing/manager.py @@ -920,7 +920,7 @@ class DagFileProcessorManager(LoggingMixin): self._file_stats[processor.file_path] = stat file_name = os.path.splitext(os.path.basename(processor.file_path))[0].replace(os.sep, '.') - Stats.timing(f'dag_processing.last_duration.{file_name}', stat.last_duration) + Stats.timing(f'dag_processing.last_duration.{file_name}', cast(float, stat.last_duration)) def collect_results(self) -> None: """Collect the result from any finished DAG processors"""

Stats.timing should never receive None as value - https://github.com/jsocol/pystatsd/blob/f3f304b4b2c3d5eddeb9f4977d9c82c64c37a052/statsd/client/base.py#L26-L30

Oh OK. Right on, let me try that.

If I move the cast() here we're golden:

for file_path in files_paths_to_queue: if file_path not in self._file_stats: self._file_stats[file_path] = DagFileStat( num_dags=0, import_errors=0, last_finish_time=None, last_duration=cast(float, None), run_count=0, )

Thanks for the tip @kaxil!

josh-fell · 2021-12-31T15:58:43Z

I was also going to take care of the context manager situation @uranusjr pointed out. Since the ignore comments are being moved, I'll just revert the formatting change.

josh-fell · 2022-01-06T15:17:12Z

How should we handle the StreamLogWriter.encoding attribute and the failing tests? Add encoding back, remove the failing test, use sys.getdefaultencoding() and update the test, something else?

potiuk · 2022-01-06T18:05:50Z

sys.getdefaultencoding() would be my vote

ashb · 2022-01-13T14:37:02Z

Fixed in #20795

josh-fell requested review from XD-DENG, ashb, ephraimbuddy, jedcunningham and kaxil as code owners December 22, 2021 19:11

boring-cyborg bot added the area:Scheduler including HA (high availability) scheduler label Dec 22, 2021

josh-fell commented Dec 22, 2021

View reviewed changes

josh-fell mentioned this pull request Dec 22, 2021

Re-enable MyPy #19891

Closed

10 tasks

kaxil reviewed Dec 23, 2021

View reviewed changes

airflow/models/taskinstance.py Show resolved Hide resolved

uranusjr reviewed Dec 24, 2021

View reviewed changes

potiuk added the mypy Fixing MyPy problems after bumpin MyPy to 0.990 label Dec 31, 2021

kaxil reviewed Dec 31, 2021

View reviewed changes

kaxil mentioned this pull request Jan 4, 2022

Fxing MyPy issues for airflow/dag_processing/processor.py #20650

Closed

josh-fell force-pushed the fix-mypy-errors-dag-processing branch from f2e4c3d to dab7b66 Compare January 6, 2022 03:58

josh-fell and others added 5 commits January 10, 2022 16:27

Fix Mypy errors in airflow/dag_processing

1a9bad4

fixup! Fix Mypy errors in airflow/dag_processing

998c611

Restore code-formatting for redirect context managers

ad1ab53

Update DagFileStat typing

b7c4f28

Update StreamLogWriter encoding value

506daf2

josh-fell force-pushed the fix-mypy-errors-dag-processing branch from dab7b66 to 506daf2 Compare January 10, 2022 21:27

josh-fell mentioned this pull request Jan 12, 2022

Fix remaining mypy issues in "core" Airflow #20795

Merged

ashb closed this Jan 13, 2022

kaxil added the changelog:skip Changes that should be skipped from the changelog (CI, tests, etc..) label Jan 13, 2022

josh-fell deleted the fix-mypy-errors-dag-processing branch January 18, 2022 15:01

		with redirect_stdout(StreamLogWriter(log, logging.INFO)), ( # type: ignore[type-var]
		redirect_stderr(StreamLogWriter(log, logging.WARN)) # type: ignore[type-var]

Fix Mypy errors in airflow/dag_processing #20470

Fix Mypy errors in airflow/dag_processing #20470

Uh oh!

Conversation

josh-fell commented Dec 22, 2021

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

potiuk Dec 23, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

josh-fell Dec 22, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

uranusjr left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

potiuk Dec 29, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

potiuk commented Dec 31, 2021

Uh oh!

josh-fell commented Dec 31, 2021

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

josh-fell commented Dec 31, 2021

Uh oh!

josh-fell commented Jan 6, 2022

Uh oh!

potiuk commented Jan 6, 2022

Uh oh!

ashb commented Jan 13, 2022

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

potiuk Dec 23, 2021 •

edited

Loading

josh-fell Dec 22, 2021 •

edited

Loading

uranusjr left a comment •

edited

Loading

potiuk Dec 29, 2021 •

edited

Loading