potiuk commented on code in PR #30255:
URL: https://github.com/apache/airflow/pull/30255#discussion_r1161340144
##########
airflow/task/task_runner/base_task_runner.py:
##########
@@ -46,18 +49,23 @@ class BaseTaskRunner(LoggingMixin):
Invoke the `airflow tasks run` command with raw mode enabled in a
subprocess.
- :param local_task_job: The local task job associated with running the
- associated task instance.
+ :param base_job: The job associated with running the associated task
instance. The job_runner for it
+ should be LocalTaskJobRunner
"""
- def __init__(self, local_task_job):
- # Pass task instance context into log handlers to setup the logger.
- super().__init__(local_task_job.task_instance)
- self._task_instance = local_task_job.task_instance
+ def __init__(self, base_job: BaseJob):
+ self.job_runner: LocalTaskJobRunner = cast(LocalTaskJobRunner,
base_job.job_runner)
+ if not hasattr(self.job_runner, "task_instance"):
+ raise ValueError(
+ "BaseTaskRunner can only be used with LocalTaskJobRunner and "
+ "have task_instance field defined"
+ )
+ super().__init__(self.job_runner.task_instance)
+ self._task_instance = self.job_runner.task_instance
popen_prepend = []
if self._task_instance.run_as_user:
- self.run_as_user = self._task_instance.run_as_user
+ self.run_as_user: str | None = self._task_instance.run_as_user
Review Comment:
I think it can. There is quite a lot of conditional code there - for example
`from_ti` method:
```
run_as_user=ti.run_as_user if hasattr(ti, "run_as_user") else
None,
```
I wanted to avoid making unnecessary changes in this PR, and maybe we can do
some other fixes when we actually implement AIP-44 (there will be likely
non-refactor cleanup changes when we do it).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
