foldbank-mock/backend/foldbank/api.py

import logging
from collections import OrderedDict
from random import randrange

from django.conf import settings
from django.contrib.auth import get_user_model
from drf_yasg import openapi
from drf_yasg.utils import swagger_auto_schema
from foldbank.models import Account, RecurringPayment, Transaction, User
from rest_framework import generics, serializers
from rest_framework.pagination import LimitOffsetPagination as _LimitOffsetPagination
from rest_framework.response import Response

UserModel = get_user_model()
logger = logging.getLogger(__name__)


def create_serializer_class(name, fields):
    return type(name, (serializers.Serializer,), fields)


def inline_serializer(*, fields, data=None, **kwargs):
    serializer_class = create_serializer_class(name="", fields=fields)

    if data is not None:
        return serializer_class(data=data, **kwargs)

    return serializer_class(**kwargs)


def get_paginated_response(
    *,
    pagination_class,
    serializer_class,
    queryset,
    request,
    view,
    mod_func=None,
):
    paginator = pagination_class()

    page = paginator.paginate_queryset(queryset, request, view=view)

    if page is not None:
        if mod_func:
            page = mod_func(page)
        serializer = serializer_class(page, many=True)
        return paginator.get_paginated_response(serializer.data)

    serializer = serializer_class(queryset, many=True)

    return Response(serializer.data)


class LimitOffsetPagination(_LimitOffsetPagination):
    default_limit = 10
    max_limit = 50

    def get_paginated_data(self, data):
        return OrderedDict(
            [
                ("limit", self.limit),
                ("offset", self.offset),
                ("count", self.count),
                ("next", self.get_next_link()),
                ("previous", self.get_previous_link()),
                ("results", data),
            ]
        )

    def get_paginated_response(self, data):
        """
        We redefine this method in order to return `limit` and `offset`.
        This is used by the frontend to construct the pagination itself.
        """
        return Response(
            OrderedDict(
                [
                    ("limit", self.limit),
                    ("offset", self.offset),
                    ("count", self.count),
                    ("next", self.get_next_link()),
                    ("previous", self.get_previous_link()),
                    ("results", data),
                ]
            ),
        )


def get_transactions(user):
    return Transaction.objects.filter(from_account__user=user).order_by("-created_at")


class TransactionsAPI(generics.ListAPIView):
    authentication_classes = []
    permission_classes = []

    class Pagination(LimitOffsetPagination):
        pass

    class TransactionsInputSerializer(serializers.Serializer):
        bank_id = serializers.UUIDField(default=None)

    class TransactionsOutputSerializer(serializers.Serializer):
        id = serializers.UUIDField()
        created_at = serializers.DateTimeField()
        amount = serializers.IntegerField()
        tag = inline_serializer(
            required=True,
            fields={
                "title": serializers.CharField(required=True),
                "icon_type": serializers.CharField(required=True),
            },
        )
        to_account = inline_serializer(
            required=True,
            fields={
                "holders_name": serializers.CharField(required=True),
            },
        )

    success_response = openapi.Response(
        "Success Response", TransactionsOutputSerializer
    )

    @swagger_auto_schema(
        query_serializer=TransactionsInputSerializer,
        operation_summary="Transactions List",
        responses={
            200: success_response,
        },
        tags=["Transactions"],
    )
    def get(self, request):
        qs = []

        serializer = self.TransactionsInputSerializer(data=request.query_params)
        serializer.is_valid(raise_exception=True)

        # TODO: hardcoding nishant to mitigate user login
        user = User.objects.get(username="nishant")
        qs = get_transactions(user=user)

        def modify_qs(qs):
            return qs

        return get_paginated_response(
            pagination_class=self.Pagination,
            serializer_class=self.TransactionsOutputSerializer,
            queryset=qs,
            request=request,
            view=self,
            mod_func=modify_qs,
        )


def get_bank_accounts(user):
    return Account.objects.filter(user=user)


class AccountsAPI(generics.ListAPIView):
    authentication_classes = []
    permission_classes = []

    class Pagination(LimitOffsetPagination):
        pass

    class AccountsOutputSerializer(serializers.Serializer):
        id = serializers.UUIDField()
        created_at = serializers.DateTimeField()
        balance = serializers.IntegerField()
        bank = inline_serializer(
            required=True,
            fields={
                "name": serializers.CharField(required=True),
                "logo": serializers.CharField(required=True),
            },
        )
        account_number = serializers.CharField()
        ifsc_code = serializers.CharField()
        swift_bic = serializers.CharField()
        holders_name = serializers.CharField()

    success_response = openapi.Response("Success Response", AccountsOutputSerializer)

    @swagger_auto_schema(
        operation_summary="Accounts List",
        responses={
            200: success_response,
        },
        tags=["Accounts"],
    )
    def get(self, request):
        qs = []

        # TODO: hardcoding nishant to mitigate user login
        user = User.objects.get(username="nishant")
        qs = get_bank_accounts(user=user)

        def modify_qs(qs):
            return qs

        return get_paginated_response(
            pagination_class=self.Pagination,
            serializer_class=self.AccountsOutputSerializer,
            queryset=qs,
            request=request,
            view=self,
            mod_func=modify_qs,
        )


def get_upcoming_recurring_payments(user):
    return RecurringPayment.objects.filter(from_account__user=user)


class UpcomingRecurringPaymentAPI(generics.ListAPIView):
    authentication_classes = []
    permission_classes = []

    class Pagination(LimitOffsetPagination):
        pass

    class UpcomingRecurringPaymentOutputSerializer(serializers.Serializer):
        id = serializers.UUIDField()
        amount = serializers.IntegerField()
        due_on = serializers.DateTimeField()
        frequency = serializers.DurationField()
        to_account = inline_serializer(
            required=True,
            fields={
                "holders_name": serializers.CharField(required=True),
            },
        )


    success_response = openapi.Response("Success Response", UpcomingRecurringPaymentOutputSerializer)

    @swagger_auto_schema(
        operation_summary="Recurring Payments List",
        responses={
            200: success_response,
        },
        tags=["Recurring Payments"],
    )
    def get(self, request):
        qs = []

        # TODO: hardcoding nishant to mitigate user login
        user = User.objects.get(username="nishant")
        qs = get_upcoming_recurring_payments(user=user)

        def modify_qs(qs):
            return qs

        return get_paginated_response(
            pagination_class=self.Pagination,
            serializer_class=self.UpcomingRecurringPaymentOutputSerializer,
            queryset=qs,
            request=request,
            view=self,
            mod_func=modify_qs,
        )