1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
|
"""Pattern matching registry."""
from __future__ import annotations
from fnmatch import fnmatch
from re import match as rematch
from typing import Callable, cast
from .utils.compat import entrypoints
from .utils.encoding import bytes_to_str
MatcherFunction = Callable[[str, str], bool]
class MatcherNotInstalled(Exception):
"""Matcher not installed/found."""
class MatcherRegistry:
"""Pattern matching function registry."""
MatcherNotInstalled = MatcherNotInstalled
matcher_pattern_first = ["pcre", ]
def __init__(self) -> None:
self._matchers: dict[str, MatcherFunction] = {}
self._default_matcher: MatcherFunction | None = None
def register(self, name: str, matcher: MatcherFunction) -> None:
"""Add matcher by name to the registry."""
self._matchers[name] = matcher
def unregister(self, name: str) -> None:
"""Remove matcher by name from the registry."""
try:
self._matchers.pop(name)
except KeyError:
raise self.MatcherNotInstalled(
f'No matcher installed for {name}'
)
def _set_default_matcher(self, name: str) -> None:
"""Set the default matching method.
:param name: The name of the registered matching method.
For example, `glob` (default), `pcre`, or any custom
methods registered using :meth:`register`.
:raises MatcherNotInstalled: If the matching method requested
is not available.
"""
try:
self._default_matcher = self._matchers[name]
except KeyError:
raise self.MatcherNotInstalled(
f'No matcher installed for {name}'
)
def match(
self,
data: bytes,
pattern: bytes,
matcher: str | None = None,
matcher_kwargs: dict[str, str] | None = None
) -> bool:
"""Call the matcher."""
if matcher and not self._matchers.get(matcher):
raise self.MatcherNotInstalled(
f'No matcher installed for {matcher}'
)
match_func = self._matchers[matcher or 'glob']
if matcher in self.matcher_pattern_first:
first_arg = bytes_to_str(pattern)
second_arg = bytes_to_str(data)
else:
first_arg = bytes_to_str(data)
second_arg = bytes_to_str(pattern)
return match_func(first_arg, second_arg, **matcher_kwargs or {})
#: Global registry of matchers.
registry = MatcherRegistry()
"""
.. function:: match(data, pattern, matcher=default_matcher,
matcher_kwargs=None):
Match `data` by `pattern` using `matcher`.
:param data: The data that should be matched. Must be string.
:param pattern: The pattern that should be applied. Must be string.
:keyword matcher: An optional string representing the mathcing
method (for example, `glob` or `pcre`).
If :const:`None` (default), then `glob` will be used.
:keyword matcher_kwargs: Additional keyword arguments that will be passed
to the specified `matcher`.
:returns: :const:`True` if `data` matches pattern,
:const:`False` otherwise.
:raises MatcherNotInstalled: If the matching method requested is not
available.
"""
match = registry.match
"""
.. function:: register(name, matcher):
Register a new matching method.
:param name: A convenient name for the mathing method.
:param matcher: A method that will be passed data and pattern.
"""
register = registry.register
"""
.. function:: unregister(name):
Unregister registered matching method.
:param name: Registered matching method name.
"""
unregister = registry.unregister
def register_glob() -> None:
"""Register glob into default registry."""
registry.register('glob', fnmatch)
def register_pcre() -> None:
"""Register pcre into default registry."""
registry.register('pcre', cast(MatcherFunction, rematch))
# Register the base matching methods.
register_glob()
register_pcre()
# Default matching method is 'glob'
registry._set_default_matcher('glob')
# Load entrypoints from installed extensions
for ep, args in entrypoints('kombu.matchers'):
register(ep.name, *args)
|
