[routing] handle fqtn parameter in domain schema + get_api_routes (/) + /user -> /halfapi/current_user
This commit is contained in:
parent
f0d980e035
commit
3530f53820
|
@ -17,7 +17,7 @@ from halfapi.lib.jwt_middleware import JWTAuthenticationBackend
|
|||
|
||||
from halfapi.lib.responses import *
|
||||
from halfapi.lib.routes import gen_starlette_routes
|
||||
from halfapi.lib.schemas import schema_json
|
||||
from halfapi.lib.schemas import get_api_routes, schema_json
|
||||
|
||||
|
||||
"""
|
||||
|
@ -25,19 +25,17 @@ Base routes definition
|
|||
|
||||
Only debug or doc routes, that should not be available in production
|
||||
"""
|
||||
routes = [
|
||||
Route('/', lambda request, *args, **kwargs: ORJSONResponse('It Works!')),
|
||||
routes = [ Route('/', get_api_routes) ]
|
||||
|
||||
Route('/user', lambda request, *args, **kwargs:
|
||||
ORJSONResponse({'user':request.user.json})
|
||||
if type(request.user) != UnauthenticatedUser
|
||||
else ORJSONResponse({'user':False})),
|
||||
|
||||
Route('/payload', lambda request, *args, **kwargs:
|
||||
ORJSONResponse({'payload':str(request.payload)})),
|
||||
|
||||
Route('/schema', schema_json)
|
||||
] if not PRODUCTION else []
|
||||
if not PRODUCTION:
|
||||
routes += [
|
||||
Route('/halfapi/current_user', lambda request, *args, **kwargs:
|
||||
ORJSONResponse({'user':request.user.json})
|
||||
if type(request.user) != UnauthenticatedUser
|
||||
else ORJSONResponse({'user': None})),
|
||||
Route('/halfapi/schema', schema_json)
|
||||
]
|
||||
|
||||
for domain, m_domain in DOMAINSDICT.items():
|
||||
for route in gen_starlette_routes(m_domain):
|
||||
|
|
|
@ -49,6 +49,8 @@ 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')
|
||||
|
||||
for verb in VERBS:
|
||||
params = route_params.get(verb)
|
||||
if params is None:
|
||||
|
@ -67,13 +69,14 @@ def gen_routes(route_params, path, m_router):
|
|||
'verb':verb,
|
||||
'path':f"/{'/'.join([ elt for elt in path if elt ])}",
|
||||
'params':params,
|
||||
'fct': fct }
|
||||
'fct': fct,
|
||||
'fqtn': fqtn }
|
||||
|
||||
|
||||
def gen_router_routes(m_router, path=[]):
|
||||
"""
|
||||
[
|
||||
('path', [acl], fct)
|
||||
('path', [acl], fct, fqtn)
|
||||
]
|
||||
"""
|
||||
|
||||
|
|
|
@ -91,3 +91,33 @@ def gen_starlette_routes(m_dom):
|
|||
),
|
||||
methods=[route['verb']])
|
||||
)
|
||||
|
||||
|
||||
def api_routes(m_dom):
|
||||
"""
|
||||
Yields the description objects for HalfAPI app routes
|
||||
|
||||
Parameters:
|
||||
m_dom (module): the halfapi module
|
||||
|
||||
Returns:
|
||||
Generator[Dict]
|
||||
"""
|
||||
|
||||
m_dom_acl = importlib.import_module('.acl', m_dom.__name__)
|
||||
|
||||
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__)
|
||||
}
|
||||
|
|
@ -1,4 +1,5 @@
|
|||
from .routes import gen_starlette_routes
|
||||
from ..conf import DOMAINSDICT
|
||||
from .routes import gen_starlette_routes, api_routes
|
||||
from .responses import *
|
||||
from starlette.schemas import SchemaGenerator
|
||||
from starlette.routing import Router
|
||||
|
@ -6,6 +7,11 @@ schemas = SchemaGenerator(
|
|||
{"openapi": "3.0.0", "info": {"title": "HalfAPI", "version": "1.0"}}
|
||||
)
|
||||
|
||||
async def get_api_routes(request, *args, **kwargs):
|
||||
return ORJSONResponse({
|
||||
domain: api_routes(m_domain)
|
||||
for domain, m_domain in DOMAINSDICT.items()
|
||||
})
|
||||
|
||||
async def schema_json(request, *args, **kwargs):
|
||||
return ORJSONResponse(
|
||||
|
|
|
@ -1,4 +1,4 @@
|
|||
[pytest]
|
||||
testpaths = tests halfapi
|
||||
addopts = --doctest-modules -rP --log-level debug
|
||||
addopts = --doctest-modules
|
||||
doctest_optionflags = ELLIPSIS
|
||||
|
|
|
@ -5,17 +5,14 @@ from starlette.testclient import TestClient
|
|||
from halfapi.app import application
|
||||
import json
|
||||
|
||||
def test_itworks():
|
||||
def test_get_api_routes():
|
||||
c = TestClient(application)
|
||||
r = json.loads(c.get('/').text)
|
||||
assert r == 'It Works!'
|
||||
r = c.get('/')
|
||||
d_r = r.json()
|
||||
assert isinstance(d_r, dict)
|
||||
|
||||
def test_user():
|
||||
c = TestClient(application)
|
||||
r = c.get('/user')
|
||||
assert r.status_code == 200
|
||||
|
||||
def test_user():
|
||||
def test_current_user():
|
||||
c = TestClient(application)
|
||||
r = c.get('/schema')
|
||||
r = c.get('/halfapi/current_user')
|
||||
assert r.status_code == 200
|
||||
|
|
Loading…
Reference in New Issue