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 Milestone:
  43     def __init__(self, config: "Config",
  44                  identifier: str, canonical_bug_id: int):
  45         self.config = config
  46         self.identifier = identifier
  47         self.canonical_bug_id = canonical_bug_id
  48
  49     def __eq__(self, other):
  50         return self.identifier == other.identifier
  51
  52     def __hash__(self):
  53         return hash(self.identifier)
  54
  55     def __repr__(self):
  56         return f"Milestone(config=..., " \
  57             f"identifier={self.identifier!r}, " \
  58             f"canonical_bug_id={self.canonical_bug_id})"
  59
  60
  61 class Config:
  62     def __init__(self, bugzilla_url: str, people: Dict[str, Person],
  63                  milestones: Dict[str, Milestone]):
  64         self.bugzilla_url = bugzilla_url
  65         self.people = people
  66         self.milestones = milestones
  67
  68     def __repr__(self):
  69         return f"Config(bugzilla_url={self.bugzilla_url!r}, " \
  70             f"people={self.people!r}, " \
  71             f"milestones={self.milestones!r})"
  72
  73     @cached_property
  74     def all_names(self) -> Dict[str, Person]:
  75         # also checks for any name clashes and raises
  76         # ConfigParseError if any are detected
  77         retval = self.people.copy()
  78
  79         for person in self.people.values():
  80             for alias in person.aliases:
  81                 other_person = retval.get(alias)
  82                 if other_person is not None:
  83                     if alias in self.people:
  84                         raise ConfigParseError(
  85                             f"alias is not allowed to be the same as any"
  86                             f" person's identifier: in person entry for "
  87                             f"{person.identifier!r}: {alias!r} is also the "
  88                             f"identifier for person"
  89                             f" {other_person.identifier!r}")
  90                     raise ConfigParseError(
  91                         f"alias is not allowed to be the same as another"
  92                         f" person's alias: in person entry for "
  93                         f"{person.identifier!r}: {alias!r} is also an alias "
  94                         f"for person {other_person.identifier!r}")
  95                 retval[alias] = person
  96         return retval
  97
  98     @cached_property
  99     def canonical_bug_ids(self) -> Dict[int, Milestone]:
 100         # also checks for any bug id clashes and raises
 101         # ConfigParseError if any are detected
 102         retval = {}
 103         for milestone in self.milestones.values():
 104             other_milestone = retval.get(milestone.canonical_bug_id)
 105             if other_milestone is not None:
 106                 raise ConfigParseError(
 107                     f"canonical_bug_id is not allowed to be the same as "
 108                     f"another milestone's canonical_bug_id: in milestone "
 109                     f"entry for {milestone.identifier!r}: "
 110                     f"{milestone.canonical_bug_id} is also the "
 111                     f"canonical_bug_id for milestone "
 112                     f"{other_milestone.identifier!r}")
 113             retval[milestone.canonical_bug_id] = milestone
 114         return retval
 115
 116     def _parse_person(self, identifier: str, value: Any) -> Person:
 117         def raise_aliases_must_be_list_of_strings():
 118             raise ConfigParseError(
 119                 f"`aliases` field in person entry for {identifier!r} must "
 120                 f"be a list of strings")
 121
 122         if not isinstance(value, dict):
 123             raise ConfigParseError(
 124                 f"person entry for {identifier!r} must be a table")
 125         aliases = set()
 126         email = None
 127         for k, v in value.items():
 128             assert isinstance(k, str)
 129             if k == "aliases":
 130                 if not isinstance(v, list):
 131                     raise_aliases_must_be_list_of_strings()
 132                 for alias in v:
 133                     if isinstance(alias, str):
 134                         if alias in aliases:
 135                             raise ConfigParseError(
 136                                 f"duplicate alias in person entry for "
 137                                 f"{identifier!r}: {alias!r}")
 138                         aliases.add(alias)
 139                     else:
 140                         raise_aliases_must_be_list_of_strings()
 141             elif k == "email":
 142                 if not isinstance(v, str):
 143                     raise ConfigParseError(
 144                         f"`email` field in person entry for {identifier!r} "
 145                         f"must be a string")
 146                 email = v
 147             else:
 148                 raise ConfigParseError(
 149                     f"unknown field in person entry for {identifier!r}: `{k}`")
 150         return Person(config=self, identifier=identifier,
 151                       aliases=aliases, email=email)
 152
 153     def _parse_people(self, people: Any):
 154         if not isinstance(people, dict):
 155             raise ConfigParseError("`people` field must be a table")
 156         for identifier, value in people.items():
 157             assert isinstance(identifier, str)
 158             self.people[identifier] = self._parse_person(identifier, value)
 159
 160         # self.all_names checks for name clashes and raises ConfigParseError
 161         # if any are detected, so the following line is needed:
 162         self.all_names
 163
 164     def _parse_milestone(self, identifier: str, value: Any) -> Milestone:
 165         if not isinstance(value, dict):
 166             raise ConfigParseError(
 167                 f"milestones entry for {identifier!r} must be a table")
 168         canonical_bug_id = None
 169         for k, v in value.items():
 170             assert isinstance(k, str)
 171             if k == "canonical_bug_id":
 172                 if not isinstance(v, int):
 173                     raise ConfigParseError(
 174                         f"`canonical_bug_id` field in milestones entry for "
 175                         f"{identifier!r} must be an integer")
 176                 canonical_bug_id = v
 177             else:
 178                 raise ConfigParseError(f"unknown field in milestones entry "
 179                                        f"for {identifier!r}: `{k}`")
 180         if canonical_bug_id is None:
 181             raise ConfigParseError(f"`canonical_bug_id` field is missing in "
 182                                    f"milestones entry for {identifier!r}")
 183         return Milestone(config=self, identifier=identifier,
 184                          canonical_bug_id=canonical_bug_id)
 185
 186     def _parse_milestones(self, milestones: Any):
 187         if not isinstance(milestones, dict):
 188             raise ConfigParseError("`milestones` field must be a table")
 189         for identifier, value in milestones.items():
 190             assert isinstance(identifier, str)
 191             self.milestones[identifier] = \
 192                 self._parse_milestone(identifier, value)
 193
 194         # self.canonical_bug_ids checks for bug id clashes and raises
 195         # ConfigParseError if any are detected, so the following line
 196         # is needed:
 197         self.canonical_bug_ids
 198
 199     @staticmethod
 200     def _from_toml(parsed_toml: Dict[str, Any]) -> "Config":
 201         people = None
 202         bugzilla_url = None
 203         milestones = None
 204         for k, v in parsed_toml.items():
 205             assert isinstance(k, str)
 206             if k == "people":
 207                 people = v
 208             elif k == "milestones":
 209                 milestones = v
 210             elif k == "bugzilla_url":
 211                 if not isinstance(v, str):
 212                     raise ConfigParseError("`bugzilla_url` must be a string")
 213                 bugzilla_url = v
 214             else:
 215                 raise ConfigParseError(f"unknown config entry: `{k}`")
 216
 217         if bugzilla_url is None:
 218             raise ConfigParseError("`bugzilla_url` field is missing")
 219
 220         config = Config(bugzilla_url=bugzilla_url, people={}, milestones={})
 221
 222         if people is None:
 223             raise ConfigParseError("`people` table is missing")
 224         else:
 225             config._parse_people(people)
 226
 227         if milestones is None:
 228             raise ConfigParseError("`milestones` table is missing")
 229         else:
 230             config._parse_milestones(milestones)
 231
 232         return config
 233
 234     @staticmethod
 235     def from_str(text: str) -> "Config":
 236         try:
 237             parsed_toml = toml.loads(text)
 238         except toml.TomlDecodeError as e:
 239             new_err = ConfigParseError(f"TOML parse error: {e}")
 240             raise new_err.with_traceback(sys.exc_info()[2])
 241         return Config._from_toml(parsed_toml)
 242
 243     @staticmethod
 244     def from_file(file: Any) -> "Config":
 245         if isinstance(file, list):
 246             raise TypeError("list is not a valid file or path")
 247         try:
 248             parsed_toml = toml.load(file)
 249         except toml.TomlDecodeError as e:
 250             new_err = ConfigParseError(f"TOML parse error: {e}")
 251             raise new_err.with_traceback(sys.exc_info()[2])
 252         return Config._from_toml(parsed_toml)