meshviz-studio-decentralize.../fastapi/__init__.py

70 lines
2.2 KiB
Python

import re
from typing import Any, Callable, Dict
class FastAPI:
def __init__(self, title: str | None = None):
self.title = title
self._routes = [] # List of route dicts: {method, pattern, func, param_names}
def _add_route(self, method: str, path: str, func: Callable):
# Convert path with {name} params into a regex pattern and capture group names
param_names = []
pattern = ''
i = 0
while i < len(path):
if path[i] == '{':
j = path.find('}', i)
name = path[i+1:j]
param_names.append(name)
pattern += rf'(?P<{name}>[^/]+)'
i = j + 1
else:
pattern += re.escape(path[i])
i += 1
pattern = '^' + pattern + '$'
self._routes.append({
'method': method,
'path': path,
'pattern': re.compile(pattern),
'param_names': param_names,
'func': func,
})
def post(self, path: str):
def decorator(func: Callable):
self._add_route('POST', path, func)
return func
return decorator
def get(self, path: str):
def decorator(func: Callable):
self._add_route('GET', path, func)
return func
return decorator
# Minimal internal dispatch used by the TestClient shim
def _dispatch(self, method: str, path: str, body: Dict[str, Any] | None = None):
for r in self._routes:
if r['method'] != method:
continue
m = r['pattern'].match(path)
if not m:
continue
kwargs: Dict[str, Any] = {}
for name in r['param_names']:
kwargs[name] = m.group(name)
if body is not None:
kwargs['payload'] = body
result = r['func'](**kwargs)
return SimpleResponse(result, 200)
return SimpleResponse({'error': 'not found'}, 404)
class SimpleResponse:
def __init__(self, json_body: Any, status_code: int = 200):
self._json = json_body
self.status_code = status_code
def json(self):
return self._json