File: /var/www/vhosts/dial-copper-dev.2amigos.us/dialpad-copper-crm/api/conf/config.py
"""Module with configuration access BL
"""
import os
import re
from enum import Enum
from os.path import dirname, join as path_join
from pathlib import Path
from typing import Union
import yaml
from pycoppercrm.networking import (
create_session, CopperCrmSession, SessionMode
)
from api.common.crypto import encrypt_content
from api.common.utils import extract_from_json
_SETTING_RE = re.compile(r'^(\w+\.)+\w+$')
_CONFIG_FILE_PATH = path_join(Path(dirname(__file__)), 'config.yml')
def _get_setting(settings: dict, setting_code: str) -> Union[str, list, int]:
"""Gets a setting by code e.g: db.conn_str = settings[db][conn_str]
Args:
settings (dict): settings dictionary
setting_code (str): code of the setting to be splitted by dots
Returns:
Union[str, list, int]: Setting value
"""
return extract_from_json(settings, setting_code)
def _dump_yaml(settings: dict, file_path: str):
"""Dumps a dict into a yml
Args:
settings (dict): settings to be written
file_path (str): yaml file path
"""
with open(file_path, 'w') as f:
yaml.dump(settings, f)
def _load_yaml(file_path: str) -> dict:
"""Loads a yaml file
Args:
file_path (str): file path
Returns:
dict: loaded settings
"""
with open(file_path, 'r') as f:
return yaml.load(f.read(), yaml.SafeLoader)
class ConfigScope(Enum):
"""Enumeration with supported config scopes
"""
# Unit tests
UNIT_TEST = 1
# Local or remote server
SERVER = 2
@property
def is_server(self) -> bool:
"""Boolean flag to determine if the config scope is
on server or unit tests/
Returns:
bool -- True if SERVER False otherwise
"""
return self.value == 2
class ConfigWrapper:
"""Yaml Config accessor"""
def __init__(self, settings: dict):
self._settings = settings
@property
def config_scope(self) -> ConfigScope:
"""Current configuration scope
Returns:
ConfigScope: enumeration of current app running scope
"""
# Defaulting to server to ensure runtime stability
app_scope = os.environ.get("DC_APP_SCOPE", ConfigScope.SERVER.name)
return ConfigScope[app_scope]
# NOTE: See note on self.set_app_scope
# return ConfigScope[self.get_setting('app.scope')]
@property
def db_cstr_cipher(self) -> str:
"""Encrypted db connection str
Returns:
dict: encrypted settings
"""
scope = self.config_scope
setting = 'app_crypto.app_server_db' if scope.is_server else 'app.unit_test_db'
return self.get_setting(setting)
@property
def dialpad_oauth_url(self) -> str:
"""Convinience property to access the configured dialpad OAuth URL
"""
return self.get_setting('dialpad.oauth_url')
@property
def dialpad_logout_url(self) -> str:
"""Convinience property to access the configured dialpad Logout URL
"""
return self.get_setting('dialpad.logout_url')
@property
def dialpad_oauth_secret(self) -> str:
"""Convinience property to access the configured dialpad OAuth Secret
"""
return self.get_setting('dialpad.oauth_secret')
@property
def gds_credentials_path(self) -> str:
"""Full path for Google data store credentials"""
file_dir = Path(dirname(__file__))
file_name = self.get_setting('app_crypto.app_gds_credentials_file')
return path_join(file_dir, file_name)
def get_setting(self, setting_code: str) -> Union[str, list, int]:
"""Gets a setting by code e.g: db.conn_str = settings[db][conn_str]
Args:
setting_code (str): code of the setting to be splitted by dots
Returns:
Union[str, list, int]: Setting value
"""
if not _SETTING_RE.match(setting_code):
raise ValueError(f'Invalid setting code: {setting_code}')
return _get_setting(self._settings, setting_code)
def update_setting(self, setting_code: str, new_value: str, encrypt=False) -> str:
"""Updates a setting in the config file
Args:
setting_code (str): code or path to the setting
new_value (str): value to update
encrypt (bool, optional): Encrypt using master.key. Defaults to False.
Returns:
str: old setting value
"""
setting_path = setting_code.split('.')
setting = self._settings.get(next(iter(setting_path)))
for i in range(1, len(setting_path) - 1):
setting = setting.get(setting_path[i])
key = setting_path.pop()
old = setting.get(key)
setting[key] = new_value if not encrypt else encrypt_content(new_value)
return old
def save(self):
"""updates the settings with this class values
"""
env_settings = {k: v for k, v in self._settings.items() if k != 'app'}
main_settings = {k: v for k, v in self._settings.items() if k == 'app'}
env_path = path_join(Path(dirname(__file__)), self.get_setting('app.full_config_file'))
_dump_yaml(main_settings, _CONFIG_FILE_PATH)
_dump_yaml(env_settings, env_path)
def copper_session(self, copper_key: str, user_email: str) -> CopperCrmSession:
"""Builds a CopperCrm session based on the config
Args:
copper_key (str): Copper API Key
user_email (str): Copper user email
Returns:
CopperCrmSession: CopperCrmSession
"""
mode = SessionMode.SANDBOX if self.get_setting('copper.sandbox') else SessionMode.LIVE
return create_session(copper_key, user_email, mode)
@staticmethod
def read_settings() -> dict:
"""Read settings from config.yml path
Returns:
dict: config.yml settings
"""
main_config = _load_yaml(_CONFIG_FILE_PATH)
file_path = _get_setting(main_config, 'app.full_config_file')
env_path = path_join(Path(dirname(__file__)), file_path)
env_config = _load_yaml(env_path)
return {**main_config, **env_config}
@classmethod
def read_config(cls) -> 'ConfigWrapper':
"""Instance from api configuration file
Returns:
ConfigWrapper: App Settings accessor
"""
return cls(cls.read_settings())
def set_app_scope(scope: ConfigScope):
"""Sets an scope in the config file
Args:
scope (ConfigScope): Running app scope
"""
# NOTE: Google cloud doesn't support updates on local FS
# Using env variables for this.
os.environ["DC_APP_SCOPE"] = scope.name
# wrapper = ConfigWrapper.read_config()
# wrapper.update_setting('app.scope', scope.name)
# wrapper.save()