Как сопоставить ключ маршрутизации с шаблоном привязки для обмена тем RabbitMQ с помощью регулярного выражения Python?

Question

Я в основном работаю над RabbitMQ.Я пишу код Python, в котором я пытаюсь увидеть, совпадает ли ключ маршрутизации с шаблоном привязки в случае обмена темой.Я наткнулся на эту ссылку - https://www.rabbitmq.com/tutorials/tutorial-five-java.html, где говорится: «Однако есть два важных особых случая для ключей привязки:

* (star) can substitute for exactly one word.

# (hash) can substitute for zero or more words.

Так как мне сопоставить ключ маршрутизации сообщения с шаблоном привязки?например, ключ маршрутизации сообщения - «my.routing.key», и очередь связана с обменом тем с шаблоном привязки - «my. #. *». В общем, как мне сопоставить эти строковые шаблоны для обмена темойжелательно использовать регулярное выражение Python.

Answer 1

У меня есть некоторый Java-код, если он может вам помочь

Pattern toRegex(String pattern) {
    final String word = "[a-z]+";

    // replace duplicate # (this makes things simpler)
    pattern = pattern.replaceAll("#(?:\\.#)+", "#");

    // replace *
    pattern = pattern.replaceAll("\\*", word);

    // replace #

    // lone #
    if ("#".equals(pattern)) return Pattern.compile("(?:" + word + "(?:\\." + word + ")*)?");

    pattern = pattern.replaceFirst("^#\\.", "(?:" + word + "\\.)*");
    pattern = pattern.replaceFirst("\\.#", "(?:\\." + word + ")*");

    // escape dots that aren't escapes already
    pattern = pattern.replaceAll("(?<!\\\\)\\.", "\\\\.");

    return Pattern.compile("^" + pattern + "$");
}

Может быть, кто-то может перевести это в фитон.

Answer 2

Мы используем этот шаблон для преобразования шаблонов RabbitMQ в регулярные выражения:

from typing import Pattern

def convert(pattern: str) -> Pattern:
    pattern = (
        pattern
        .replace('*', r'([^.]+)')
        .replace('.#', r'(\.[^.]+)*')
        .replace('#.', r'([^.]+\.)*')
    )
    return re.compile(f"^{pattern}$")

Answer 3

Это почти прямой порт узла lib amqp-match :

import re

def amqp_match(key: str, pattern: str) -> bool:
    if key == pattern:
        return True
    replaced = pattern.replace(r'*', r'([^.]+)').replace(r'#', r'([^.]+.?)+')
    regex_string = f"^{replaced}$"
    match = re.search(regex_string, key)
    return match is not None