src/liquidcode/routing/group.py - develop-0.4 - liquid-g/liquid-code

src/liquidcode/routing/group.py
63 строки2 KB
User
0.4.14 - Debug-режим, ограничение middleware и WebSocket порт
06 июл 2026, 16:41
06 июл 2026, 16:41f3c973d
from typing import Callable, List, Optional

from .router import Router


class RouteGroup:
    MAX_GROUP_MIDDLEWARES = 50  # Ограничение для group middleware

    def __init__(self, router: Router, prefix: str = '', middlewares: Optional[List[Callable]] = None):
        self.router = router
        self.prefix = prefix.rstrip('/')
        if self.prefix and not self.prefix.startswith('/'):
            self.prefix = '/' + self.prefix
        self.middlewares = middlewares or []
        if len(self.middlewares) >= self.MAX_GROUP_MIDDLEWARES:
            raise RuntimeError(
                f"Достигнут предел group middleware ({self.MAX_GROUP_MIDDLEWARES}). "
                "Это может привести к RecursionError."
            )

    def route(self, path: str, methods: Optional[List[str]] = None):
        def decorator(func):
            full_path = self.prefix + path
            if not full_path.startswith('/'):
                full_path = '/' + full_path
            self.router.add_route(
                path=full_path,
                controller=func,
                methods=methods or ['GET'],
                group_middlewares=self.middlewares
            )
            return func

        return decorator

    def add_route(self, path: str, controller: Callable, methods: Optional[List[str]] = None):
        full_path = self.prefix + path
        if not full_path.startswith('/'):
            full_path = '/' + full_path
        self.router.add_route(
            path=full_path,
            controller=controller,
            methods=methods or ['GET'],
            group_middlewares=self.middlewares
        )

    def group(self, prefix: str, middlewares: Optional[List[Callable]] = None) -> 'RouteGroup':
        new_prefix = self.prefix + prefix
        combined_middlewares = self.middlewares + (middlewares or [])
        if len(combined_middlewares) >= self.MAX_GROUP_MIDDLEWARES:
            raise RuntimeError(
                f"Достигнут предел group middleware ({self.MAX_GROUP_MIDDLEWARES}). "
                "Это может привести к RecursionError."
            )
        return RouteGroup(self.router, new_prefix, combined_middlewares)

    def __enter__(self):
        """Поддержка контекстного менеджера."""
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Завершение контекстного менеджера."""
        pass