Scheduler/Webserver should never use Internal API, so calling any method decorated with internal_api_call should still execute them locally
from airflow.exceptions import AirflowConfigException
from airflow.logging_config import configure_logging
from airflow.models import import_all_models
from airflow.utils import cli as cli_utils
from airflow.utils.cli import setup_locations, setup_logging
from airflow.utils.process_utils import check_if_pidfile_process_is_running
from airflow.www.extensions.init_dagbag import init_dagbag
from airflow.www.extensions.init_jinja_globals import init_jinja_globals
from airflow.www.extensions.init_manifest_files import configure_manifest_files
from airflow.www.extensions.init_security import init_xframe_protection
from airflow.www.extensions.init_views import init_api_internal, init_error_handlers
log = logging.getLogger(__name__)
app: Flask | None = None
@cli_utils.action_cli
def internal_api(args):
    """Starts Airflow Internal API."""
    print(settings.HEADER)
    access_logfile = args.access_logfile or "-"
    error_logfile = args.error_logfile or "-"
    access_logformat = args.access_logformat
    num_workers = args.workers
    worker_timeout = args.worker_timeout
    if args.debug:
        log.info(f"Starting the Internal API server on port {args.port} and host {args.hostname}.")
        app = create_app(testing=conf.getboolean("core", "unit_test_mode"))
        app.run(
            debug=True,
            use_reloader=not app.config["TESTING"],
            port=args.port,
            host=args.hostname,