[wip][responses] allow to change return format with "format" route argument, add ODSResponse

This commit is contained in:
Maxime Alves LIRMM@home 2021-12-01 21:16:19 +01:00
parent 20cada4fa0
commit a8e5cfc0ff
2 changed files with 68 additions and 27 deletions

View File

@ -17,7 +17,7 @@ import yaml
from starlette.exceptions import HTTPException
from halfapi.lib import acl
from halfapi.lib.responses import ORJSONResponse
from halfapi.lib.responses import ORJSONResponse, ODSResponse
from halfapi.lib.router import read_router
from halfapi.lib.constants import VERBS
@ -46,39 +46,48 @@ class NoDomainsException(Exception):
def route_decorator(fct: FunctionType, ret_type: str = 'json') -> Coroutine:
""" Returns an async function that can be mounted on a router
"""
if ret_type == 'json':
@wraps(fct)
@acl.args_check
async def wrapped(request, *args, **kwargs):
fct_args_spec = inspect.getfullargspec(fct).args
fct_args = request.path_params.copy()
@wraps(fct)
@acl.args_check
async def wrapped(request, *args, **kwargs):
fct_args_spec = inspect.getfullargspec(fct).args
fct_args = request.path_params.copy()
if 'halfapi' in fct_args_spec:
fct_args['halfapi'] = {
'user': request.user if
'user' in request else None,
'config': request.scope['config'],
'domain': request.scope['domain'],
if 'halfapi' in fct_args_spec:
fct_args['halfapi'] = {
'user': request.user if
'user' in request else None,
'config': request.scope['config'],
'domain': request.scope['domain'],
}
}
if 'data' in fct_args_spec:
fct_args['data'] = kwargs.get('data')
if 'data' in fct_args_spec:
fct_args['data'] = kwargs.get('data')
try:
""" If format argument is specified (either by get or by post param)
"""
ret_type = fct_args.get('data', {}).get('format', 'json')
try:
if ret_type == 'json':
return ORJSONResponse(fct(**fct_args))
except NotImplementedError as exc:
raise HTTPException(501) from exc
except Exception as exc:
# TODO: Write tests
if not isinstance(exc, HTTPException):
raise HTTPException(500) from exc
raise exc
elif ret_type == 'ods':
res = fct(**fct_args)
assert isinstance(res, list)
for elt in res:
assert isinstance(elt, dict)
else:
raise Exception('Return type not available')
return ODSResponse(res)
else:
raise NotImplementedError
except NotImplementedError as exc:
raise HTTPException(501) from exc
except Exception as exc:
# TODO: Write tests
if not isinstance(exc, HTTPException):
raise HTTPException(500) from exc
raise exc
return wrapped

View File

@ -13,10 +13,13 @@ Classes :
- PlainTextResponse
- ServiceUnavailableResponse
- UnauthorizedResponse
- ODSResponse
"""
from datetime import date
import decimal
import typing
from io import BytesIO
import orjson
# asgi framework
@ -110,3 +113,32 @@ class HJSONResponse(ORJSONResponse):
"""
def render(self, content: typing.Generator):
return super().render(list(content))
class ODSResponse(Response):
def __init__(self, d_rows: typing.List[typing.Dict]):
try:
import pyexcel as pe
except ImportError:
""" ODSResponse is not handled
"""
super().__init__(content=
'pyexcel is not installed, ods format not available'
)
return
with BytesIO() as ods_file:
# rows.insert(0, rownames)
self.sheet = pe.Sheet(d_rows)
self.sheet.save_to_memory(
file_type='ods',
stream=ods_file)
filename = f'{date.today()}.ods'
super().__init__(
content=ods_file.getvalue(),
headers={
'Content-Type': 'application/vnd.oasis.opendocument.spreadsheet; charset=UTF-8',
'Content-Disposition': f'attachment; filename="{filename}"'},
status_code = 200)