[wip][routing] add routing functions in /lib/domains

This commit is contained in:
Maxime Alves LIRMM 2020-09-22 12:57:36 +02:00
parent 7337246fc1
commit c54101c3e6
9 changed files with 181 additions and 0 deletions

117
halfapi/lib/domain.py Normal file
View File

@ -0,0 +1,117 @@
#!/usr/bin/env python3
import importlib
import typing as t
VERBS = ('GET', 'POST', 'PUT', 'PATCH', 'DELETE')
def get_fct_name(http_verb, path: t.List):
"""
Returns the predictable name of the function for a route
Parameters:
- http_verb (str): The Route's HTTP method (GET, POST, ...)
- path (str): A path beginning by '/' for the route
Returns:
str: The *unique* function name for a route and it's verb
Examples:
>>> get_fct_name('foo', 'bar')
Traceback (most recent call last):
...
Exception: Malformed path
>>> get_fct_name('get', '/')
'get_'
>>> get_fct_name('GET', '/')
'get_'
>>> get_fct_name('POST', '/foo')
'post_foo'
>>> get_fct_name('POST', '/foo/bar')
'post_foo_bar'
>>> get_fct_name('DEL', '/foo/{boo}/{far}/bar')
'del_foo_BOO_FAR_bar'
>>> get_fct_name('DEL', '/foo/{boo:zoo}')
'del_foo_BOO'
"""
fct_name = [http_verb.lower()]
for elt in path:
if elt and elt[0] == '{':
fct_name.append(elt[1:-1].split(':')[0].upper())
else:
fct_name.append(elt)
return '_'.join(fct_name)
def route_generator(route_params, path, m_router):
for verb in VERBS:
params = route_params.get(verb)
if params is None:
continue
if not len(params):
print(f'No ACL for route [{verb}] "/".join(path)')
try:
fct_name = get_fct_name(verb, [path[-1]])
fct = getattr(m_router, fct_name)
except AttributeError:
print(f'{fct_name} is not defined in {m_router.__name__}')
continue
yield {
'verb':verb,
'path':'/'.join([ elt for elt in path if elt ]),
'params':params,
'fct': fct }
def router_scanner(m_router, path=[]):
"""
[
('path', [acl], fct)
]
"""
if not hasattr(m_router, 'ROUTES'):
print(f'Missing *ROUTES* constant in *{domain}.routers*')
routes = m_router.ROUTES
pathlen = len(path)
for subpath, route_params in routes.items():
path.append(subpath)
for route in route_generator(route_params, path, m_router):
yield route
subroutes = route_params.get('SUBROUTES', [])
for subroute in subroutes:
path.append(subroute)
submod = importlib.import_module(f'.{subroute}', m_router.__name__)
for route_scan in router_scanner(submod, path):
yield route_scan
path.pop()
if pathlen < len(path):
path.pop()
def domain_scanner(domain):
m_domain = importlib.import_module(domain)
if not hasattr(m_domain, 'routers'):
raise Exception(f'No *routers* module in *{domain}*')
m_router = importlib.import_module('.routers', domain)
return router_scanner(m_router)

View File

@ -0,0 +1,5 @@
ROUTES={
'': {
'SUBROUTES': ['abc','act']
}
}

View File

@ -0,0 +1,5 @@
ROUTES={
'': {
'SUBROUTES': ['alphabet', 'pinnochio']
}
}

View File

@ -0,0 +1,12 @@
ROUTES={
'': {
'GET': [{'acl':None, 'in':None}]
},
'{test:uuid}': {
'GET': [{'acl':None, 'in':None}],
'POST': [{'acl':None, 'in':None}],
'PATCH': [{'acl':None, 'in':None}],
'PUT': [{'acl':None, 'in':None}]
}
}

View File

@ -0,0 +1,2 @@
ROUTES={
}

View File

@ -0,0 +1,5 @@
ROUTES={
'': {
'SUBROUTES': ['personne']
}
}

View File

@ -0,0 +1,13 @@
ROUTES={
'': {
'GET': [
{'acl':None, 'out':('id')}
],
},
'{user_id:uuid}': {
'GET': [
{'acl':None, 'out':('id')}
],
'SUBROUTES': ['eo']
}
}

View File

@ -0,0 +1,10 @@
from starlette.responses import Response
ROUTES={
'': {
'GET': [{'acl': 'None', 'in': ['ok']}]
}
}
async def get_(req):
return Response()

12
tests/test_lib_domain.py Normal file
View File

@ -0,0 +1,12 @@
#!/usr/bin/env python3
from halfapi.lib.domain import VERBS, router_scanner
def test_route_scanner():
from .dummy_domain import routers
for route in router_scanner(routers):
print(f'[{route["verb"]}] {route["path"]} {route["fct"]}')
assert route['verb'] in VERBS
assert isinstance(route['path'], str)
assert len(route['params']) > 0
assert hasattr(route['fct'], '__call__')