src/osm_ngsa/osm_mon/core/config.py

   1 #######################################################################################
   2 # Copyright ETSI Contributors and Others.
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at
   7 #
   8 #    http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  13 # implied.
  14 # See the License for the specific language governing permissions and
  15 # limitations under the License.
  16 #######################################################################################
  17 """Global configuration managed by environment variables."""
  18
  19 import logging
  20 import os
  21
  22 import pkg_resources
  23 import yaml
  24
  25 logger = logging.getLogger(__name__)
  26
  27
  28 class Config:
  29     def __init__(self, config_file: str = ""):
  30         self.conf = {}
  31         self._read_config_file(config_file)
  32         self._read_env()
  33
  34     def _read_config_file(self, config_file):
  35         if not config_file:
  36             path = "config.yaml"
  37             config_file = pkg_resources.resource_filename(__name__, path)
  38         with open(config_file) as f:
  39             self.conf = yaml.safe_load(f)
  40
  41     def get(self, section, field=None):
  42         if not field:
  43             return self.conf[section]
  44         return self.conf[section].get(field)
  45
  46     def set(self, section, field, value):
  47         if section not in self.conf:
  48             self.conf[section] = {}
  49         self.conf[section][field] = value
  50
  51     def _read_env(self):
  52         for env in os.environ:
  53             if not env.startswith("OSMMON_"):
  54                 continue
  55             elements = env.lower().split("_")
  56             if len(elements) < 3:
  57                 logger.warning(
  58                     "Environment variable %s=%s does not comply with required format. Section and/or field missing.",
  59                     env,
  60                     os.getenv(env),
  61                 )
  62                 continue
  63             section = elements[1]
  64             field = "_".join(elements[2:])
  65             value = os.getenv(env)
  66             if section not in self.conf:
  67                 self.conf[section] = {}
  68             self.conf[section][field] = value