foxbank/server/foxbank_server/apis/transactions.py

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)