Coverage for /pythoncovmergedfiles/medio/medio/src/airflow/airflow/jobs/base_job_runner.py: 58%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# Licensed to the Apache Software Foundation (ASF) under one
2# or more contributor license agreements. See the NOTICE file
3# distributed with this work for additional information
4# regarding copyright ownership. The ASF licenses this file
5# to you under the Apache License, Version 2.0 (the
6# "License"); you may not use this file except in compliance
7# with the License. You may obtain a copy of the License at
8#
9# http://www.apache.org/licenses/LICENSE-2.0
10#
11# Unless required by applicable law or agreed to in writing,
12# software distributed under the License is distributed on an
13# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14# KIND, either express or implied. See the License for the
15# specific language governing permissions and limitations
16# under the License.
18from __future__ import annotations
20from typing import TYPE_CHECKING
22from airflow.exceptions import AirflowException
23from airflow.utils.session import NEW_SESSION, provide_session
25if TYPE_CHECKING:
26 from sqlalchemy.orm import Session
28 from airflow.jobs.job import Job
29 from airflow.serialization.pydantic.job import JobPydantic
32class BaseJobRunner:
33 """Abstract class for job runners to derive from."""
35 job_type = "undefined"
37 def __init__(self, job: Job) -> None:
38 if job.job_type and job.job_type != self.job_type:
39 raise AirflowException(
40 f"The job is already assigned a different job_type: {job.job_type}."
41 f"This is a bug and should be reported."
42 )
43 job.job_type = self.job_type
44 self.job: Job = job
46 def _execute(self) -> int | None:
47 """
48 Execute the logic connected to the runner.
50 This method should be overridden by subclasses.
52 :meta private:
53 :return: return code if available, otherwise None
54 """
55 raise NotImplementedError()
57 @provide_session
58 def heartbeat_callback(self, session: Session = NEW_SESSION) -> None:
59 """
60 Execute callback during heartbeat.
62 This method can be overwritten by the runners.
63 """
65 @classmethod
66 @provide_session
67 def most_recent_job(cls, session: Session = NEW_SESSION) -> Job | JobPydantic | None:
68 """Return the most recent job of this type, if any, based on last heartbeat received."""
69 from airflow.jobs.job import most_recent_job
71 return most_recent_job(cls.job_type, session=session)