codekingpro/portable-devtools

1from __future__ import annotations2 3import re4import typing as t5from dataclasses import dataclass6from dataclasses import field7 8from .converters import ValidationError9from .exceptions import NoMatch10from .exceptions import RequestAliasRedirect11from .exceptions import RequestPath12from .rules import Rule13from .rules import RulePart14 15 16class SlashRequired(Exception):17    pass18 19 20@dataclass21class State:22    """A representation of a rule state.23 24    This includes the *rules* that correspond to the state and the25    possible *static* and *dynamic* transitions to the next state.26    """27 28    dynamic: list[tuple[RulePart, State]] = field(default_factory=list)29    rules: list[Rule] = field(default_factory=list)30    static: dict[str, State] = field(default_factory=dict)31 32 33class StateMachineMatcher:34    def __init__(self, merge_slashes: bool) -> None:35        self._root = State()36        self.merge_slashes = merge_slashes37 38    def add(self, rule: Rule) -> None:39        state = self._root40        for part in rule._parts:41            if part.static:42                state.static.setdefault(part.content, State())43                state = state.static[part.content]44            else:45                for test_part, new_state in state.dynamic:46                    if test_part == part:47                        state = new_state48                        break49                else:50                    new_state = State()51                    state.dynamic.append((part, new_state))52                    state = new_state53        state.rules.append(rule)54 55    def update(self) -> None:56        # For every state the dynamic transitions should be sorted by57        # the weight of the transition58        state = self._root59 60        def _update_state(state: State) -> None:61            state.dynamic.sort(key=lambda entry: entry[0].weight)62            for new_state in state.static.values():63                _update_state(new_state)64            for _, new_state in state.dynamic:65                _update_state(new_state)66 67        _update_state(state)68 69    def match(70        self, domain: str, path: str, method: str, websocket: bool71    ) -> tuple[Rule, t.MutableMapping[str, t.Any]]:72        # To match to a rule we need to start at the root state and73        # try to follow the transitions until we find a match, or find74        # there is no transition to follow.75 76        have_match_for = set()77        websocket_mismatch = False78 79        def _match(80            state: State, parts: list[str], values: list[str]81        ) -> tuple[Rule, list[str]] | None:82            # This function is meant to be called recursively, and will attempt83            # to match the head part to the state's transitions.84            nonlocal have_match_for, websocket_mismatch85 86            # The base case is when all parts have been matched via87            # transitions. Hence if there is a rule with methods &88            # websocket that work return it and the dynamic values89            # extracted.90            if parts == []:91                for rule in state.rules:92                    if rule.methods is not None and method not in rule.methods:93                        have_match_for.update(rule.methods)94                    elif rule.websocket != websocket:95                        websocket_mismatch = True96                    else:97                        return rule, values98 99                # Test if there is a match with this path with a100                # trailing slash, if so raise an exception to report101                # that matching is possible with an additional slash102                if "" in state.static:103                    for rule in state.static[""].rules:104                        if websocket == rule.websocket and (105                            rule.methods is None or method in rule.methods106                        ):107                            if rule.strict_slashes:108                                raise SlashRequired()109                            else:110                                return rule, values111                return None112 113            part = parts[0]114            # To match this part try the static transitions first115            if part in state.static:116                rv = _match(state.static[part], parts[1:], values)117                if rv is not None:118                    return rv119            # No match via the static transitions, so try the dynamic120            # ones.121            for test_part, new_state in state.dynamic:122                target = part123                remaining = parts[1:]124                # A final part indicates a transition that always125                # consumes the remaining parts i.e. transitions to a126                # final state.127                if test_part.final:128                    target = "/".join(parts)129                    remaining = []130                match = re.compile(test_part.content).match(target)131                if match is not None:132                    if test_part.suffixed:133                        # If a part_isolating=False part has a slash suffix, remove the134                        # suffix from the match and check for the slash redirect next.135                        suffix = match.groups()[-1]136                        if suffix == "/":137                            remaining = [""]138 139                    converter_groups = sorted(140                        match.groupdict().items(), key=lambda entry: entry[0]141                    )142                    groups = [143                        value144                        for key, value in converter_groups145                        if key[:11] == "__werkzeug_"146                    ]147                    rv = _match(new_state, remaining, values + groups)148                    if rv is not None:149                        return rv150 151            # If there is no match and the only part left is a152            # trailing slash ("") consider rules that aren't153            # strict-slashes as these should match if there is a final154            # slash part.155            if parts == [""]:156                for rule in state.rules:157                    if rule.strict_slashes:158                        continue159                    if rule.methods is not None and method not in rule.methods:160                        have_match_for.update(rule.methods)161                    elif rule.websocket != websocket:162                        websocket_mismatch = True163                    else:164                        return rule, values165 166            return None167 168        try:169            rv = _match(self._root, [domain, *path.split("/")], [])170        except SlashRequired:171            raise RequestPath(f"{path}/") from None172 173        if self.merge_slashes and rv is None:174            # Try to match again, but with slashes merged175            path = re.sub("/{2,}", "/", path)176            try:177                rv = _match(self._root, [domain, *path.split("/")], [])178            except SlashRequired:179                raise RequestPath(f"{path}/") from None180            if rv is None or rv[0].merge_slashes is False:181                raise NoMatch(have_match_for, websocket_mismatch)182            else:183                raise RequestPath(f"{path}")184        elif rv is not None:185            rule, values = rv186 187            result = {}188            for name, value in zip(rule._converters.keys(), values):189                try:190                    value = rule._converters[name].to_python(value)191                except ValidationError:192                    raise NoMatch(have_match_for, websocket_mismatch) from None193                result[str(name)] = value194            if rule.defaults:195                result.update(rule.defaults)196 197            if rule.alias and rule.map.redirect_defaults:198                raise RequestAliasRedirect(result, rule.endpoint)199 200            return rule, result201 202        raise NoMatch(have_match_for, websocket_mismatch)203