class_docstr: Small webserver to serve scheduler health check.
class_docstr_tok: ["Small","webserver","to","serve","scheduler","health","check","."]
code: def do_GET(self):
      if self.path == "/health":
          try:
              with create_session() as session:
                  scheduler_job = session.scalar(
                      select(Job)
                      .filter_by(job_type=SchedulerJobRunner.job_type)
                      .filter_by(hostname=get_hostname())
                      .order_by(Job.latest_heartbeat.desc())
                      .limit(1)
                  )
              if scheduler_job and scheduler_job.is_alive():
                  self.send_response(200)
                  self.end_headers()
              else:
                  self.send_error(503)
          except Exception:
              log.exception("Exception when executing Health check")
              self.send_error(503)
      else:
          self.send_error(404)
  
file_path: airflow/airflow/utils/scheduler_health.py
imports: from __future__ import annotations
  import logging
  from http.server import BaseHTTPRequestHandler, HTTPServer
  from sqlalchemy import select
  from airflow.configuration import conf
  from airflow.jobs.job import Job
  from airflow.jobs.scheduler_job_runner import SchedulerJobRunner
  from airflow.utils.net import get_hostname
  from airflow.utils.session import create_session
label_desc: Use image node_id 1 for calling the HealthServer obj's underlying member method code with example usage: obj.do_GET() without return types