import logging

from airflow import configuration
from airflow.executors.base_executor import BaseExecutor
from airflow.executors.local_executor import LocalExecutor
from airflow.executors.sequential_executor import SequentialExecutor

# TODO Fix this emergency fix
try:
    from airflow.executors.celery_executor import CeleryExecutor
except:
    pass
try:
    from airflow.contrib.executors.mesos_executor import MesosExecutor
except:
    pass

from airflow.utils import AirflowException

_EXECUTOR = configuration.get('core', 'EXECUTOR')

if _EXECUTOR == 'LocalExecutor':
    DEFAULT_EXECUTOR = LocalExecutor()
elif _EXECUTOR == 'CeleryExecutor':
    DEFAULT_EXECUTOR = CeleryExecutor()
elif _EXECUTOR == 'SequentialExecutor':
    DEFAULT_EXECUTOR = SequentialExecutor()
elif _EXECUTOR == 'MesosExecutor':
    DEFAULT_EXECUTOR = MesosExecutor()
else:
    # Loading plugins
    from airflow.plugins_manager import executors as _executors
    for _executor in _executors:
        globals()[_executor.__name__] = _executor
    if _EXECUTOR in globals():
        DEFAULT_EXECUTOR = globals()[_EXECUTOR]()
    else:
        raise AirflowException("Executor {0} not supported.".format(_EXECUTOR))

logging.info("Using executor " + _EXECUTOR)

import logging

from airflow import configuration
from airflow.executors.base_executor import BaseExecutor
from airflow.executors.local_executor import LocalExecutor
from airflow.executors.sequential_executor import SequentialExecutor

from airflow.utils import AirflowException

_EXECUTOR = configuration.get('core', 'EXECUTOR')

if _EXECUTOR == 'LocalExecutor':
    DEFAULT_EXECUTOR = LocalExecutor()
elif _EXECUTOR == 'CeleryExecutor':
    from airflow.executors.celery_executor import CeleryExecutor
    DEFAULT_EXECUTOR = CeleryExecutor()
elif _EXECUTOR == 'SequentialExecutor':
    DEFAULT_EXECUTOR = SequentialExecutor()
elif _EXECUTOR == 'MesosExecutor':
    from airflow.contrib.executors.mesos_executor import MesosExecutor
    DEFAULT_EXECUTOR = MesosExecutor()
else:
    # Loading plugins
    from airflow.plugins_manager import executors as _executors
    for _executor in _executors:
        globals()[_executor.__name__] = _executor
    if _EXECUTOR in globals():
        DEFAULT_EXECUTOR = globals()[_EXECUTOR]()
    else:
        raise AirflowException("Executor {0} not supported.".format(_EXECUTOR))

logging.info("Using executor " + _EXECUTOR)

Remove hack by only importing when configured
Remove hack by only importing when configured

  import logging
  
  from airflow import configuration
  from airflow.executors.base_executor import BaseExecutor
  from airflow.executors.local_executor import LocalExecutor
  from airflow.executors.sequential_executor import SequentialExecutor
- 
- # TODO Fix this emergency fix
- try:
-     from airflow.executors.celery_executor import CeleryExecutor
- except:
-     pass
- try:
-     from airflow.contrib.executors.mesos_executor import MesosExecutor
- except:
-     pass
  
  from airflow.utils import AirflowException
  
  _EXECUTOR = configuration.get('core', 'EXECUTOR')
  
  if _EXECUTOR == 'LocalExecutor':
      DEFAULT_EXECUTOR = LocalExecutor()
  elif _EXECUTOR == 'CeleryExecutor':
+     from airflow.executors.celery_executor import CeleryExecutor
      DEFAULT_EXECUTOR = CeleryExecutor()
  elif _EXECUTOR == 'SequentialExecutor':
      DEFAULT_EXECUTOR = SequentialExecutor()
  elif _EXECUTOR == 'MesosExecutor':
+     from airflow.contrib.executors.mesos_executor import MesosExecutor
      DEFAULT_EXECUTOR = MesosExecutor()
  else:
      # Loading plugins
      from airflow.plugins_manager import executors as _executors
      for _executor in _executors:
          globals()[_executor.__name__] = _executor
      if _EXECUTOR in globals():
          DEFAULT_EXECUTOR = globals()[_EXECUTOR]()
      else:
          raise AirflowException("Executor {0} not supported.".format(_EXECUTOR))
  
  logging.info("Using executor " + _EXECUTOR)
  
Remove hack by only importing when configured
## Code Before:
import logging

from airflow import configuration
from airflow.executors.base_executor import BaseExecutor
from airflow.executors.local_executor import LocalExecutor
from airflow.executors.sequential_executor import SequentialExecutor

# TODO Fix this emergency fix
try:
    from airflow.executors.celery_executor import CeleryExecutor
except:
    pass
try:
    from airflow.contrib.executors.mesos_executor import MesosExecutor
except:
    pass

from airflow.utils import AirflowException

_EXECUTOR = configuration.get('core', 'EXECUTOR')

if _EXECUTOR == 'LocalExecutor':
    DEFAULT_EXECUTOR = LocalExecutor()
elif _EXECUTOR == 'CeleryExecutor':
    DEFAULT_EXECUTOR = CeleryExecutor()
elif _EXECUTOR == 'SequentialExecutor':
    DEFAULT_EXECUTOR = SequentialExecutor()
elif _EXECUTOR == 'MesosExecutor':
    DEFAULT_EXECUTOR = MesosExecutor()
else:
    # Loading plugins
    from airflow.plugins_manager import executors as _executors
    for _executor in _executors:
        globals()[_executor.__name__] = _executor
    if _EXECUTOR in globals():
        DEFAULT_EXECUTOR = globals()[_EXECUTOR]()
    else:
        raise AirflowException("Executor {0} not supported.".format(_EXECUTOR))

logging.info("Using executor " + _EXECUTOR)

## Instruction:
Remove hack by only importing when configured
## Code After:
import logging

from airflow import configuration
from airflow.executors.base_executor import BaseExecutor
from airflow.executors.local_executor import LocalExecutor
from airflow.executors.sequential_executor import SequentialExecutor

from airflow.utils import AirflowException

_EXECUTOR = configuration.get('core', 'EXECUTOR')

if _EXECUTOR == 'LocalExecutor':
    DEFAULT_EXECUTOR = LocalExecutor()
elif _EXECUTOR == 'CeleryExecutor':
    from airflow.executors.celery_executor import CeleryExecutor
    DEFAULT_EXECUTOR = CeleryExecutor()
elif _EXECUTOR == 'SequentialExecutor':
    DEFAULT_EXECUTOR = SequentialExecutor()
elif _EXECUTOR == 'MesosExecutor':
    from airflow.contrib.executors.mesos_executor import MesosExecutor
    DEFAULT_EXECUTOR = MesosExecutor()
else:
    # Loading plugins
    from airflow.plugins_manager import executors as _executors
    for _executor in _executors:
        globals()[_executor.__name__] = _executor
    if _EXECUTOR in globals():
        DEFAULT_EXECUTOR = globals()[_EXECUTOR]()
    else:
        raise AirflowException("Executor {0} not supported.".format(_EXECUTOR))

logging.info("Using executor " + _EXECUTOR)
