This document is for Kombu's development version, which can be significantly different from previous releases. Get the stable docs here: 5.0.
Source code for kombu.matcher
"""Pattern matching registry."""
from fnmatch import fnmatch
from re import match as rematch
from .utils.compat import entrypoints
from .utils.encoding import bytes_to_str
[docs]class MatcherRegistry:
"""Pattern matching function registry."""
MatcherNotInstalled = MatcherNotInstalled
matcher_pattern_first = ["pcre", ]
def __init__(self):
self._matchers = {}
self._default_matcher = None
[docs] def register(self, name, matcher):
"""Add matcher by name to the registry."""
self._matchers[name] = matcher
[docs] def unregister(self, name):
"""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):
"""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}'
)
[docs] def match(self, data, pattern, matcher=None, matcher_kwargs=None):
"""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 convience 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
[docs]def register_glob():
"""Register glob into default registry."""
registry.register('glob', fnmatch)
[docs]def register_pcre():
"""Register pcre into default registry."""
registry.register('pcre', 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)
