macao-pos/pos/rest.py

117 lines
3.4 KiB
Python

from functools import wraps
from aiohttp.web import json_response
from pos.database import User, ProductCategory, AccessToken
def needs(*needed):
def decorator(func):
@wraps(func)
async def wrapper(request):
request_json = await request.json()
if not all(k in request_json.keys() for k in needed) or
return json_response({
'err': 'malformed_request',
'msg': 'Missing one or more keys: {}.'.format(
", ".join(needed))
}, status=400)
else:
return func(request)
return wrapper
return decorator
def auth_required(func):
@wraps(func)
async def wrapper(request):
db = request.app['db']
headers = request.headers
if 'Authorization' not in headers.keys():
return json_response({'err': 'malformed_request',
'msg': 'Missing Authorization header.'},
status=400)
else:
remote_token = headers['Authorization']
with db.get_session() as session:
token = session.query(AccessToken) \
.filter_by(token=remote_token) \
.one_or_none()
if not token:
return json_response({'err': 'unauthorized',
'msg': 'The token is not valid.'},
status=401)
elif not token.is_valid():
return json_response({'err': 'forbidden',
'msg': 'The token has expired.'},
status=403)
else:
return await func(request)
return wrapper
@needs('username', 'password')
async def token_create(request):
db = request.app['db']
request_json = await request.json()
username = request_json['username']
password = request_json['password']
with db.get_session() as session:
user = session.query(User) \
.filter_by(username=username) \
.one_or_none()
if not user or user.password != password:
return json_response({'err': 'invalid_credentials'},
status=400)
with db.get_session() as session:
token = AccessToken(user=user)
session.add(token)
return json_response({
'token': token.token,
'created_at': token.created_at.isoformat(),
'expires_at': token.created_at.isoformat()
})
@auth_required
async def token_destroy(request):
db = request.app['db']
remote_token = request.headers['Authorization']
with db.get_session() as session:
token = session.query(AccessToken) \
.filter_by(token=remote_token) \
.one_or_none()
token.is_active = False
session.add(token)
return json_response({}, status=200)
@auth_required
async def product_list(request):
db = request.app['db']
with db.get_session() as session:
categories = session.query(ProductCategory).all()
return json_response({
'categories': [{
'uid': c.uid,
'name': c.name,
'products': [{
'uid': p.uid,
'name': p.name,
'price': p.price
} for p in c.products]
} for c in categories]
})