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

src/liquidcode/routing/router.py
129 строк5 KB
User
0.4.7 - Рефакторинг routing.py в пакет
05 июл 2026, 17:09
05 июл 2026, 17:0997bff95
import inspect
from typing import Any, Callable, Dict, List, Optional, Tuple, Type, Union

# from .group import RouteGroup - отложен для избежания circular import
from .route_class import Route


class Router:
    def __init__(self) -> None:
        self._routes: List[Route] = []

    def add_route(
            self,
            path: str,
            controller: Union[Callable, Tuple[Type, str]],
            methods: List[str],
            group_middlewares: Optional[List[Callable]] = None,
    ) -> None:
        if not callable(controller) and not (
                isinstance(controller, tuple)
                and len(controller) == 2
                and isinstance(controller[0], type)
                and isinstance(controller[1], str)
        ):
            raise TypeError("controller must be callable or tuple (class, method_name)")

        if not methods:
            methods = ['GET']
        methods = [m.upper() for m in methods]
        if not all(isinstance(m, str) for m in methods):
            raise ValueError("methods must be a list of strings")

        for route in self._routes:
            if route.path == path and set(route.methods) & set(methods):
                raise ValueError(f"Route conflict: path '{path}' with methods {methods} already exists")

        route = Route(path, methods, controller, group_middlewares)
        self._routes.append(route)

        # Автоматически добавляем варианты со слешем и без
        if path != '/' and not path.endswith('/'):
            path_with_slash = path + '/'
            conflict = any(r.path == path_with_slash and set(r.methods) & set(methods) for r in self._routes)
            if not conflict:
                self._routes.append(Route(path_with_slash, methods, controller, group_middlewares))
        elif path != '/' and path.endswith('/'):
            path_without_slash = path.rstrip('/')
            conflict = any(r.path == path_without_slash and set(r.methods) & set(methods) for r in self._routes)
            if not conflict:
                self._routes.append(Route(path_without_slash, methods, controller, group_middlewares))

    def add_routes_from_controller(self, controller_class: Type) -> None:
        prefix = getattr(controller_class, '_controller_prefix', '')
        class_middlewares = getattr(controller_class, '_controller_middlewares', '')

        for _, method in inspect.getmembers(controller_class, predicate=inspect.isfunction):
            if hasattr(method, '_route_path') and hasattr(method, '_route_methods'):
                path = method._route_path
                if prefix:
                    if path.startswith('/'):
                        path = path[1:]
                    full_path = prefix + '/' + path if path else prefix
                    if not full_path.startswith('/'):
                        full_path = '/' + full_path
                else:
                    full_path = path
                methods = method._route_methods
                if not isinstance(methods, list):
                    methods = list(methods) if methods else ['GET']
                self.add_route(
                    path=full_path,
                    controller=(controller_class, method.__name__),
                    methods=methods,
                    group_middlewares=class_middlewares
                )

    def match(
            self, path: str, method: str
    ) -> Optional[Tuple[Union[Callable, Tuple[Type, str]], Dict[str, str], List[Callable]]]:
        method = method.upper()
        for route in self._routes:
            if method not in route.methods:
                continue
            params = route.match(path)
            if params is not None:
                return route.controller, params, route.group_middlewares
        return None

    def get_allowed_methods(self, path: str) -> List[str]:
        allowed = set()
        for route in self._routes:
            if route.match(path) is not None:
                allowed.update(route.methods)
        return sorted(allowed)

    def find_route(
            self, path: str, method: str
    ) -> Optional[Tuple[Union[Callable, Tuple[Type, str]], Dict[str, str]]]:
        result = self.match(path, method)
        if result:
            controller, params, _ = result
            return controller, params
        return None

    def has_route(self, path: str, method: str) -> bool:
        method = method.upper()
        for route in self._routes:
            if route.path == path and method in route.methods:
                return True
        return False

    def all_routes(self) -> List[Dict[str, Any]]:
        result = []
        for route in self._routes:
            controller_info = (
                route.controller.__qualname__
                if callable(route.controller)
                else f"{route.controller[0].__name__}.{route.controller[1]}"
            )
            result.append({
                'path': route.path,
                'methods': route.methods,
                'controller': controller_info,
            })
        return result

    def group(self, prefix: str, middlewares: Optional[List[Callable]] = None):
        from .group import RouteGroup
        return RouteGroup(self, prefix, middlewares)