Project, Web Technologies, Year 3, Semester 1
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 

125 lines
5.1 KiB

from datetime import date, datetime, timezone
from flask.views import MethodView
from flask_smorest import Blueprint
from marshmallow import Schema, fields
import re
from ..decorators import ensure_logged_in
from ..db_utils import get_transactions, get_account, get_accounts, insert_transaction, whose_account, insert_notification, get_forex_rate
from ..models import Account, Notification, Transaction
from ..utils.iban import check_iban
from .. import decorators, returns
bp = Blueprint('transactions', __name__, description='Bank transfers and other transactions')
@bp.route('/')
class TransactionsList(MethodView):
class TransactionsParams(Schema):
account_id = fields.Int(min=1)
class TransactionsGetResponse(returns.SuccessSchema):
transactions = fields.List(fields.Nested(Transaction.TransactionSchema))
@ensure_logged_in
@bp.response(401, returns.ErrorSchema, description='Login failure or not allowed')
@bp.doc(security=[{'Token': []}])
@bp.arguments(TransactionsParams, as_kwargs=True, location='query')
@bp.response(200, TransactionsGetResponse)
def get(self, account_id: int):
"""Get transactions for a certain account"""
if whose_account(account_id) != decorators.user_id:
return returns.abort(returns.UNAUTHORIZED)
# return returns.success(
# transactions=[t.to_json() for t in get_transactions(account_id)]
# )
return returns.success(
transactions=get_transactions(account_id)
)
class TransactionsCreateParams(Schema):
account_id = fields.Int(min=1)
destination_iban = fields.Str()
amount = fields.Int(min=1)
description = fields.Str(default='')
class TransactionsCreateResponse(returns.SuccessSchema):
transaction = fields.Nested(Transaction.TransactionSchema)
@ensure_logged_in
@bp.response(401, returns.ErrorSchema, description='Login failure or not allowed')
@bp.response(404, returns.ErrorSchema, description='Destination account not found')
@bp.response(422, returns.ErrorSchema, description='Invalid account')
@bp.doc(security=[{'Token': []}])
@bp.arguments(TransactionsCreateParams, as_kwargs=True)
@bp.response(200, TransactionsCreateResponse)
def post(self, account_id: int, destination_iban: str, amount: int, description: str = ''):
"""Create a send_transfer transaction"""
if whose_account(account_id) != decorators.user_id:
return returns.abort(returns.UNAUTHORIZED)
account: Account = get_account(account_id=account_id)
if account is None:
return returns.abort(returns.invalid_argument('account_id'))
amount = -1 * abs(amount)
if account.balance + amount < 0:
return returns.abort(returns.NO_BALANCE)
# Check if IBAN is valid
destination_iban = re.sub(r'\s', '', destination_iban)
if not check_iban(destination_iban):
return returns.abort(returns.INVALID_IBAN)
date = datetime.now(timezone.utc).astimezone()
# Check if transaction is to another FoxBank account
reverse_transaction = None
if destination_iban[4:8] == 'FOXB':
for acc in get_accounts():
if destination_iban == acc.iban:
rate = get_forex_rate(account.currency, acc.currency)
reverse_transaction = Transaction.new_transaction(
date_time=date,
transaction_type='receive_transfer',
status='processed',
other_party={'iban': account.iban,},
extra={
'currency': acc.currency,
'amount': int(-amount * rate),
'description': description,
'originalAmount': -amount,
'originalCurrency': account.currency,
},
)
insert_transaction(acc.id, reverse_transaction)
formatted_iban = re.sub(r'(.{4})', r'\1 ', account.iban).strip()
notification = Notification.new_notification(
body=f'Transfer of {-amount // 100}.{-amount % 100:0>2} {account.currency} received from {formatted_iban} in your {acc.custom_name or acc.account_type} account.',
date_time=date,
read=False,
)
insert_notification(whose_account(acc.id), notification)
break
else:
return returns.abort(returns.NOT_FOUND)
transaction = Transaction.new_transaction(
date_time=date,
transaction_type='send_transfer',
status=('processed' if reverse_transaction is not None else 'pending'),
other_party={'iban': destination_iban,},
extra={
'currency': account.currency,
'amount': amount,
'description': description,
},
)
insert_transaction(account_id, transaction)
return returns.success(transaction=transaction)