[routing] fix route description when there is multiple verbs

This commit is contained in:
Maxime Alves LIRMM 2020-09-29 11:12:45 +02:00
parent d93fb23bba
commit e223c0791c
3 changed files with 119 additions and 59 deletions

View File

@ -1,10 +1,11 @@
#!/usr/bin/env python3
import importlib
import typing as typ
from types import ModuleType
from typing import Generator, Dict, List
VERBS = ('GET', 'POST', 'PUT', 'PATCH', 'DELETE')
def get_fct_name(http_verb, path: str):
def get_fct_name(http_verb: str, path: str) -> str:
"""
Returns the predictable name of the function for a route
@ -48,15 +49,15 @@ def get_fct_name(http_verb, path: str):
return '_'.join(fct_name)
def gen_routes(route_params, path, m_router):
fqtn = route_params.get('FQTN')
def gen_routes(route_params: Dict, path: List, m_router: ModuleType) -> Generator:
d_res = {'fqtn': route_params.get('FQTN')}
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)')
if len(params) == 0:
print(f'No ACL for route [{verb}] "/".join(path)')
try:
fct_name = get_fct_name(verb, path[-1])
@ -65,38 +66,40 @@ def gen_routes(route_params, path, m_router):
print(f'{fct_name} is not defined in {m_router.__name__}')
continue
yield {
'verb':verb,
'path':f"/{'/'.join([ elt for elt in path if elt ])}",
'params':params,
'fct': fct,
'fqtn': fqtn }
d_res[verb] = {'fct': fct, 'params': params}
yield f"/{'/'.join([ elt for elt in path if elt ])}", d_res
def gen_router_routes(m_router, path=[]):
def gen_router_routes(m_router, path=None):
"""
[
('path', [acl], fct, fqtn)
]
{
'/truc/toto': {
}
}
"""
if not hasattr(m_router, 'ROUTES'):
print(f'Missing *ROUTES* constant in *{m_router.__name__}*')
if path is None:
path = []
routes = m_router.ROUTES
for subpath, route_params in routes.items():
path.append(subpath)
for route in gen_routes(route_params, path, m_router):
yield route
for r_path, d_route in gen_routes(route_params, path, m_router):
yield r_path, d_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 gen_router_routes(submod, path):
yield route_scan
for r_path, d_route in gen_router_routes(submod, path):
yield r_path, d_route
path.pop()

View File

@ -2,19 +2,14 @@
from functools import wraps
import importlib
import sys
from typing import Callable, List, Tuple, Dict
from typing import Callable, List, Tuple, Dict, Generator
from types import ModuleType
from halfapi.conf import (PROJECT_NAME, DB_NAME, HOST, PORT,
PRODUCTION, DOMAINS)
# from halfapi.db import (
# Domain,
# APIRouter,
# APIRoute,
# AclFunction,
# Acl)
from halfapi.lib.responses import *
from halfapi.lib.domain import gen_domain_routes
from halfapi.lib.domain import gen_domain_routes, VERBS
from starlette.exceptions import HTTPException
from starlette.routing import Mount, Route
from starlette.requests import Request
@ -22,7 +17,7 @@ from starlette.requests import Request
class DomainNotFoundError(Exception):
pass
def route_acl_decorator(fct: Callable, acls_mod, params: List[Dict]):
def route_acl_decorator(fct: Callable, params: List[Dict]):
"""
Decorator for async functions that calls pre-conditions functions
and appends kwargs to the target function
@ -31,8 +26,6 @@ def route_acl_decorator(fct: Callable, acls_mod, params: List[Dict]):
Parameters:
fct (Callable):
The function to decorate
acls_mod (Module):
The module that contains the pre-condition functions (acls)
params List[Dict]:
A list of dicts that have an "acl" key that points to a function
@ -68,40 +61,43 @@ def acl_mock(fct):
#
##
def gen_starlette_routes(m_dom):
def gen_starlette_routes(m_dom: ModuleType) -> Generator:
"""
Yields the Route objects for HalfAPI app
Parameters:
m_dom (module): the halfapi module
m_dom (ModuleType): the halfapi module
Returns:
Generator[Route]
Generator(Route)
"""
m_dom_acl = importlib.import_module('.acl', m_dom.__name__)
for route in gen_domain_routes(m_dom.__name__):
yield (
Route(route['path'],
route_acl_decorator(
route['fct'],
m_dom_acl,
route['params'],
),
methods=[route['verb']])
)
for path, d_route in gen_domain_routes(m_dom.__name__):
for verb in VERBS:
if verb not in d_route.keys():
continue
yield (
Route(path,
route_acl_decorator(
d_route[verb]['fct'],
d_route[verb]['params']
),
methods=[verb])
)
def api_routes(m_dom):
def api_routes(m_dom: ModuleType) -> Generator:
"""
Yields the description objects for HalfAPI app routes
Parameters:
m_dom (module): the halfapi module
m_dom (ModuleType): the halfapi module
Returns:
Generator[Dict]
Generator(Dict)
"""
m_dom_acl = importlib.import_module('.acl', m_dom.__name__)
@ -109,15 +105,26 @@ def api_routes(m_dom):
def pop_acl(r):
if 'acl' in r.keys():
r.pop('acl')
print(r)
return r
return {
route['path']: {
'params': list(map(pop_acl, route['params'])),
'verb': route['verb'],
'fqtn': route['fqtn']
}
for route in gen_domain_routes(m_dom.__name__)
}
def str_acl(params):
for param in params:
if 'acl' not in param.keys():
continue
param['acl'] = param['acl'].__name__
return params
d_res = {}
for path, d_route in gen_domain_routes(m_dom.__name__):
d_res[path] = {'fqtn': d_route['fqtn'] }
for verb in VERBS:
if verb not in d_route.keys():
continue
d_res[path][verb] = {
'params': str_acl(d_route[verb]['params']),
'fct': d_route[verb]['fct'].__name__
}
yield path, d_res

View File

@ -1,3 +1,5 @@
from types import ModuleType
from typing import Dict
from ..conf import DOMAINSDICT
from .routes import gen_starlette_routes, api_routes
from .responses import *
@ -7,18 +9,66 @@ schemas = SchemaGenerator(
{"openapi": "3.0.0", "info": {"title": "HalfAPI", "version": "1.0"}}
)
async def get_api_routes(request, *args, **kwargs):
"""
responses:
200:
description: Returns the current API routes description (HalfAPI 0.2.1)
as a JSON object
example:
{
"dummy_domain": {
"/abc/alphabet/organigramme": {
"fqtn": null,
"params": [
{}
],
"verb": "GET"
},
"/act/personne/": {
"fqtn": "acteur.personne",
"params": [
{}
"verb": "GET"
}
}
}
"""
#TODO: LADOC
return ORJSONResponse({
domain: api_routes(m_domain)
domain: { path: route for path, route in api_routes(m_domain) }
for domain, m_domain in DOMAINSDICT.items()
})
async def schema_json(request, *args, **kwargs):
"""
responses:
200:
description: Returns the current API routes description (OpenAPI v3)
as a JSON object
"""
return ORJSONResponse(
schemas.get_schema(routes=request.app.routes))
def schema_dict_dom(m_domain):
def schema_dict_dom(m_domain: ModuleType) -> Dict:
"""
Returns the API schema of the *m_domain* domain as a python dictionnary
Parameters:
m_domain (ModuleType): The module to scan for routes
Returns:
Dict: A dictionnary containing the description of the API using the
| OpenAPI standard
"""
routes = [
elt for elt in gen_starlette_routes(m_domain) ]
return schemas.get_schema(routes=routes)