airflow/contrib/kubernetes/worker_configuration.py
def _get_environment(self):
        """Defines any necessary environment variables for the pod executor"""
        env = {}

        for env_var_name, env_var_val in six.iteritems(self.kube_config.kube_env_vars):
            env[env_var_name] = env_var_val

        env["AIRFLOW__CORE__EXECUTOR"] = "LocalExecutor"

        if self.kube_config.airflow_configmap:
            env['AIRFLOW_HOME'] = self.worker_airflow_home
            env['AIRFLOW__CORE__DAGS_FOLDER'] = self.worker_airflow_dags
        if (not self.kube_config.airflow_configmap and
                'AIRFLOW__CORE__SQL_ALCHEMY_CONN' not in self.kube_config.kube_secrets):
            env['AIRFLOW__CORE__SQL_ALCHEMY_CONN'] = conf.get("core", "SQL_ALCHEMY_CONN")
        if self.kube_config.git_dags_folder_mount_point:
            # /root/airflow/dags/repo/dags
            dag_volume_mount_path = os.path.join(
                self.kube_config.git_dags_folder_mount_point,
                self.kube_config.git_sync_dest,  # repo
                self.kube_config.git_subpath     # dags
            )
            env['AIRFLOW__CORE__DAGS_FOLDER'] = dag_volume_mount_path
        return env
def _get_environment(self):
        """Defines any necessary environment variables for the pod executor"""
        env = {}

        for env_var_name, env_var_val in six.iteritems(self.kube_config.kube_env_vars):
            env[env_var_name] = env_var_val

        env["AIRFLOW__CORE__EXECUTOR"] = "LocalExecutor"

        if self.kube_config.airflow_configmap:
            env['AIRFLOW_HOME'] = self.worker_airflow_home
            env['AIRFLOW__CORE__DAGS_FOLDER'] = self.worker_airflow_dags
        if (not self.kube_config.airflow_configmap and
                'AIRFLOW__CORE__SQL_ALCHEMY_CONN' not in self.kube_config.kube_secrets):
            env['AIRFLOW__CORE__SQL_ALCHEMY_CONN'] = conf.get("core", "SQL_ALCHEMY_CONN")
        if self.kube_config.git_dags_folder_mount_point:
            # /root/airflow/dags/repo/dags
            dag_volume_mount_path = os.path.join(
                self.kube_config.git_dags_folder_mount_point,
                self.kube_config.git_sync_dest,  # repo
                self.kube_config.git_subpath     # dags
            )
            env['AIRFLOW__CORE__DAGS_FOLDER'] = dag_volume_mount_path
        return env
def _get_environment ( self ) : env = { } for env_var_name , env_var_val in six . iteritems ( self . kube_config . kube_env_vars ) : env [ env_var_name ] = env_var_val env [ "AIRFLOW__CORE__EXECUTOR" ] = "LocalExecutor" if self . kube_config . airflow_configmap : env [ 'AIRFLOW_HOME' ] = self . worker_airflow_home env [ 'AIRFLOW__CORE__DAGS_FOLDER' ] = self . worker_airflow_dags if ( not self . kube_config . airflow_configmap and 'AIRFLOW__CORE__SQL_ALCHEMY_CONN' not in self . kube_config . kube_secrets ) : env [ 'AIRFLOW__CORE__SQL_ALCHEMY_CONN' ] = conf . get ( "core" , "SQL_ALCHEMY_CONN" ) if self . kube_config . git_dags_folder_mount_point : # /root/airflow/dags/repo/dags dag_volume_mount_path = os . path . join ( self . kube_config . git_dags_folder_mount_point , self . kube_config . git_sync_dest , # repo self . kube_config . git_subpath # dags ) env [ 'AIRFLOW__CORE__DAGS_FOLDER' ] = dag_volume_mount_path return env
Defines any necessary environment variables for the pod executor
Defines any necessary environment variables for the pod executor