images/airflow/2.10.1/python/mwaa/config/airflow.py [256:333]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
    }


def _get_essential_airflow_webserver_config() -> Dict[str, str]:
    """
    Retrieve the environment variables for Airflow's "webserver" configuration section.

    :returns A dictionary containing the environment variables.
    """

    flask_secret_key = {}
    flask_secret_secret = os.environ.get("MWAA__WEBSERVER__SECRET")
    if flask_secret_secret:
        try:
            flask_secret_key = {
                "AIRFLOW__WEBSERVER__SECRET_KEY": json.loads(flask_secret_secret)[
                    "secret_key"
                ]
            }
        except:
            logger.warning(
                "Invalid value for the webserver secret key. Value not printed "
                "for security reasons.",
            )

    return {
        "AIRFLOW__WEBSERVER__CONFIG_FILE": "/python/mwaa/webserver/webserver_config.py",
        **flask_secret_key,
    }


def _get_essential_airflow_api_config() -> Dict[str, str]:
    """
    Retrieve the environment variables for Airflow's "api" configuration section.

    :returns A dictionary containing the environment variables.
    """
    api_config: Dict[str, str] = {}
    if os.environ.get("MWAA__CORE__AUTH_TYPE", "").lower() == "none":
        api_config["AIRFLOW__API__AUTH_BACKENDS"] = "airflow.api.auth.backend.default"

    return api_config


def get_essential_airflow_config(executor_type: str) -> Dict[str, str]:
    """
    Retrieve the environment variables required to set Airflow configurations.

    These environment variables are essential and cannot be overridden by the customer.

    :param executor_type A string indicating the type of executor to use.

    :returns A dictionary containing the environment variables.
    """
    return {
        **_get_essential_airflow_executor_config(executor_type),
        **_get_essential_airflow_core_config(),
        **_get_essential_airflow_db_config(),
        **_get_essential_airflow_logging_config(),
        **_get_mwaa_cloudwatch_integration_config(),
        **_get_essential_airflow_scheduler_config(),
        **_get_essential_airflow_webserver_config(),
        **_get_essential_airflow_api_config(),
    }


def get_opinionated_airflow_config() -> Dict[str, str]:
    """
    Retrieve the environment variables required to set Airflow configurations.

    These environment variables can be overridden by the customer.

    :returns A dictionary containing the environment variables.
    """
    return {
        **_get_opinionated_airflow_core_config(),
        **_get_opinionated_airflow_scheduler_config(),
        **_get_opinionated_airflow_secrets_config(),
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -



images/airflow/2.9.2/python/mwaa/config/airflow.py [243:320]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
    }


def _get_essential_airflow_webserver_config() -> Dict[str, str]:
    """
    Retrieve the environment variables for Airflow's "webserver" configuration section.

    :returns A dictionary containing the environment variables.
    """

    flask_secret_key = {}
    flask_secret_secret = os.environ.get("MWAA__WEBSERVER__SECRET")
    if flask_secret_secret:
        try:
            flask_secret_key = {
                "AIRFLOW__WEBSERVER__SECRET_KEY": json.loads(flask_secret_secret)[
                    "secret_key"
                ]
            }
        except:
            logger.warning(
                "Invalid value for the webserver secret key. Value not printed "
                "for security reasons.",
            )

    return {
        "AIRFLOW__WEBSERVER__CONFIG_FILE": "/python/mwaa/webserver/webserver_config.py",
        **flask_secret_key,
    }


def _get_essential_airflow_api_config() -> Dict[str, str]:
    """
    Retrieve the environment variables for Airflow's "api" configuration section.

    :returns A dictionary containing the environment variables.
    """
    api_config: Dict[str, str] = {}
    if os.environ.get("MWAA__CORE__AUTH_TYPE", "").lower() == "none":
        api_config["AIRFLOW__API__AUTH_BACKENDS"] = "airflow.api.auth.backend.default"

    return api_config


def get_essential_airflow_config(executor_type: str) -> Dict[str, str]:
    """
    Retrieve the environment variables required to set Airflow configurations.

    These environment variables are essential and cannot be overridden by the customer.

    :param executor_type A string indicating the type of executor to use.

    :returns A dictionary containing the environment variables.
    """
    return {
        **_get_essential_airflow_executor_config(executor_type),
        **_get_essential_airflow_core_config(),
        **_get_essential_airflow_db_config(),
        **_get_essential_airflow_logging_config(),
        **_get_mwaa_cloudwatch_integration_config(),
        **_get_essential_airflow_scheduler_config(),
        **_get_essential_airflow_webserver_config(),
        **_get_essential_airflow_api_config(),
    }


def get_opinionated_airflow_config() -> Dict[str, str]:
    """
    Retrieve the environment variables required to set Airflow configurations.

    These environment variables can be overridden by the customer.

    :returns A dictionary containing the environment variables.
    """
    return {
        **_get_opinionated_airflow_core_config(),
        **_get_opinionated_airflow_scheduler_config(),
        **_get_opinionated_airflow_secrets_config(),
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -