masq
/
halfapi
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HalfAPI class clean and rewrite

This commit is contained in:
Maxime Alves LIRMM 2021-11-22 18:08:02 +01:00
parent ad9bd45ba0
commit 0173eb6d72
11 changed files with 251 additions and 219 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

145
halfapi/app.py
View File

@ -1,138 +1,9 @@
#!/usr/bin/env python3
"""
app.py is the file that is read when launching the application using an asgi
runner.
from .halfapi import HalfAPI
from .conf import PRODUCTION, SECRET, DOMAINS, CONFIG
It defines the following globals :
- routes (contains the Route objects for the application)
- application (the asgi application itself - a starlette object)
"""
import logging
import time
# asgi framework
from starlette.applications import Starlette
from starlette.authentication import UnauthenticatedUser
from starlette.middleware import Middleware
from starlette.routing import Route
from starlette.responses import Response, PlainTextResponse
from starlette.middleware.authentication import AuthenticationMiddleware
from timing_asgi import TimingMiddleware
from timing_asgi.integrations import StarletteScopeToName
# module libraries
from .lib.domain_middleware import DomainMiddleware
from .lib.timing import HTimingClient
from halfapi.lib.jwt_middleware import JWTAuthenticationBackend
from halfapi.lib.responses import (ORJSONResponse, UnauthorizedResponse,
NotFoundResponse, InternalServerErrorResponse, NotImplementedResponse,
ServiceUnavailableResponse)
from halfapi.lib.routes import gen_starlette_routes, debug_routes
from halfapi.lib.schemas import get_api_routes, get_api_domain_routes, schema_json, get_acls
from halfapi.logging import logger, config_logging
from halfapi import __version__
class HalfAPI:
def __init__(self, config=None):
config_logging(logging.DEBUG)
if config:
SECRET = config.get('SECRET')
PRODUCTION = config.get('PRODUCTION')
DOMAINS = config.get('DOMAINS', {})
CONFIG = config.get('CONFIG', {
'domains': DOMAINS
})
else:
from halfapi.conf import CONFIG, SECRET, PRODUCTION, DOMAINS
routes = [ Route('/', get_api_routes(DOMAINS)) ]
for route in self.routes():
routes.append(route)
if not PRODUCTION:
for route in debug_routes():
routes.append( route )
if DOMAINS:
for route in gen_starlette_routes(DOMAINS):
routes.append(route)
for domain, m_domain in DOMAINS.items():
routes.append(
Route(
f'/{domain}',
get_api_domain_routes(m_domain)
)
)
self.application = Starlette(
debug=not PRODUCTION,
routes=routes,
exception_handlers={
401: UnauthorizedResponse,
404: NotFoundResponse,
500: InternalServerErrorResponse,
501: NotImplementedResponse,
503: ServiceUnavailableResponse
}
)
self.application.add_middleware(
DomainMiddleware,
config=CONFIG
)
if SECRET:
self.SECRET = SECRET
self.application.add_middleware(
AuthenticationMiddleware,
backend=JWTAuthenticationBackend(secret_key=SECRET)
)
if not PRODUCTION:
self.application.add_middleware(
TimingMiddleware,
client=HTimingClient(),
metric_namer=StarletteScopeToName(prefix="halfapi",
starlette_app=self.application)
)
logger.info('CONFIG:\n%s', CONFIG)
@property
def version(self):
return __version__
async def version_async(self, request, *args, **kwargs):
return Response(self.version)
def routes(self):
""" Halfapi default routes
"""
async def get_user(request, *args, **kwargs):
return ORJSONResponse({'user':request.user})
yield Route('/halfapi/whoami', get_user)
yield Route('/halfapi/schema', schema_json)
yield Route('/halfapi/acls', get_acls)
yield Route('/halfapi/version', self.version_async)
application = HalfAPI().application
application = HalfAPI({
'PRODUCTION': PRODUCTION,
'SECRET': SECRET,
'DOMAINS': DOMAINS,
'CONFIG': CONFIG,
}).application

168
halfapi/halfapi.py Normal file
View File

@ -0,0 +1,168 @@
#!/usr/bin/env python3
"""
app.py is the file that is read when launching the application using an asgi
runner.
It defines the following globals :
- routes (contains the Route objects for the application)
- application (the asgi application itself - a starlette object)
"""
import logging
import time
from datetime import datetime
# asgi framework
from starlette.applications import Starlette
from starlette.authentication import UnauthenticatedUser
from starlette.exceptions import HTTPException
from starlette.middleware import Middleware
from starlette.routing import Route, Mount
from starlette.requests import Request
from starlette.responses import Response, PlainTextResponse
from starlette.middleware.authentication import AuthenticationMiddleware
from timing_asgi import TimingMiddleware
from timing_asgi.integrations import StarletteScopeToName
# module libraries
from .lib.domain_middleware import DomainMiddleware
from .lib.timing import HTimingClient
from .lib.domain import NoDomainsException
from halfapi.lib.jwt_middleware import JWTAuthenticationBackend
from halfapi.lib.responses import (ORJSONResponse, UnauthorizedResponse,
NotFoundResponse, InternalServerErrorResponse, NotImplementedResponse,
ServiceUnavailableResponse)
from halfapi.lib.routes import gen_domain_routes, JSONRoute
from halfapi.lib.schemas import get_api_routes, get_api_domain_routes, schema_json, get_acls
from halfapi.logging import logger, config_logging
from halfapi import __version__
class HalfAPI:
def __init__(self, config, routes_dict=None):
config_logging(logging.DEBUG)
SECRET = config.get('SECRET')
PRODUCTION = config.get('PRODUCTION', True)
DOMAINS = config.get('DOMAINS', {})
CONFIG = config.get('CONFIG', {
'domains': DOMAINS
})
if not (DOMAINS or routes_dict):
raise NoDomainsException()
self.PRODUCTION = PRODUCTION
self.CONFIG = CONFIG
self.DOMAINS = DOMAINS
self.SECRET = SECRET
""" The base route contains the route schema
"""
self.api_routes = get_api_routes(DOMAINS)
routes = [ Route('/', JSONRoute(self.api_routes)) ]
""" HalfAPI routes (if not PRODUCTION, includes debug routes)
"""
routes.append(
Mount('/halfapi', routes=list(self.routes()))
)
if DOMAINS:
""" Mount the domain routes
"""
for domain, m_domain in DOMAINS.items():
if domain not in self.api_routes.keys():
raise Exception(f'The domain does not have a schema: {domain}')
routes.append(
Route(f'/{domain}', JSONRoute(self.api_routes[domain]))
)
routes.append(
Mount(f'/{domain}', routes=gen_domain_routes(m_domain))
)
self.application = Starlette(
debug=not PRODUCTION,
routes=routes,
exception_handlers={
401: UnauthorizedResponse,
404: NotFoundResponse,
500: InternalServerErrorResponse,
501: NotImplementedResponse,
503: ServiceUnavailableResponse
}
)
self.application.add_middleware(
DomainMiddleware,
config=CONFIG
)
if SECRET:
self.SECRET = SECRET
self.application.add_middleware(
AuthenticationMiddleware,
backend=JWTAuthenticationBackend(secret_key=SECRET)
)
if not PRODUCTION:
self.application.add_middleware(
TimingMiddleware,
client=HTimingClient(),
metric_namer=StarletteScopeToName(prefix="halfapi",
starlette_app=self.application)
)
@property
def version(self):
return __version__
async def version_async(self, request, *args, **kwargs):
return Response(self.version)
def routes(self):
""" Halfapi default routes
"""
async def get_user(request, *args, **kwargs):
return ORJSONResponse({'user':request.user})
yield Route('/whoami', get_user)
yield Route('/schema', schema_json)
yield Route('/acls', get_acls)
yield Route('/version', self.version_async)
""" Halfapi debug routes definition
"""
if self.PRODUCTION:
return
""" Debug routes
"""
async def debug_log(request: Request, *args, **kwargs):
logger.debug('debuglog# %s', {datetime.now().isoformat()})
logger.info('debuglog# %s', {datetime.now().isoformat()})
logger.warning('debuglog# %s', {datetime.now().isoformat()})
logger.error('debuglog# %s', {datetime.now().isoformat()})
logger.critical('debuglog# %s', {datetime.now().isoformat()})
return Response('')
yield Route('/log', debug_log)
async def error_code(request: Request, *args, **kwargs):
code = request.path_params['code']
raise HTTPException(code)
yield Route('/error/{code:int}', error_code)
async def exception(request: Request, *args, **kwargs):
raise Exception('Test exception')
yield Route('/exception', exception)

5
halfapi/lib/domain.py
View File

@ -36,6 +36,11 @@ class UndefinedRoute(Exception):
class UndefinedFunction(Exception):
pass
class NoDomainsException(Exception):
""" The exception that is raised when HalfAPI is called without domains
"""
pass
def route_decorator(fct: FunctionType, ret_type: str = 'json') -> Coroutine:
""" Returns an async function that can be mounted on a router
"""

80
halfapi/lib/routes.py
View File

@ -4,10 +4,10 @@ Routes module
Fonctions :
- route_acl_decorator
- gen_domain_routes
- gen_starlette_routes
- api_routes
- api_acls
- debug_routes
Exception :
- DomainNotFoundError
@ -15,7 +15,7 @@ Exception :
"""
from datetime import datetime
from functools import partial, wraps
from typing import Callable, List, Dict, Generator, Tuple
from typing import Callable, Coroutine, List, Dict, Generator, Tuple, Any
from types import ModuleType, FunctionType
from starlette.exceptions import HTTPException
@ -24,6 +24,7 @@ from starlette.requests import Request
from starlette.responses import Response, PlainTextResponse
from halfapi.lib.domain import gen_router_routes, domain_acls
from halfapi.lib.responses import ORJSONResponse
from ..conf import DOMAINSDICT
from ..logging import logger
@ -32,7 +33,24 @@ class DomainNotFoundError(Exception):
""" Exception when a domain is not importable
"""
def route_acl_decorator(fct: Callable = None, params: List[Dict] = None):
def JSONRoute(data: Any) -> Coroutine:
"""
Returns a route function that returns the data as JSON
Parameters:
data (Any):
The data to return
Returns:
async function
"""
async def wrapped(request, *args, **kwargs):
return ORJSONResponse(data)
return wrapped
def route_acl_decorator(fct: Callable = None, params: List[Dict] = None) -> Coroutine:
"""
Decorator for async functions that calls pre-conditions functions
and appends kwargs to the target function
@ -94,6 +112,26 @@ def route_acl_decorator(fct: Callable = None, params: List[Dict] = None):
return caller
def gen_domain_routes(m_domain: ModuleType):
"""
Yields the Route objects for a domain
Parameters:
m_domains: ModuleType
Returns:
Generator(Route)
"""
for path, verb, fct, params in gen_router_routes(m_domain, []):
yield (
Route(f'/{path}',
route_acl_decorator(
fct,
params
),
methods=[verb])
)
def gen_starlette_routes(d_domains: Dict[str, ModuleType]) -> Generator:
"""
Yields the Route objects for HalfAPI app
@ -104,18 +142,8 @@ def gen_starlette_routes(d_domains: Dict[str, ModuleType]) -> Generator:
Returns:
Generator(Route)
"""
for domain_name, m_domain in d_domains.items():
for path, verb, fct, params in gen_router_routes(m_domain, []):
yield (
Route(f'/{domain_name}/{path}',
route_acl_decorator(
fct,
params
),
methods=[verb])
)
yield from gen_domain_routes(m_domain)
def api_routes(m_dom: ModuleType) -> Tuple[Dict, Dict]:
@ -177,27 +205,3 @@ def api_acls(request):
res[domain][acl_name] = fct_result
return res
def debug_routes():
""" Halfapi debug routes definition
"""
async def debug_log(request: Request, *args, **kwargs):
logger.debug('debuglog# %s', {datetime.now().isoformat()})
logger.info('debuglog# %s', {datetime.now().isoformat()})
logger.warning('debuglog# %s', {datetime.now().isoformat()})
logger.error('debuglog# %s', {datetime.now().isoformat()})
logger.critical('debuglog# %s', {datetime.now().isoformat()})
return Response('')
yield Route('/halfapi/log', debug_log)
async def error_code(request: Request, *args, **kwargs):
code = request.path_params['code']
raise HTTPException(code)
yield Route('/halfapi/error/{code:int}', error_code)
async def exception(request: Request, *args, **kwargs):
raise Exception('Test exception')
yield Route('/halfapi/exception', exception)

