Rename JobRunner modules to *_job_runner and base_job* to job

The #30255 introduced "JobRunner" concept and decoupled the job logic
from the ORM polymorphic *Job objects. The change was implemented
in the way to minimise the review effort needed, so it avoided renaming
the modules for the runners (from `_job` to `_job_runner`).

Also BaseJob lost its "polymorphism" properties so the package, and class name
can be renamed to simply job.

This PR completes the JobRunner concept introduction by applying the
renames.

Closes: #30296

Author: potiuk

airflow/api_connexion/endpoints/health_endpoint.py

@@ -18,8 +18,8 @@
 
 from airflow.api_connexion.schemas.health_schema import health_schema
 from airflow.api_connexion.types import APIResponse
-from airflow.jobs.scheduler_job import SchedulerJobRunner
-from airflow.jobs.triggerer_job import TriggererJobRunner
+from airflow.jobs.scheduler_job_runner import SchedulerJobRunner
+from airflow.jobs.triggerer_job_runner import TriggererJobRunner
 
 HEALTHY = "healthy"
 UNHEALTHY = "unhealthy"

airflow/api_connexion/schemas/job_schema.py

@@ -19,7 +19,7 @@
 
 from marshmallow_sqlalchemy import SQLAlchemySchema, auto_field
 
-from airflow.jobs.base_job import BaseJob
+from airflow.jobs.job import Job
 
 
 class JobSchema(SQLAlchemySchema):
@@ -28,7 +28,7 @@ class JobSchema(SQLAlchemySchema):
     class Meta:
         """Meta."""
 
-        model = BaseJob
+        model = Job
 
     id = auto_field(dump_only=True)
     dag_id = auto_field(dump_only=True)

airflow/cli/commands/dag_command.py

@@ -35,7 +35,7 @@
 from airflow.cli.simple_table import AirflowConsole
 from airflow.configuration import conf
 from airflow.exceptions import AirflowException, RemovedInAirflow3Warning
-from airflow.jobs.base_job import BaseJob
+from airflow.jobs.job import Job
 from airflow.models import DagBag, DagModel, DagRun, TaskInstance
 from airflow.models.dag import DAG
 from airflow.models.serialized_dag import SerializedDagModel
@@ -391,15 +391,13 @@ def dag_list_jobs(args, dag: DAG | None = None, session: Session = NEW_SESSION)
 
         if not dag:
             raise SystemExit(f"DAG: {args.dag_id} does not exist in 'dag' table")
-        queries.append(BaseJob.dag_id == args.dag_id)
+        queries.append(Job.dag_id == args.dag_id)
 
     if args.state:
-        queries.append(BaseJob.state == args.state)
+        queries.append(Job.state == args.state)
 
     fields = ["dag_id", "state", "job_type", "start_date", "end_date"]
-    all_jobs = (
-        session.query(BaseJob).filter(*queries).order_by(BaseJob.start_date.desc()).limit(args.limit).all()
-    )
+    all_jobs = session.query(Job).filter(*queries).order_by(Job.start_date.desc()).limit(args.limit).all()
     all_jobs = [{f: str(job.__getattribute__(f)) for f in fields} for job in all_jobs]
 
     AirflowConsole().print_as(

airflow/cli/commands/dag_processor_command.py

@@ -26,14 +26,14 @@
 
 from airflow import settings
 from airflow.configuration import conf
-from airflow.jobs.base_job import BaseJob
+from airflow.jobs.job import Job
 from airflow.utils import cli as cli_utils
 from airflow.utils.cli import setup_locations, setup_logging
 
 log = logging.getLogger(__name__)
 
 
-def _create_dag_processor_job(args: Any) -> BaseJob:
+def _create_dag_processor_job(args: Any) -> Job:
     """Creates DagFileProcessorProcess instance."""
     from airflow.dag_processing.manager import DagFileProcessorManager
 
@@ -47,7 +47,7 @@ def _create_dag_processor_job(args: Any) -> BaseJob:
         dag_ids=[],
         pickle_dags=args.do_pickle,
     )
