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 __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]


[docs]class MatcherNotInstalled(Exception): """Matcher not installed/found."""
[docs]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
[docs] def register(self, name: str, matcher: MatcherFunction) -> None: """Add matcher by name to the registry.""" self._matchers[name] = matcher
[docs] 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}' )
[docs] 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
[docs]def register_glob() -> None: """Register glob into default registry.""" registry.register('glob', fnmatch)
[docs]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)