import logging, os from typing import Optional, Union from .backend._Backend import _Backend from .backend.BackendEnum import BackendEnum from .backend.inmemory.InMemoryBackend import InMemoryBackend from .backend.redis.RedisBackend import RedisBackend LOGGER = logging.getLogger(__name__) BACKENDS = { BackendEnum.INMEMORY.value: InMemoryBackend, BackendEnum.REDIS.value: RedisBackend, #BackendEnum.MONGODB.value: MongoDBBackend, #BackendEnum.RETHINKDB.value: RethinkDBBackend, #BackendEnum.ETCD.value: EtcdBackend, } DEFAULT_DB_BACKEND = BackendEnum.INMEMORY def get_database_backend(backend : Optional[Union[str, BackendEnum]] = None, **settings) -> _Backend: # return an instance of Database initialized with selected backend. # The backend is selected using following criteria (first that is not None is selected): # 1. user selected by parameter (backend=...) # 2. environment variable DB_BACKEND # 3. default backend: INMEMORY if backend is None: backend = os.environ.get('DB_BACKEND', DEFAULT_DB_BACKEND) if backend is None: raise Exception('Database Backend not specified') if isinstance(backend, BackendEnum): backend = backend.value backend_class = BACKENDS.get(backend) if backend_class is None: raise Exception('Unsupported DatabaseBackend({})'.format(backend)) LOGGER.info('Selected Database Backend: {}'.format(backend)) return backend_class(**settings)