9
halfapi/lib/schemas.py
View File

@ -25,7 +25,7 @@ SCHEMAS = SchemaGenerator(
{"openapi": "3.0.0", "info": {"title": "HalfAPI", "version": __version__}}
)
def get_api_routes(domains: Dict[str, ModuleType]) -> Coroutine:
def get_api_routes(domains: Dict[str, ModuleType]) -> Dict:
"""
description: Returns the current API routes dictionary
as a JSON object
@ -63,16 +63,11 @@ def get_api_routes(domains: Dict[str, ModuleType]) -> Coroutine:
}
}
"""
routes = {
return {
domain: api_routes(m_domain)[0]
for domain, m_domain in domains.items()
}
async def wrapped(request, *args, **kwargs):
return ORJSONResponse(routes)
return wrapped
def get_api_domain_routes(m_domain: ModuleType) -> Coroutine:
routes, _ = api_routes(m_domain)

17
tests/conftest.py
View File

@ -18,6 +18,7 @@ from starlette.responses import PlainTextResponse
from starlette.middleware.authentication import AuthenticationMiddleware
from starlette.testclient import TestClient
from halfapi import __version__
from halfapi.halfapi import HalfAPI
from halfapi.cli.cli import cli
from halfapi.cli.init import init, format_halfapi_etc
from halfapi.cli.domain import domain, create_domain
@ -95,7 +96,6 @@ def cli_runner():
yield cli_runner_
@pytest.fixture
def halfapicli(cli_runner):
def caller(*args):
@ -279,22 +279,27 @@ def dummy_project():
def routers():
sys.path.insert(0, './tests')
from dummy_domain import routers
from .dummy_domain import routers
return routers
@pytest.fixture
def application_debug():
from halfapi.app import HalfAPI
def application_debug(routers):
return HalfAPI({
'SECRET':'turlututu',
'PRODUCTION':False
'PRODUCTION':False,
'DOMAINS': {
'dummy_domain': routers
},
'CONFIG':{
'domains': {'dummy_domain':routers},
'domain_config': {'dummy_domain': {'test': True}}
}
}).application
@pytest.fixture
def application_domain(routers):
from halfapi.app import HalfAPI
return HalfAPI({
'SECRET':'turlututu',
'PRODUCTION':True,

5
tests/dummy_domain/act/__init__.py
View File

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

13
tests/dummy_domain/act/personne/__init__.py
View File

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

10
tests/dummy_domain/act/personne/eo.py
View File

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

15
tests/test_app.py Normal file
View File

@ -0,0 +1,15 @@
import os
from starlette.applications import Starlette
from unittest.mock import MagicMock, patch
from halfapi.halfapi import HalfAPI
from halfapi.lib.domain import NoDomainsException
def test_halfapi_dummy_domain():
with patch('starlette.applications.Starlette') as mock:
mock.return_value = MagicMock()
halfapi = HalfAPI({
'DOMAINS': {
'dummy_domain': '.routers'
}
})

3
tests/test_lib_schemas.py
View File

@ -14,7 +14,6 @@ def test_schemas_dict_dom():
'dummy_domain':routers})
assert isinstance(schema, dict)
def test_get_api_routes(project_runner, application_debug):
c = TestClient(application_debug)
r = c.get('/')
@ -43,5 +42,3 @@ def test_get_api_dummy_domain_routes(application_domain, routers):
assert 'GET' in d_r['abc/alphabet']
assert len(d_r['abc/alphabet']['GET']) > 0
assert 'acl' in d_r['abc/alphabet']['GET'][0]