Scheduler/Webserver should never use Internal API, so calling any method decorated with internal_api_call should still execute them locally
umask=int(settings.DAEMON_UMASK, 8),
                )
                with ctx:
                    subprocess.Popen(run_args, close_fds=True)
                    
                    
                    while True:
                        sleep(0.1)
                        gunicorn_master_proc_pid = read_pid_from_pidfile(pid_file)
                        if gunicorn_master_proc_pid:
                            break
                    
                    gunicorn_master_proc = psutil.Process(gunicorn_master_proc_pid)
                    monitor_gunicorn(gunicorn_master_proc.pid)
        else:
            with subprocess.Popen(run_args, close_fds=True) as gunicorn_master_proc:
                monitor_gunicorn(gunicorn_master_proc.pid)
def create_app(config=None, testing=False):
    """Create a new instance of Airflow Internal API app"""
    flask_app = Flask(__name__)
    flask_app.config["APP_NAME"] = "Airflow Internal API"
    flask_app.config["TESTING"] = testing
    flask_app.config["SQLALCHEMY_DATABASE_URI"] = conf.get("database", "SQL_ALCHEMY_CONN")
    url = make_url(flask_app.config["SQLALCHEMY_DATABASE_URI"])
    if url.drivername == "sqlite" and url.database and not url.database.startswith("/"):
        raise AirflowConfigException(
            f'Cannot use relative path: `{conf.get("database", "SQL_ALCHEMY_CONN")}` to connect to sqlite. '
            "Please use absolute path such as `sqlite:////tmp/airflow.db`."
        )
    flask_app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False