matcher.py 7.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185
  1. import re
  2. import typing as t
  3. from dataclasses import dataclass
  4. from dataclasses import field
  5. from .converters import ValidationError
  6. from .exceptions import NoMatch
  7. from .exceptions import RequestAliasRedirect
  8. from .exceptions import RequestPath
  9. from .rules import Rule
  10. from .rules import RulePart
  11. class SlashRequired(Exception):
  12. pass
  13. @dataclass
  14. class State:
  15. """A representation of a rule state.
  16. This includes the *rules* that correspond to the state and the
  17. possible *static* and *dynamic* transitions to the next state.
  18. """
  19. dynamic: t.List[t.Tuple[RulePart, "State"]] = field(default_factory=list)
  20. rules: t.List[Rule] = field(default_factory=list)
  21. static: t.Dict[str, "State"] = field(default_factory=dict)
  22. class StateMachineMatcher:
  23. def __init__(self, merge_slashes: bool) -> None:
  24. self._root = State()
  25. self.merge_slashes = merge_slashes
  26. def add(self, rule: Rule) -> None:
  27. state = self._root
  28. for part in rule._parts:
  29. if part.static:
  30. state.static.setdefault(part.content, State())
  31. state = state.static[part.content]
  32. else:
  33. for test_part, new_state in state.dynamic:
  34. if test_part == part:
  35. state = new_state
  36. break
  37. else:
  38. new_state = State()
  39. state.dynamic.append((part, new_state))
  40. state = new_state
  41. state.rules.append(rule)
  42. def update(self) -> None:
  43. # For every state the dynamic transitions should be sorted by
  44. # the weight of the transition
  45. state = self._root
  46. def _update_state(state: State) -> None:
  47. state.dynamic.sort(key=lambda entry: entry[0].weight)
  48. for new_state in state.static.values():
  49. _update_state(new_state)
  50. for _, new_state in state.dynamic:
  51. _update_state(new_state)
  52. _update_state(state)
  53. def match(
  54. self, domain: str, path: str, method: str, websocket: bool
  55. ) -> t.Tuple[Rule, t.MutableMapping[str, t.Any]]:
  56. # To match to a rule we need to start at the root state and
  57. # try to follow the transitions until we find a match, or find
  58. # there is no transition to follow.
  59. have_match_for = set()
  60. websocket_mismatch = False
  61. def _match(
  62. state: State, parts: t.List[str], values: t.List[str]
  63. ) -> t.Optional[t.Tuple[Rule, t.List[str]]]:
  64. # This function is meant to be called recursively, and will attempt
  65. # to match the head part to the state's transitions.
  66. nonlocal have_match_for, websocket_mismatch
  67. # The base case is when all parts have been matched via
  68. # transitions. Hence if there is a rule with methods &
  69. # websocket that work return it and the dynamic values
  70. # extracted.
  71. if parts == []:
  72. for rule in state.rules:
  73. if rule.methods is not None and method not in rule.methods:
  74. have_match_for.update(rule.methods)
  75. elif rule.websocket != websocket:
  76. websocket_mismatch = True
  77. else:
  78. return rule, values
  79. # Test if there is a match with this path with a
  80. # trailing slash, if so raise an exception to report
  81. # that matching is possible with an additional slash
  82. if "" in state.static:
  83. for rule in state.static[""].rules:
  84. if websocket == rule.websocket and (
  85. rule.methods is None or method in rule.methods
  86. ):
  87. if rule.strict_slashes:
  88. raise SlashRequired()
  89. else:
  90. return rule, values
  91. return None
  92. part = parts[0]
  93. # To match this part try the static transitions first
  94. if part in state.static:
  95. rv = _match(state.static[part], parts[1:], values)
  96. if rv is not None:
  97. return rv
  98. # No match via the static transitions, so try the dynamic
  99. # ones.
  100. for test_part, new_state in state.dynamic:
  101. target = part
  102. remaining = parts[1:]
  103. # A final part indicates a transition that always
  104. # consumes the remaining parts i.e. transitions to a
  105. # final state.
  106. if test_part.final:
  107. target = "/".join(parts)
  108. remaining = []
  109. match = re.compile(test_part.content).match(target)
  110. if match is not None:
  111. rv = _match(new_state, remaining, values + list(match.groups()))
  112. if rv is not None:
  113. return rv
  114. # If there is no match and the only part left is a
  115. # trailing slash ("") consider rules that aren't
  116. # strict-slashes as these should match if there is a final
  117. # slash part.
  118. if parts == [""]:
  119. for rule in state.rules:
  120. if rule.strict_slashes:
  121. continue
  122. if rule.methods is not None and method not in rule.methods:
  123. have_match_for.update(rule.methods)
  124. elif rule.websocket != websocket:
  125. websocket_mismatch = True
  126. else:
  127. return rule, values
  128. return None
  129. try:
  130. rv = _match(self._root, [domain, *path.split("/")], [])
  131. except SlashRequired:
  132. raise RequestPath(f"{path}/") from None
  133. if self.merge_slashes and rv is None:
  134. # Try to match again, but with slashes merged
  135. path = re.sub("/{2,}?", "/", path)
  136. try:
  137. rv = _match(self._root, [domain, *path.split("/")], [])
  138. except SlashRequired:
  139. raise RequestPath(f"{path}/") from None
  140. if rv is None:
  141. raise NoMatch(have_match_for, websocket_mismatch)
  142. else:
  143. raise RequestPath(f"{path}")
  144. elif rv is not None:
  145. rule, values = rv
  146. result = {}
  147. for name, value in zip(rule._converters.keys(), values):
  148. try:
  149. value = rule._converters[name].to_python(value)
  150. except ValidationError:
  151. raise NoMatch(have_match_for, websocket_mismatch) from None
  152. result[str(name)] = value
  153. if rule.defaults:
  154. result.update(rule.defaults)
  155. if rule.alias and rule.map.redirect_defaults:
  156. raise RequestAliasRedirect(result, rule.endpoint)
  157. return rule, result
  158. raise NoMatch(have_match_for, websocket_mismatch)