_config.py 4.3 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495
  1. # Copyright 2023 Avaiga Private Limited
  2. #
  3. # Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
  4. # the License. You may obtain a copy of the License at
  5. #
  6. # http://www.apache.org/licenses/LICENSE-2.0
  7. #
  8. # Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
  9. # an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
  10. # specific language governing permissions and limitations under the License.
  11. from copy import copy
  12. from typing import Dict
  13. from .global_app.global_app_config import GlobalAppConfig
  14. from .section import Section
  15. from .unique_section import UniqueSection
  16. class _Config:
  17. DEFAULT_KEY = "default"
  18. def __init__(self):
  19. self._sections: Dict[str, Dict[str, Section]] = {}
  20. self._unique_sections: Dict[str, UniqueSection] = {}
  21. self._global_config: GlobalAppConfig = GlobalAppConfig()
  22. def _clean(self):
  23. self._global_config._clean()
  24. for unique_section in self._unique_sections.values():
  25. unique_section._clean()
  26. for sections in self._sections.values():
  27. for section in sections.values():
  28. section._clean()
  29. @classmethod
  30. def _default_config(cls):
  31. config = _Config()
  32. config._global_config = GlobalAppConfig.default_config()
  33. return config
  34. def _update(self, other_config):
  35. self._global_config._update(other_config._global_config._to_dict())
  36. if other_config._unique_sections:
  37. for section_name, other_section in other_config._unique_sections.items():
  38. if section := self._unique_sections.get(section_name, None):
  39. section._update(other_section._to_dict())
  40. else:
  41. self._unique_sections[section_name] = copy(other_config._unique_sections[section_name])
  42. if other_config._sections:
  43. for section_name, other_non_unique_sections in other_config._sections.items():
  44. if non_unique_sections := self._sections.get(section_name, None):
  45. self.__update_sections(non_unique_sections, other_non_unique_sections)
  46. else:
  47. self._sections[section_name] = {}
  48. self.__add_sections(self._sections[section_name], other_non_unique_sections)
  49. def __add_sections(self, entity_config, other_entity_configs):
  50. for cfg_id, sub_config in other_entity_configs.items():
  51. entity_config[cfg_id] = copy(sub_config)
  52. self.__point_nested_section_to_self(sub_config)
  53. def __update_sections(self, entity_config, other_entity_configs):
  54. if self.DEFAULT_KEY in other_entity_configs:
  55. if self.DEFAULT_KEY in entity_config:
  56. entity_config[self.DEFAULT_KEY]._update(other_entity_configs[self.DEFAULT_KEY]._to_dict())
  57. else:
  58. entity_config[self.DEFAULT_KEY] = other_entity_configs[self.DEFAULT_KEY]
  59. for cfg_id, sub_config in other_entity_configs.items():
  60. if cfg_id != self.DEFAULT_KEY:
  61. if cfg_id in entity_config:
  62. entity_config[cfg_id]._update(sub_config._to_dict(), entity_config.get(self.DEFAULT_KEY))
  63. else:
  64. entity_config[cfg_id] = copy(sub_config)
  65. entity_config[cfg_id]._update(sub_config._to_dict(), entity_config.get(self.DEFAULT_KEY))
  66. self.__point_nested_section_to_self(sub_config)
  67. def __point_nested_section_to_self(self, section):
  68. """Loop through attributes of a Section to find if any attribute has a list of Section as value.
  69. If there is, update each nested Section by the corresponding instance in self.
  70. Args:
  71. section (Section): The Section to search for nested sections.
  72. """
  73. for _, attr_value in vars(section).items():
  74. # ! This will fail if an attribute is a dictionary, or nested list of Sections.
  75. if not isinstance(attr_value, list):
  76. continue
  77. for index, item in enumerate(attr_value):
  78. if not isinstance(item, Section):
  79. continue
  80. if sub_item := self._sections.get(item.name, {}).get(item.id, None):
  81. attr_value[index] = sub_item