L o a d i n g
Асинхронный веб на Python: как выжать максимум из Flask с кастомным роутингом и деревьями решений Python

Когда речь заходит о разработке сайтов на Python, многие сразу думают про Flask или Django. Django — это "батарейки в комплекте", но если вы настоящий кодер, который хочет держать всё под контролем и писать логику с нуля, Flask — ваш выбор. Сегодня я расскажу, как сделать что-то нетривиальное: кастомный роутинг с использованием префиксного дерева (trie) и асинхронной обработки запросов для веб-приложения. Это не просто "Hello, World" — это уровень, где вы начинаете чувствовать себя архитектором.

Зачем это нужно?

Представьте: у вас есть веб-приложение с динамическими маршрутами, которые генерируются на лету. Например, система управления API с версиями вроде /api/v1/users, /api/v2/users, и вы хотите, чтобы роутинг был не просто списком правил, а умной структурой, которая масштабируется и работает быстро даже с тысячами эндпоинтов. Добавим сюда асинхронность, чтобы обрабатывать сотни запросов в секунду без потери производительности. Звучит как вызов? Тогда погнали.

Шаг 1: Префиксное дерево для роутинга

Flask по умолчанию использует простую систему маршрутов, где правила хранятся в списке и проверяются последовательно. Это работает для маленьких приложений, но если у вас сотни или тысячи маршрутов, поиск становится бутылочным горлышком. Давайте заменим это на trie.

Вот пример реализации:


class TrieNode:
    def __init__(self):
        self.children = {}
        self.is_end = False
        self.handler = None

class Router:
    def __init__(self):
        self.root = TrieNode()

    def add_route(self, path, handler):
        node = self.root
        parts = path.strip('/').split('/')
        for part in parts:
            if part not in node.children:
                node.children[part] = TrieNode()
            node = node.children[part]
        node.is_end = True
        node.handler = handler

    def match(self, path):
        node = self.root
        parts = path.strip('/').split('/')
        for part in parts:
            if part not in node.children:
                return None
            node = node.children[part]
        return node.handler if node.is_end else None

# Пример использования
router = Router()

def user_handler():
    return "Hello from users!"

router.add_route('/api/v1/users', user_handler)
handler = router.match('/api/v1/users')
print(handler())  # "Hello from users!"

Что тут происходит? Мы создали структуру данных trie, где каждый узел — это часть пути. Поиск маршрута теперь работает за O(m), где m — длина пути, а не O(n), как в линейном списке правил, где n — количество маршрутов. Это уже круто, но давайте добавим динамические параметры, типа /api/v1/users/<user_id>.

Шаг 2: Динамические маршруты

Для поддержки параметров вроде <user_id> модифицируем trie:


class TrieNode:
    def __init__(self):
        self.children = {}
        self.params = {}  # Для хранения параметров
        self.is_end = False
        self.handler = None

class Router:
    def __init__(self):
        self.root = TrieNode()

    def add_route(self, path, handler):
        node = self.root
        parts = path.strip('/').split('/')
        for part in parts:
            if part.startswith('<') and part.endswith('>'):
                param_name = part[1:-1]
                if '*' not in node.children:
                    node.children['*'] = TrieNode()
                node = node.children['*']
                node.params[param_name] = None
            else:
                if part not in node.children:
                    node.children[part] = TrieNode()
                node = node.children[part]
        node.is_end = True
        node.handler = handler

    def match(self, path):
        node = self.root
        parts = path.strip('/').split('/')
        params = {}
        for i, part in enumerate(parts):
            if part in node.children:
                node = node.children[part]
            elif '*' in node.children:
                node = node.children['*']
                param_name = list(node.params.keys())[0]  # Берем первый параметр
                params[param_name] = part
            else:
                return None, {}
        return (node.handler, params) if node.is_end else (None, {})

# Тестим
router = Router()

def user_by_id(user_id):
    return f"User ID: {user_id}"

router.add_route('/api/v1/users/<user_id>', user_by_id)
handler, params = router.match('/api/v1/users/123')
print(handler(**params))  # "User ID: 123"

Теперь роутер понимает динамические параметры и передает их в обработчик. Это уже похоже на что-то мощное.

Шаг 3: Асинхронность с Quart

Flask не поддерживает асинхронность из коробки, но есть Quart — его асинхронный аналог, совместимый с Flask API. Интегрируем наш роутер:


from quart import Quart, request

app = Quart(__name__)
router = Router()

async def user_by_id(user_id):
    await asyncio.sleep(1)  # Симуляция долгой операции
    return f"User ID: {user_id}"

router.add_route('/api/v1/users/<user_id>', user_by_id)

@app.route('/', defaults={'path': ''})
@app.route('/<path:path>')
async def catch_all(path):
    handler, params = router.match(path)
    if handler:
        return await handler(**params)
    return "Not Found", 404

if __name__ == '__main__':
    app.run()

Теперь каждый запрос обрабатывается асинхронно. Если у вас много I/O-операций (запросы к базе, внешние API), это спасает от блокировок и дает прирост производительности.

Шаг 4: Оптимизация и реальная жизнь

  • Кэширование маршрутов: Добавьте LRU-кэш для часто запрашиваемых путей с помощью functools.lru_cache или кастомной реализации.
  • Валидация параметров: Расширьте trie, чтобы поддерживать типы параметров, например <int:user_id>.
  • Обработка ошибок: Оберните вызовы handler в try-except и возвращайте красивые JSON-ошибки.

Пример с валидацией:


def add_route(self, path, handler):
    node = self.root
    parts = path.strip('/').split('/')
    for part in parts:
        if part.startswith('<') and part.endswith('>'):
            param_type, param_name = part[1:-1].split(':') if ':' in part[1:-1] else ('str', part[1:-1])
            if '*' not in node.children:
                node.children['*'] = TrieNode()
            node = node.children['*']
            node.params[param_name] = param_type
        else:
            if part not in node.children:
                node.children[part] = TrieNode()
            node = node.children[part]
    node.is_end = True
    node.handler = handler

def match(self, path):
    node = self.root
    parts = path.strip('/').split('/')
    params = {}
    for i, part in enumerate(parts):
        if part in node.children:
            node = node.children[part]
        elif '*' in node.children:
            node = node.children['*']
            param_name, param_type = list(node.params.items())[0]
            if param_type == 'int' and not part.isdigit():
                return None, {}
            params[param_name] = int(part) if param_type == 'int' else part
        else:
            return None, {}
    return (node.handler, params) if node.is_end else (None, {})

Теперь /api/v1/users/<int:user_id> принимает только числа. Круто, да?

Итог

Мы построили кастомный роутинг с trie, добавили поддержку параметров, сделали всё асинхронным и даже прикрутили базовую валидацию. Это не просто веб-разработка — это инженерия. Такой подход пригодится для высоконагруженных систем, где важны скорость и гибкость. Попробуйте развернуть это у себя, добавить middleware для логирования или авторизации — и вы поймете, почему Python в руках опытного программиста становится настоящим оружием.

Если хотите углубиться дальше, советую глянуть в сторону aiohttp или изучить, как trie можно оптимизировать сжатием путей (path compression). Удачи в кодинге!

Написать комментарий

Вы можете оставить комментарий автору статьи Обязательные поля помечены *