src/budget_sync/config.py

   1 import toml
   2 import sys
   3 from typing import Set, Dict, Any, Optional
   4 from functools import cached_property
   5
   6
   7 class ConfigParseError(Exception):
   8     pass
   9
  10
  11 class Person:
  12     aliases: Set[str]
  13     email: Optional[str]
  14
  15     def __init__(self, config: "Config", identifier: str,
  16                  aliases: Optional[Set[str]] = None,
  17                  email: Optional[str] = None):
  18         self.config = config
  19         self.identifier = identifier
  20         if aliases is None:
  21             aliases = set()
  22         self.aliases = aliases
  23         self.email = email
  24
  25     @cached_property
  26     def all_names(self) -> Set[str]:
  27         retval = self.aliases.copy()
  28         retval.add(self.identifier)
  29         return retval
  30
  31     def __eq__(self, other):
  32         return self.identifier == other.identifier
  33
  34     def __hash__(self):
  35         return hash(self.identifier)
  36
  37     def __repr__(self):
  38         return f"Person(config=..., identifier={self.identifier!r}, " \
  39             f"aliases={self.aliases!r}, email={self.email!r})"
  40
  41
  42 class Config:
  43     def __init__(self, bugzilla_url: str, people: Dict[str, Person]):
  44         self.bugzilla_url = bugzilla_url
  45         self.people = people
  46
  47     def __repr__(self):
  48         return f"Config(bugzilla_url={self.bugzilla_url!r}, " \
  49             f"people={self.people!r})"
  50
  51     @cached_property
  52     def all_names(self) -> Dict[str, Person]:
  53         # also checks for any name clashes and raises
  54         # ConfigParseError if any are detected
  55         retval = self.people.copy()
  56
  57         for person in self.people.values():
  58             for alias in person.aliases:
  59                 other_person = retval.get(alias)
  60                 if other_person is not None:
  61                     if alias in self.people:
  62                         raise ConfigParseError(
  63                             f"alias is not allowed to be the same as any"
  64                             f" person's identifier: in person entry for "
  65                             f"{person.identifier!r}: {alias!r} is also the "
  66                             f"identifier for person"
  67                             f" {other_person.identifier!r}")
  68                     raise ConfigParseError(
  69                         f"alias is not allowed to be the same as another"
  70                         f" person's alias: in person entry for "
  71                         f"{person.identifier!r}: {alias!r} is also an alias "
  72                         f"for person {other_person.identifier!r}")
  73                 retval[alias] = person
  74         return retval
  75
  76     def _parse_person(self, identifier: str, value: Any) -> Person:
  77         def raise_aliases_must_be_list_of_strings():
  78             raise ConfigParseError(
  79                 f"`aliases` field in person entry for {identifier!r} must "
  80                 f"be a list of strings")
  81
  82         if not isinstance(value, dict):
  83             raise ConfigParseError(
  84                 f"person entry for {identifier!r} must be a table")
  85         aliases = set()
  86         email = None
  87         for k, v in value.items():
  88             assert isinstance(k, str)
  89             if k == "aliases":
  90                 if not isinstance(v, list):
  91                     raise_aliases_must_be_list_of_strings()
  92                 for alias in v:
  93                     if isinstance(alias, str):
  94                         if alias in aliases:
  95                             raise ConfigParseError(
  96                                 f"duplicate alias in person entry for "
  97                                 f"{identifier!r}: {alias!r}")
  98                         aliases.add(alias)
  99                     else:
 100                         raise_aliases_must_be_list_of_strings()
 101             elif k == "email":
 102                 if not isinstance(v, str):
 103                     raise ConfigParseError(
 104                         f"`email` field in person entry for {identifier!r} "
 105                         f"must be a string")
 106                 email = v
 107             else:
 108                 raise ConfigParseError(
 109                     f"unknown field in person entry for {identifier!r}: `{k}`")
 110         return Person(config=self, identifier=identifier,
 111                       aliases=aliases, email=email)
 112
 113     def _parse_people(self, people: Any):
 114         if not isinstance(people, dict):
 115             raise ConfigParseError("`people` field must be a table")
 116         for identifier, value in people.items():
 117             assert isinstance(identifier, str)
 118             self.people[identifier] = self._parse_person(identifier, value)
 119
 120         # self.all_names checks for name clashes and raises ConfigParseError
 121         # if any are detected, so the following line is needed:
 122         self.all_names
 123
 124     @staticmethod
 125     def _from_toml(parsed_toml: Dict[str, Any]) -> "Config":
 126         people = None
 127         bugzilla_url = None
 128         for k, v in parsed_toml.items():
 129             assert isinstance(k, str)
 130             if k == "people":
 131                 people = v
 132             elif k == "bugzilla_url":
 133                 if not isinstance(v, str):
 134                     raise ConfigParseError("`bugzilla_url` must be a string")
 135                 bugzilla_url = v
 136             else:
 137                 raise ConfigParseError(f"unknown config entry: `{k}`")
 138
 139         if bugzilla_url is None:
 140             raise ConfigParseError("`bugzilla_url` field is missing")
 141
 142         config = Config(bugzilla_url=bugzilla_url, people={})
 143
 144         if people is None:
 145             raise ConfigParseError("`people` table is missing")
 146         else:
 147             config._parse_people(people)
 148
 149         return config
 150
 151     @staticmethod
 152     def from_str(text: str) -> "Config":
 153         try:
 154             parsed_toml = toml.loads(text)
 155         except toml.TomlDecodeError as e:
 156             new_err = ConfigParseError(f"TOML parse error: {e}")
 157             raise new_err.with_traceback(sys.exc_info()[2])
 158         return Config._from_toml(parsed_toml)
 159
 160     @staticmethod
 161     def from_file(file: Any) -> "Config":
 162         if isinstance(file, list):
 163             raise TypeError("list is not a valid file or path")
 164         try:
 165             parsed_toml = toml.load(file)
 166         except toml.TomlDecodeError as e:
 167             new_err = ConfigParseError(f"TOML parse error: {e}")
 168             raise new_err.with_traceback(sys.exc_info()[2])
 169         return Config._from_toml(parsed_toml)