Initial upload: API Mock CLI v0.1.0
Some checks failed
CI / test (3.10) (push) Has been cancelled
CI / test (3.11) (push) Has been cancelled
CI / test (3.12) (push) Has been cancelled
CI / test (3.9) (push) Has been cancelled
CI / lint (push) Has been cancelled
CI / type-check (push) Has been cancelled
CI / build (push) Has been cancelled
Some checks failed
CI / test (3.10) (push) Has been cancelled
CI / test (3.11) (push) Has been cancelled
CI / test (3.12) (push) Has been cancelled
CI / test (3.9) (push) Has been cancelled
CI / lint (push) Has been cancelled
CI / type-check (push) Has been cancelled
CI / build (push) Has been cancelled
This commit is contained in:
64
src/core/matcher.py
Normal file
64
src/core/matcher.py
Normal file
@@ -0,0 +1,64 @@
|
|||||||
|
import re
|
||||||
|
from typing import Optional, Dict, Any, List
|
||||||
|
from src.models.endpoint import Endpoint
|
||||||
|
|
||||||
|
|
||||||
|
class Matcher:
|
||||||
|
HTTP_METHODS = ["GET", "POST", "PUT", "DELETE", "PATCH", "HEAD", "OPTIONS"]
|
||||||
|
|
||||||
|
def __init__(self):
|
||||||
|
self.patterns: Dict[str, List[Dict[str, Any]]] = {
|
||||||
|
method: [] for method in self.HTTP_METHODS
|
||||||
|
}
|
||||||
|
|
||||||
|
def compile_pattern(self, path: str) -> re.Pattern:
|
||||||
|
param_names = {}
|
||||||
|
regex_path = re.sub(r"\{(\w+)\}", lambda m: param_names.__setitem__(m.group(1), m.group(0)) or "(?P<" + m.group(1) + ">[^/]+)", path)
|
||||||
|
regex_path = f"^{regex_path}$"
|
||||||
|
return re.compile(regex_path)
|
||||||
|
|
||||||
|
def add_endpoint(self, endpoint: Endpoint) -> None:
|
||||||
|
method = endpoint.method.upper()
|
||||||
|
if method not in self.HTTP_METHODS:
|
||||||
|
raise ValueError(f"Invalid HTTP method: {method}")
|
||||||
|
pattern = self.compile_pattern(endpoint.path)
|
||||||
|
self.patterns[method].append({
|
||||||
|
"pattern": pattern,
|
||||||
|
"endpoint": endpoint,
|
||||||
|
"params": {}
|
||||||
|
})
|
||||||
|
|
||||||
|
def match(self, path: str, method: str) -> Optional[tuple[Endpoint, Dict[str, str]]]:
|
||||||
|
method = method.upper()
|
||||||
|
if method not in self.HTTP_METHODS:
|
||||||
|
return None
|
||||||
|
for route in self.patterns[method]:
|
||||||
|
match = route["pattern"].match(path)
|
||||||
|
if match:
|
||||||
|
return route["endpoint"], match.groupdict()
|
||||||
|
return None
|
||||||
|
|
||||||
|
def get_endpoints(self, method: Optional[str] = None) -> List[Endpoint]:
|
||||||
|
if method:
|
||||||
|
method = method.upper()
|
||||||
|
return [r["endpoint"] for r in self.patterns.get(method, [])]
|
||||||
|
all_endpoints = []
|
||||||
|
for routes in self.patterns.values():
|
||||||
|
all_endpoints.extend(r["endpoint"] for r in routes)
|
||||||
|
return all_endpoints
|
||||||
|
|
||||||
|
def remove_endpoint(self, path: str, method: str) -> bool:
|
||||||
|
method = method.upper()
|
||||||
|
original_count = len(self.patterns[method])
|
||||||
|
self.patterns[method] = [
|
||||||
|
r for r in self.patterns[method]
|
||||||
|
if r["endpoint"].path != path or r["endpoint"].method.upper() != method
|
||||||
|
]
|
||||||
|
return len(self.patterns[method]) < original_count
|
||||||
|
|
||||||
|
def clear(self, method: Optional[str] = None) -> None:
|
||||||
|
if method:
|
||||||
|
self.patterns[method.upper()] = []
|
||||||
|
else:
|
||||||
|
for m in self.HTTP_METHODS:
|
||||||
|
self.patterns[m] = []
|
||||||
Reference in New Issue
Block a user