311 lines
10 KiB
Python
311 lines
10 KiB
Python
"""Balance plugin
|
|
|
|
Extends users plugin with balance functions
|
|
"""
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from flaschengeist.utils.HTTP import no_content
|
|
from flask import Blueprint, request, jsonify
|
|
from werkzeug.exceptions import Forbidden, BadRequest
|
|
|
|
from flaschengeist import logger
|
|
from flaschengeist.utils import HTTP
|
|
from flaschengeist.models.session import Session
|
|
from flaschengeist.utils.datetime import from_iso_format
|
|
from flaschengeist.utils.decorators import login_required
|
|
from flaschengeist.controller import userController
|
|
from flaschengeist.plugins import Plugin, before_update_user
|
|
|
|
from . import balance_controller, permissions, models
|
|
|
|
balance_bp = Blueprint("balance", __name__)
|
|
|
|
|
|
class BalancePlugin(Plugin):
|
|
models = models
|
|
|
|
def __init__(self, config):
|
|
super().__init__(blueprint=balance_bp, permissions=permissions.permissions)
|
|
|
|
@before_update_user
|
|
def set_default_limit(user):
|
|
if "limit" in config:
|
|
limit = config["limit"]
|
|
logger.debug("Setting default limit of {} to user {}".format(limit, user.userid))
|
|
balance_controller.set_limit(user, limit, override=False)
|
|
|
|
def install(self):
|
|
from flaschengeist.database import db
|
|
|
|
db.create_all()
|
|
|
|
|
|
def str2bool(string: str):
|
|
if string.lower() in ["true", "yes", "1"]:
|
|
return True
|
|
elif string.lower() in ["false", "no", "0"]:
|
|
return False
|
|
raise ValueError
|
|
|
|
|
|
@balance_bp.route("/users/<userid>/balance/shortcuts", methods=["GET", "PUT"])
|
|
@login_required()
|
|
def get_shortcuts(userid, current_session: Session):
|
|
"""Get balance shortcuts of an user
|
|
|
|
Route: ``/users/<userid>/balance/shortcuts`` | Method: ``GET`` or ``PUT``
|
|
POST-data: On ``PUT`` json encoded array of floats
|
|
|
|
Args:
|
|
userid: Userid identifying the user
|
|
current_session: Session sent with Authorization Header
|
|
|
|
Returns:
|
|
GET: JSON object containing the shortcuts as float array or HTTP error
|
|
PUT: HTTP-created or HTTP error
|
|
"""
|
|
if userid != current_session._user.userid:
|
|
raise Forbidden
|
|
|
|
user = userController.get_user(userid)
|
|
if request.method == "GET":
|
|
return jsonify(user.get_attribute("balance_shortcuts", []))
|
|
else:
|
|
data = request.get_json()
|
|
if not isinstance(data, list) or not all(isinstance(n, (int, float)) for n in data):
|
|
raise BadRequest
|
|
data.sort(reverse=True)
|
|
user.set_attribute("balance_shortcuts", data)
|
|
userController.persist()
|
|
return no_content()
|
|
|
|
|
|
@balance_bp.route("/users/<userid>/balance/limit", methods=["GET"])
|
|
@login_required()
|
|
def get_limit(userid, current_session: Session):
|
|
"""Get limit of an user
|
|
|
|
Route: ``/users/<userid>/balance/limit`` | Method: ``GET``
|
|
|
|
Args:
|
|
userid: Userid identifying the user
|
|
current_session: Session sent with Authorization Header
|
|
|
|
Returns:
|
|
JSON object containing the limit (or Null if no limit set) or HTTP error
|
|
"""
|
|
user = userController.get_user(userid)
|
|
if (user != current_session._user and not current_session._user.has_permission(permissions.SET_LIMIT)) or (
|
|
user == current_session._user and not user.has_permission(permissions.SHOW)
|
|
):
|
|
raise Forbidden
|
|
|
|
return {"limit": balance_controller.get_limit(user)}
|
|
|
|
|
|
@balance_bp.route("/users/<userid>/balance/limit", methods=["PUT"])
|
|
@login_required(permissions.SET_LIMIT)
|
|
def set_limit(userid, current_session: Session):
|
|
"""Set the limit of an user
|
|
|
|
Route: ``/users/<userid>/balance/limit`` | Method: ``PUT``
|
|
|
|
POST-data: ``{limit: float}``
|
|
|
|
Args:
|
|
userid: Userid identifying the user
|
|
current_session: Session sent with Authorization Header
|
|
|
|
Returns:
|
|
HTTP-200 or HTTP error
|
|
"""
|
|
user = userController.get_user(userid)
|
|
data = request.get_json()
|
|
try:
|
|
limit = data["limit"]
|
|
except (TypeError, KeyError):
|
|
raise BadRequest
|
|
balance_controller.set_limit(user, limit)
|
|
return HTTP.no_content()
|
|
|
|
|
|
@balance_bp.route("/users/<userid>/balance", methods=["GET"])
|
|
@login_required(permission=permissions.SHOW)
|
|
def get_balance(userid, current_session: Session):
|
|
"""Get balance of user, optionally filtered
|
|
|
|
Route: ``/users/<userid>/balance`` | Method: ``GET``
|
|
|
|
GET-parameters: ```{from?: string, to?: string}```
|
|
|
|
Args:
|
|
userid: Userid of user to get balance from
|
|
current_session: Session sent with Authorization Header
|
|
|
|
Returns:
|
|
JSON object containing credit, debit and balance or HTTP error
|
|
"""
|
|
if userid != current_session._user.userid and not current_session._user.has_permission(permissions.SHOW_OTHER):
|
|
raise Forbidden
|
|
|
|
# Might raise NotFound
|
|
user = userController.get_user(userid)
|
|
|
|
start = request.args.get("from")
|
|
if start:
|
|
start = from_iso_format(start)
|
|
else:
|
|
start = datetime.fromtimestamp(0, tz=timezone.utc)
|
|
|
|
end = request.args.get("to")
|
|
if end:
|
|
end = from_iso_format(end)
|
|
else:
|
|
end = datetime.now(tz=timezone.utc)
|
|
|
|
balance = balance_controller.get_balance(user, start, end)
|
|
return {"credit": balance[0], "debit": balance[1], "balance": balance[2]}
|
|
|
|
|
|
@balance_bp.route("/users/<userid>/balance/transactions", methods=["GET"])
|
|
@login_required(permission=permissions.SHOW)
|
|
def get_transactions(userid, current_session: Session):
|
|
"""Get transactions of user, optionally filtered
|
|
Returns also count of transactions if limit is set (e.g. just count with limit = 0)
|
|
|
|
Route: ``/users/<userid>/balance/transactions`` | Method: ``GET``
|
|
|
|
GET-parameters: ```{from?: string, to?: string, limit?: int, offset?: int}```
|
|
|
|
Args:
|
|
userid: Userid of user to get transactions from
|
|
current_session: Session sent with Authorization Header
|
|
|
|
Returns:
|
|
JSON Object {transactions: Transaction[], count?: number} or HTTP error
|
|
"""
|
|
if userid != current_session._user.userid and not current_session._user.has_permission(permissions.SHOW_OTHER):
|
|
raise Forbidden
|
|
|
|
# Might raise NotFound
|
|
user = userController.get_user(userid)
|
|
|
|
start = request.args.get("from")
|
|
if start:
|
|
start = from_iso_format(start)
|
|
end = request.args.get("to")
|
|
if end:
|
|
end = from_iso_format(end)
|
|
show_reversals = request.args.get("showReversals", False)
|
|
show_cancelled = request.args.get("showCancelled", True)
|
|
limit = request.args.get("limit")
|
|
offset = request.args.get("offset")
|
|
try:
|
|
if limit is not None:
|
|
limit = int(limit)
|
|
if offset is not None:
|
|
offset = int(offset)
|
|
if not isinstance(show_reversals, bool):
|
|
show_reversals = str2bool(show_reversals)
|
|
if not isinstance(show_cancelled, bool):
|
|
show_cancelled = str2bool(show_cancelled)
|
|
except ValueError:
|
|
raise BadRequest
|
|
|
|
transactions, count = balance_controller.get_transactions(
|
|
user, start, end, limit, offset, show_reversal=show_reversals, show_cancelled=show_cancelled
|
|
)
|
|
return {"transactions": transactions, "count": count}
|
|
|
|
|
|
@balance_bp.route("/users/<userid>/balance", methods=["PUT"])
|
|
@login_required()
|
|
def change_balance(userid, current_session: Session):
|
|
"""Change balance of an user
|
|
If ``sender`` is preset in POST-data, the action is handled as a transfer from ``sender`` to user.
|
|
|
|
Route: ``/users/<userid>/balance`` | Method: ``PUT``
|
|
|
|
POST-data: ``{amount: float, sender: string}``
|
|
|
|
Args:
|
|
userid: userid identifying user to change balance
|
|
current_session: Session sent with Authorization Header
|
|
|
|
Returns:
|
|
JSON encoded transaction (201) or HTTP error
|
|
"""
|
|
|
|
data = request.get_json()
|
|
try:
|
|
amount = data["amount"]
|
|
except (TypeError, KeyError):
|
|
raise BadRequest
|
|
|
|
sender = data.get("sender", None)
|
|
user = userController.get_user(userid)
|
|
|
|
if sender:
|
|
sender = userController.get_user(sender)
|
|
if sender == user:
|
|
raise BadRequest
|
|
|
|
if (sender == current_session._user and sender.has_permission(permissions.SEND)) or (
|
|
sender != current_session._user and current_session._user.has_permission(permissions.SEND_OTHER)
|
|
):
|
|
return HTTP.created(balance_controller.send(sender, user, amount, current_session._user))
|
|
|
|
elif (
|
|
amount < 0
|
|
and (
|
|
(user == current_session._user and user.has_permission(permissions.DEBIT_OWN))
|
|
or current_session._user.has_permission(permissions.DEBIT)
|
|
)
|
|
) or (amount > 0 and current_session._user.has_permission(permissions.CREDIT)):
|
|
return HTTP.created(balance_controller.change_balance(user, data["amount"], current_session._user))
|
|
|
|
raise Forbidden
|
|
|
|
|
|
@balance_bp.route("/balance/<int:transaction_id>", methods=["DELETE"])
|
|
@login_required()
|
|
def reverse_transaction(transaction_id, current_session: Session):
|
|
"""Reverse a transaction
|
|
|
|
Route: ``/balance/<int:transaction_id>`` | Method: ``DELETE``
|
|
|
|
Args:
|
|
transaction_id: Identifier of the transaction
|
|
current_session: Session sent with Authorization Header
|
|
|
|
Returns:
|
|
JSON encoded reversal (transaction) (201) or HTTP error
|
|
"""
|
|
|
|
transaction = balance_controller.get_transaction(transaction_id)
|
|
if current_session._user.has_permission(permissions.REVERSAL) or (
|
|
transaction.sender_ == current_session._user
|
|
and (datetime.now(tz=timezone.utc) - transaction.time).total_seconds() < 10
|
|
):
|
|
reversal = balance_controller.reverse_transaction(transaction, current_session._user)
|
|
return HTTP.created(reversal)
|
|
raise Forbidden
|
|
|
|
|
|
@balance_bp.route("/balance", methods=["GET"])
|
|
@login_required(permission=permissions.SHOW_OTHER)
|
|
def get_balances(current_session: Session):
|
|
"""Get all balances
|
|
|
|
Route: ``/balance`` | Method: ``GET``
|
|
|
|
Args:
|
|
current_session: Session sent with Authorization Header
|
|
|
|
Returns:
|
|
JSON Array containing credit, debit and userid for each user or HTTP error
|
|
"""
|
|
balances = balance_controller.get_balances()
|
|
return jsonify([{"userid": u, "credit": v[0], "debit": v[1]} for u, v in balances.items()])
|