src/budget_sync/config.py

   1 from budget_sync.ordered_set import OrderedSet
   2 from budget_sync.util import PrettyPrinter
   3 import toml
   4 import sys
   5 from typing import Mapping, Set, Dict, Any, Optional
   6 try:
   7     from functools import cached_property
   8 except ImportError:
   9     # compatability with python < 3.8
  10     from cached_property import cached_property
  11
  12
  13 class ConfigParseError(Exception):
  14     pass
  15
  16
  17 class Person:
  18     aliases: OrderedSet[str]
  19     email: Optional[str]
  20
  21     def __init__(self, config: "Config", identifier: str,
  22                  full_name: str,
  23                  aliases: Optional[OrderedSet[str]] = None,
  24                  email: Optional[str] = None):
  25         self.config = config
  26         self.identifier = identifier
  27         if aliases is None:
  28             aliases = OrderedSet()
  29         else:
  30             assert isinstance(aliases, OrderedSet)
  31         self.aliases = aliases
  32         self.email = email
  33         self.full_name = full_name
  34
  35     @cached_property
  36     def all_names(self) -> OrderedSet[str]:
  37         retval = OrderedSet(self.aliases)
  38         retval.add(self.identifier)
  39         retval.add(self.full_name)
  40         if self.email is not None:
  41             retval.add(self.email)
  42         return retval
  43
  44     @cached_property
  45     def output_markdown_file(self) -> str:
  46         return self.identifier + '.mdwn'
  47
  48     def __eq__(self, other):
  49         return self.identifier == other.identifier
  50
  51     def __hash__(self):
  52         return hash(self.identifier)
  53
  54     def __pretty_print__(self, pp: PrettyPrinter):
  55         with pp.type_pp("Person") as tpp:
  56             tpp.field("config", ...)
  57             tpp.field("identifier", self.identifier)
  58
  59     def __repr__(self):
  60         return (f"Person(config=..., identifier={self.identifier!r}, "
  61                 f"full_name={self.full_name!r}, "
  62                 f"aliases={self.aliases!r}, email={self.email!r})")
  63
  64
  65 class Milestone:
  66     def __init__(self, config: "Config",
  67                  identifier: str, canonical_bug_id: int):
  68         self.config = config
  69         self.identifier = identifier
  70         self.canonical_bug_id = canonical_bug_id
  71
  72     def __eq__(self, other):
  73         return self.identifier == other.identifier
  74
  75     def __hash__(self):
  76         return hash(self.identifier)
  77
  78     def __repr__(self):
  79         return f"Milestone(config=..., " \
  80             f"identifier={self.identifier!r}, " \
  81             f"canonical_bug_id={self.canonical_bug_id})"
  82
  83
  84 class Config:
  85     def __init__(self, bugzilla_url: str, people: Dict[str, Person],
  86                  milestones: Dict[str, Milestone]):
  87         self.bugzilla_url = bugzilla_url
  88         self.people = people
  89         self.milestones = milestones
  90
  91     def __repr__(self):
  92         return f"Config(bugzilla_url={self.bugzilla_url!r}, " \
  93             f"people={self.people!r}, " \
  94             f"milestones={self.milestones!r})"
  95
  96     @cached_property
  97     def bugzilla_url_stripped(self):
  98         return self.bugzilla_url.rstrip('/')
  99
 100     @cached_property
 101     def all_names(self) -> Dict[str, Person]:
 102         # also checks for any name clashes and raises
 103         # ConfigParseError if any are detected
 104         retval = self.people.copy()
 105
 106         for person in self.people.values():
 107             for name in person.all_names:
 108                 other_person = retval.get(name)
 109                 if other_person is not None and other_person is not person:
 110                     alias_or_email_or_full_name = "alias"
 111                     if name == person.email:
 112                         alias_or_email_or_full_name = "email"
 113                     if name == person.full_name:
 114                         alias_or_email_or_full_name = "full_name"
 115                     if name in self.people:
 116                         raise ConfigParseError(
 117                             f"{alias_or_email_or_full_name} is not allowed "
 118                             f"to be the same "
 119                             f"as any person's identifier: in person entry for "
 120                             f"{person.identifier!r}: {name!r} is also the "
 121                             f"identifier for person"
 122                             f" {other_person.identifier!r}")
 123                     raise ConfigParseError(
 124                         f"{alias_or_email_or_full_name} is not allowed "
 125                         f"to be the same as "
 126                         f"another person's alias, email, or full_name: "
 127                         f"in person entry "
 128                         f"for {person.identifier!r}: {name!r} is also an alias"
 129                         f", email, or full_name for person "
 130                         f"{other_person.identifier!r}")
 131                 retval[name] = person
 132         return retval
 133
 134     @cached_property
 135     def canonical_bug_ids(self) -> Dict[int, Milestone]:
 136         # also checks for any bug id clashes and raises
 137         # ConfigParseError if any are detected
 138         retval: Dict[int, Milestone] = {}
 139         for milestone in self.milestones.values():
 140             other_milestone = retval.get(milestone.canonical_bug_id)
 141             if other_milestone is not None:
 142                 raise ConfigParseError(
 143                     f"canonical_bug_id is not allowed to be the same as "
 144                     f"another milestone's canonical_bug_id: in milestone "
 145                     f"entry for {milestone.identifier!r}: "
 146                     f"{milestone.canonical_bug_id} is also the "
 147                     f"canonical_bug_id for milestone "
 148                     f"{other_milestone.identifier!r}")
 149             retval[milestone.canonical_bug_id] = milestone
 150         return retval
 151
 152     def _parse_person(self, identifier: str, value: Any) -> Person:
 153         def raise_aliases_must_be_list_of_strings():
 154             raise ConfigParseError(
 155                 f"`aliases` field in person entry for {identifier!r} must "
 156                 f"be a list of strings")
 157
 158         if not isinstance(value, dict):
 159             raise ConfigParseError(
 160                 f"person entry for {identifier!r} must be a table")
 161         aliases = OrderedSet()
 162         email = None
 163         full_name = None
 164         for k, v in value.items():
 165             assert isinstance(k, str)
 166             if k == "aliases":
 167                 if not isinstance(v, list):
 168                     raise_aliases_must_be_list_of_strings()
 169                 for alias in v:
 170                     if isinstance(alias, str):
 171                         if alias in aliases:
 172                             raise ConfigParseError(
 173                                 f"duplicate alias in person entry for "
 174                                 f"{identifier!r}: {alias!r}")
 175                         aliases.add(alias)
 176                     else:
 177                         raise_aliases_must_be_list_of_strings()
 178             elif k == "email":
 179                 if not isinstance(v, str):
 180                     raise ConfigParseError(
 181                         f"`email` field in person entry for {identifier!r} "
 182                         f"must be a string")
 183                 email = v
 184             elif k == "full_name":
 185                 if not isinstance(v, str):
 186                     raise ConfigParseError(
 187                         f"`full_name` field in person entry for "
 188                         f"{identifier!r} must be a string")
 189                 full_name = v
 190             else:
 191                 raise ConfigParseError(
 192                     f"unknown field in person entry for {identifier!r}: `{k}`")
 193         if full_name is None:
 194             raise ConfigParseError(f"`full_name` field is missing in "
 195                                    f"person entry for {identifier!r}")
 196         return Person(config=self, identifier=identifier,
 197                       full_name=full_name,
 198                       aliases=aliases, email=email)
 199
 200     def _parse_people(self, people: Any):
 201         if not isinstance(people, dict):
 202             raise ConfigParseError("`people` field must be a table")
 203         for identifier, value in people.items():
 204             assert isinstance(identifier, str)
 205             self.people[identifier] = self._parse_person(identifier, value)
 206
 207         # self.all_names checks for name clashes and raises ConfigParseError
 208         # if any are detected, so the following line is needed:
 209         self.all_names
 210
 211     def _parse_milestone(self, identifier: str, value: Any) -> Milestone:
 212         if not isinstance(value, dict):
 213             raise ConfigParseError(
 214                 f"milestones entry for {identifier!r} must be a table")
 215         canonical_bug_id = None
 216         for k, v in value.items():
 217             assert isinstance(k, str)
 218             if k == "canonical_bug_id":
 219                 if not isinstance(v, int):
 220                     raise ConfigParseError(
 221                         f"`canonical_bug_id` field in milestones entry for "
 222                         f"{identifier!r} must be an integer")
 223                 canonical_bug_id = v
 224             else:
 225                 raise ConfigParseError(f"unknown field in milestones entry "
 226                                        f"for {identifier!r}: `{k}`")
 227         if canonical_bug_id is None:
 228             raise ConfigParseError(f"`canonical_bug_id` field is missing in "
 229                                    f"milestones entry for {identifier!r}")
 230         return Milestone(config=self, identifier=identifier,
 231                          canonical_bug_id=canonical_bug_id)
 232
 233     def _parse_milestones(self, milestones: Any):
 234         if not isinstance(milestones, dict):
 235             raise ConfigParseError("`milestones` field must be a table")
 236         for identifier, value in milestones.items():
 237             assert isinstance(identifier, str)
 238             self.milestones[identifier] = \
 239                 self._parse_milestone(identifier, value)
 240
 241         # self.canonical_bug_ids checks for bug id clashes and raises
 242         # ConfigParseError if any are detected, so the following line
 243         # is needed:
 244         self.canonical_bug_ids
 245
 246     @staticmethod
 247     def _from_toml(parsed_toml: Mapping[str, Any]) -> "Config":
 248         people = None
 249         bugzilla_url = None
 250         milestones = None
 251         for k, v in parsed_toml.items():
 252             assert isinstance(k, str)
 253             if k == "people":
 254                 people = v
 255             elif k == "milestones":
 256                 milestones = v
 257             elif k == "bugzilla_url":
 258                 if not isinstance(v, str):
 259                     raise ConfigParseError("`bugzilla_url` must be a string")
 260                 bugzilla_url = v
 261             else:
 262                 raise ConfigParseError(f"unknown config entry: `{k}`")
 263
 264         if bugzilla_url is None:
 265             raise ConfigParseError("`bugzilla_url` field is missing")
 266
 267         config = Config(bugzilla_url=bugzilla_url, people={}, milestones={})
 268
 269         if people is None:
 270             raise ConfigParseError("`people` table is missing")
 271         else:
 272             config._parse_people(people)
 273
 274         if milestones is None:
 275             raise ConfigParseError("`milestones` table is missing")
 276         else:
 277             config._parse_milestones(milestones)
 278
 279         return config
 280
 281     @staticmethod
 282     def from_str(text: str) -> "Config":
 283         try:
 284             parsed_toml = toml.loads(text)
 285         except toml.TomlDecodeError as e:
 286             new_err = ConfigParseError(f"TOML parse error: {e}")
 287             raise new_err.with_traceback(sys.exc_info()[2])
 288         return Config._from_toml(parsed_toml)
 289
 290     @staticmethod
 291     def from_file(file: Any) -> "Config":
 292         if isinstance(file, list):
 293             raise TypeError("list is not a valid file or path")
 294         try:
 295             parsed_toml = toml.load(file)
 296         except toml.TomlDecodeError as e:
 297             new_err = ConfigParseError(f"TOML parse error: {e}")
 298             raise new_err.with_traceback(sys.exc_info()[2])
 299         return Config._from_toml(parsed_toml)