Better typing for Job and JobRunners

[potiuk]

By leveraging Generics, the typing for Runners and Job and JobPydantic is now more complete and accurate.

• Scheduler and Backfill Runners limit their code to Job and can use all the things that ORM Job allows them to do

• Other runners are limited to union of Job and JobPydantic version so that they can be run on the client side of the internal API without having all the Job features.

This is a follow up after #31182 that fixed missing job_type for DagProcessor Job and nicely extracted job to BaseRunner but broke MyPy/Typing guards implemented in the runners that should aid the AIP-44 implementation.

---

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an 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 a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[potiuk]

cc: @vincbeck @mhenc -> example of where a bit more complex typing can help us to deal with ORM/Pydantic distinction

[potiuk]

cc: @uranusjr -> I'd appreciate your keen eye on it, I am not 100% sure if this is the right way of doing it, typing system of Python when it comes to Generics is kinda convoluted and I am still learning it. The way I implemented it, however should explain the intent I had, but maybe there are better ways of implementing the intent.

[potiuk]

Thanks @uranusjr - yep, this is much nicer now :)

[potiuk]

Actually - I am at a loss now :).

had to remove it all from TYPE_CHECKING actually (Generic cannot have types defined in TYPE_CHECKING). ... And when I did ....

airflow/jobs/local_task_job_runner.py:90: error: Argument 1 to "__init__" of
"BaseJobRunner" has incompatible type "JobPydantic"; expected "JobOrJobPydantic"
 [arg-type]
            super().__init__(job)

I even tried to cast it and ...

airflow/jobs/local_task_job_runner.py:90: error: Redundant cast to "JobPydantic" 
[redundant-cast]
            super().__init__(cast(JobOrJobPydantic, job))

Typing is hard :)

[potiuk]

Somehow mypy thinks JobOrJobPydantic is actually JobPydantic :)

[potiuk]

Aactually - turned out to be quite a bit more - moving out of TYPE_CHECKING causes (of course) circular imports so things get messy (have to rethink it)

[ashb]

@potiuk Might using a protocol here be a fix? Rather than trying to say JobOrJobPydantic, define a Protocol that has the type we need, which I think might be just this:

class JobProtocol(Protocol):
    job_type: str

[potiuk]

Yeah. That was my next idea to try :)

[potiuk]

Actually, it was easier than I thought. All that was needed was to move setting self.job up to the concrete runners (while leaving passing Job/JobPydantic) to the BaseJobRunner constructor, so that it can set the right job_type.

This (counterintuitively - we do not set job as a field in baseRunner, it makes perfect sense.

• we do not set the job in base runner (not needed nor used there)
• we set the job in concrete runners (and there it has the right type)
• we won't forget to set the "job_type" by the runner when assigned to a Job as it is set in the base runner
• no need to add extra Protocol class

[uranusjr]

I’ll experiement some other approaches later as well, this assignment still fits better in the base class.