interpolation.py 2.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384
  1. import logging
  2. import os
  3. from string import Template
  4. import six
  5. from .errors import ConfigurationError
  6. log = logging.getLogger(__name__)
  7. def interpolate_environment_variables(service_dicts):
  8. mapping = BlankDefaultDict(os.environ)
  9. return dict(
  10. (service_name, process_service(service_name, service_dict, mapping))
  11. for (service_name, service_dict) in service_dicts.items()
  12. )
  13. def process_service(service_name, service_dict, mapping):
  14. return dict(
  15. (key, interpolate_value(service_name, key, val, mapping))
  16. for (key, val) in service_dict.items()
  17. )
  18. def interpolate_value(service_name, config_key, value, mapping):
  19. try:
  20. return recursive_interpolate(value, mapping)
  21. except InvalidInterpolation as e:
  22. raise ConfigurationError(
  23. 'Invalid interpolation format for "{config_key}" option '
  24. 'in service "{service_name}": "{string}"'
  25. .format(
  26. config_key=config_key,
  27. service_name=service_name,
  28. string=e.string,
  29. )
  30. )
  31. def recursive_interpolate(obj, mapping):
  32. if isinstance(obj, six.string_types):
  33. return interpolate(obj, mapping)
  34. elif isinstance(obj, dict):
  35. return dict(
  36. (key, recursive_interpolate(val, mapping))
  37. for (key, val) in obj.items()
  38. )
  39. elif isinstance(obj, list):
  40. return [recursive_interpolate(val, mapping) for val in obj]
  41. else:
  42. return obj
  43. def interpolate(string, mapping):
  44. try:
  45. return Template(string).substitute(mapping)
  46. except ValueError:
  47. raise InvalidInterpolation(string)
  48. class BlankDefaultDict(dict):
  49. def __init__(self, *args, **kwargs):
  50. super(BlankDefaultDict, self).__init__(*args, **kwargs)
  51. self.missing_keys = []
  52. def __getitem__(self, key):
  53. try:
  54. return super(BlankDefaultDict, self).__getitem__(key)
  55. except KeyError:
  56. if key not in self.missing_keys:
  57. log.warn(
  58. "The {} variable is not set. Defaulting to a blank string."
  59. .format(key)
  60. )
  61. self.missing_keys.append(key)
  62. return ""
  63. class InvalidInterpolation(Exception):
  64. def __init__(self, string):
  65. self.string = string