-    return BaseJob(
+    return Job(
         job_runner=processor.job_runner,
     )
 

airflow/cli/commands/jobs_command.py

@@ -18,7 +18,7 @@
 
 from sqlalchemy.orm import Session
 
-from airflow.jobs.base_job import BaseJob
+from airflow.jobs.job import Job
 from airflow.utils.net import get_hostname
 from airflow.utils.session import NEW_SESSION, provide_session
 from airflow.utils.state import State
@@ -32,22 +32,17 @@ def check(args, session: Session = NEW_SESSION) -> None:
     if args.hostname and args.local:
         raise SystemExit("You can't use --hostname and --local at the same time")
 
-    query = (
-        session.query(BaseJob)
-        .filter(BaseJob.state == State.RUNNING)
-        .order_by(BaseJob.latest_heartbeat.desc())
-    )
+    query = session.query(Job).filter(Job.state == State.RUNNING).order_by(Job.latest_heartbeat.desc())
     if args.job_type:
-        query = query.filter(BaseJob.job_type == args.job_type)
+        query = query.filter(Job.job_type == args.job_type)
     if args.hostname:
-        query = query.filter(BaseJob.hostname == args.hostname)
+        query = query.filter(Job.hostname == args.hostname)
     if args.local:
-        query = query.filter(BaseJob.hostname == get_hostname())
+        query = query.filter(Job.hostname == get_hostname())
     if args.limit > 0:
         query = query.limit(args.limit)
 
-    jobs: list[BaseJob] = query.all()
-    alive_jobs = [job for job in jobs if job.is_alive()]
+    alive_jobs: list[Job] = [job for job in query.all() if job.is_alive()]
 
     count_alive_jobs = len(alive_jobs)
     if count_alive_jobs == 0:

airflow/cli/commands/scheduler_command.py

@@ -28,14 +28,14 @@
 from airflow.api_internal.internal_api_call import InternalApiConfig
 from airflow.configuration import conf
 from airflow.executors.executor_loader import ExecutorLoader
-from airflow.jobs.base_job import BaseJob
-from airflow.jobs.scheduler_job import SchedulerJobRunner
+from airflow.jobs.job import Job
+from airflow.jobs.scheduler_job_runner import SchedulerJobRunner
 from airflow.utils import cli as cli_utils
 from airflow.utils.cli import process_subdir, setup_locations, setup_logging, sigint_handler, sigquit_handler
 from airflow.utils.scheduler_health import serve_health_check
 
 
-def _run_scheduler_job(job: BaseJob, *, skip_serve_logs: bool) -> None:
+def _run_scheduler_job(job: Job, *, skip_serve_logs: bool) -> None:
     InternalApiConfig.force_database_direct_access()
     enable_health_check = conf.getboolean("scheduler", "ENABLE_HEALTH_CHECK")
     with _serve_logs(skip_serve_logs), _serve_health_check(enable_health_check):
@@ -47,7 +47,7 @@ def scheduler(args):
     """Starts Airflow Scheduler."""
     print(settings.HEADER)
 
-    job = BaseJob(
+    job = Job(
         job_runner=SchedulerJobRunner(
             subdir=process_subdir(args.subdir),
             num_runs=args.num_runs,

airflow/cli/commands/standalone_command.py

@@ -30,10 +30,10 @@
 from airflow.configuration import AIRFLOW_HOME, conf, make_group_other_inaccessible
 from airflow.executors import executor_constants
 from airflow.executors.executor_loader import ExecutorLoader
-from airflow.jobs.base_job import most_recent_job
-from airflow.jobs.job_runner import BaseJobRunner
-from airflow.jobs.scheduler_job import SchedulerJobRunner
-from airflow.jobs.triggerer_job import TriggererJobRunner
+from airflow.jobs.base_job_runner import BaseJobRunner
+from airflow.jobs.job import most_recent_job
+from airflow.jobs.scheduler_job_runner import SchedulerJobRunner
+from airflow.jobs.triggerer_job_runner import TriggererJobRunner
 from airflow.utils import db
 
 

airflow/cli/commands/task_command.py

@@ -37,8 +37,8 @@
 from airflow.configuration import conf
 from airflow.exceptions import AirflowException, DagRunNotFound, TaskInstanceNotFound
 from airflow.executors.executor_loader import ExecutorLoader
-from airflow.jobs.base_job import BaseJob
-from airflow.jobs.local_task_job import LocalTaskJobRunner
+from airflow.jobs.job import Job
+from airflow.jobs.local_task_job_runner import LocalTaskJobRunner
 from airflow.listeners.listener import get_listener_manager
 from airflow.models import DagPickle, TaskInstance
 from airflow.models.dag import DAG
@@ -260,7 +260,7 @@ def _run_task_by_local_task_job(args, ti: TaskInstance) -> TaskReturnCode | None
         pool=args.pool,
         external_executor_id=_extract_external_executor_id(args),
     )
-    run_job = BaseJob(
+    run_job = Job(
         job_runner=local_task_job_runner,
         dag_id=ti.dag_id,
     )

airflow/cli/commands/triggerer_command.py

@@ -28,8 +28,8 @@
 
 from airflow import settings
 from airflow.configuration import conf
-from airflow.jobs.base_job import BaseJob
-from airflow.jobs.triggerer_job import TriggererJobRunner
+from airflow.jobs.job import Job
+from airflow.jobs.triggerer_job_runner import TriggererJobRunner
 from airflow.utils import cli as cli_utils
 from airflow.utils.cli import setup_locations, setup_logging, sigint_handler, sigquit_handler
 from airflow.utils.serve_logs import serve_logs
@@ -56,7 +56,7 @@ def triggerer(args):
     settings.MASK_SECRETS_IN_LOGS = True
     print(settings.HEADER)
     triggerer_job_runner = TriggererJobRunner(capacity=args.capacity)
-    job = BaseJob(job_runner=triggerer_job_runner)
+    job = Job(job_runner=triggerer_job_runner)
 
     if args.daemon:
         pid, stdout, stderr, log_file = setup_locations(

airflow/dag_processing/manager.py

@@ -46,7 +46,7 @@
 from airflow.callbacks.callback_requests import CallbackRequest, SlaCallbackRequest
 from airflow.configuration import conf
 from airflow.dag_processing.processor import DagFileProcessorProcess
-from airflow.jobs.base_job import perform_heartbeat
+from airflow.jobs.job import perform_heartbeat
 from airflow.models import errors
 from airflow.models.dag import DagModel
 from airflow.models.dagwarning import DagWarning
@@ -68,7 +68,7 @@
 from airflow.utils.sqlalchemy import prohibit_commit, skip_locked, with_row_locks
 
 if TYPE_CHECKING:
-    from airflow.jobs.dag_processor_job import DagProcessorJobRunner
+    from airflow.jobs.dag_processor_job_runner import DagProcessorJobRunner
 
 
 class DagParsingStat(NamedTuple):
@@ -385,7 +385,7 @@ def __init__(
         signal_conn: MultiprocessingConnection | None = None,
         async_mode: bool = True,
     ):
-        from airflow.jobs.dag_processor_job import DagProcessorJobRunner
+        from airflow.jobs.dag_processor_job_runner import DagProcessorJobRunner
 
         super().__init__()
         # known files; this will be updated every `dag_dir_list_interval` and stuff added/removed accordingly