forked from aegis/pyserveX
cython path_matcher added to reduce time on hot operations
This commit is contained in:
@@ -0,0 +1,168 @@
|
||||
"""
|
||||
Pure Python fallback for _path_matcher when Cython is not available.
|
||||
|
||||
This module provides the same interface as the Cython _path_matcher module,
|
||||
allowing the application to run without compilation.
|
||||
"""
|
||||
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
|
||||
|
||||
class FastMountedPath:
|
||||
__slots__ = ("_path", "_path_with_slash", "_path_len", "_is_root", "name", "strip_path")
|
||||
|
||||
def __init__(self, path: str, name: str = "", strip_path: bool = True):
|
||||
if path.endswith("/") and len(path) > 1:
|
||||
path = path[:-1]
|
||||
|
||||
self._path = path
|
||||
self._path_len = len(path)
|
||||
self._is_root = path == "" or path == "/"
|
||||
self._path_with_slash = path + "/" if not self._is_root else "/"
|
||||
self.name = name or path
|
||||
self.strip_path = strip_path
|
||||
|
||||
@property
|
||||
def path(self) -> str:
|
||||
return self._path
|
||||
|
||||
def matches(self, request_path: str) -> bool:
|
||||
if self._is_root:
|
||||
return True
|
||||
|
||||
req_len = len(request_path)
|
||||
|
||||
if req_len < self._path_len:
|
||||
return False
|
||||
|
||||
if req_len == self._path_len:
|
||||
return request_path == self._path
|
||||
|
||||
if request_path[self._path_len] == "/":
|
||||
return request_path[: self._path_len] == self._path
|
||||
|
||||
return False
|
||||
|
||||
def get_modified_path(self, original_path: str) -> str:
|
||||
if not self.strip_path:
|
||||
return original_path
|
||||
|
||||
if self._is_root:
|
||||
return original_path
|
||||
|
||||
new_path = original_path[self._path_len :]
|
||||
|
||||
if not new_path:
|
||||
return "/"
|
||||
|
||||
return new_path
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"FastMountedPath(path={self._path!r}, name={self.name!r})"
|
||||
|
||||
|
||||
class FastMountManager:
|
||||
__slots__ = ("_mounts", "_mount_count")
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._mounts: List[FastMountedPath] = []
|
||||
self._mount_count: int = 0
|
||||
|
||||
def add_mount(self, mount: FastMountedPath) -> None:
|
||||
self._mounts.append(mount)
|
||||
self._mounts.sort(key=lambda m: len(m.path), reverse=True)
|
||||
self._mount_count = len(self._mounts)
|
||||
|
||||
def get_mount(self, request_path: str) -> Optional[FastMountedPath]:
|
||||
for mount in self._mounts:
|
||||
if mount.matches(request_path):
|
||||
return mount
|
||||
return None
|
||||
|
||||
def remove_mount(self, path: str) -> bool:
|
||||
if path.endswith("/") and len(path) > 1:
|
||||
path = path[:-1]
|
||||
|
||||
for i, mount in enumerate(self._mounts):
|
||||
if mount._path == path:
|
||||
del self._mounts[i]
|
||||
self._mount_count -= 1
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
@property
|
||||
def mounts(self) -> List[FastMountedPath]:
|
||||
return self._mounts.copy()
|
||||
|
||||
@property
|
||||
def mount_count(self) -> int:
|
||||
return self._mount_count
|
||||
|
||||
def list_mounts(self) -> List[Dict[str, Any]]:
|
||||
return [
|
||||
{
|
||||
"path": mount._path,
|
||||
"name": mount.name,
|
||||
"strip_path": mount.strip_path,
|
||||
}
|
||||
for mount in self._mounts
|
||||
]
|
||||
|
||||
|
||||
def path_matches_prefix(request_path: str, mount_path: str) -> bool:
|
||||
mount_len = len(mount_path)
|
||||
req_len = len(request_path)
|
||||
|
||||
if mount_len == 0 or mount_path == "/":
|
||||
return True
|
||||
|
||||
if req_len < mount_len:
|
||||
return False
|
||||
|
||||
if req_len == mount_len:
|
||||
return request_path == mount_path
|
||||
|
||||
if request_path[mount_len] == "/":
|
||||
return request_path[:mount_len] == mount_path
|
||||
|
||||
return False
|
||||
|
||||
|
||||
def strip_path_prefix(original_path: str, mount_path: str) -> str:
|
||||
mount_len = len(mount_path)
|
||||
|
||||
if mount_len == 0 or mount_path == "/":
|
||||
return original_path
|
||||
|
||||
result = original_path[mount_len:]
|
||||
|
||||
if not result:
|
||||
return "/"
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def match_and_modify_path(request_path: str, mount_path: str, strip_path: bool = True) -> Tuple[bool, Optional[str]]:
|
||||
mount_len = len(mount_path)
|
||||
req_len = len(request_path)
|
||||
is_root = mount_len == 0 or mount_path == "/"
|
||||
|
||||
if is_root:
|
||||
return (True, request_path)
|
||||
|
||||
if req_len < mount_len:
|
||||
return (False, None)
|
||||
|
||||
if req_len == mount_len:
|
||||
if request_path == mount_path:
|
||||
return (True, "/" if strip_path else request_path)
|
||||
return (False, None)
|
||||
|
||||
if request_path[mount_len] == "/" and request_path[:mount_len] == mount_path:
|
||||
if strip_path:
|
||||
modified = request_path[mount_len:]
|
||||
return (True, modified if modified else "/")
|
||||
return (True, request_path)
|
||||
|
||||
return (False, None)
|
||||
Reference in New Issue
Block a user