276 lines
8 KiB
Python
276 lines
8 KiB
Python
import logging
|
|
from collections import OrderedDict
|
|
from random import randrange
|
|
|
|
from django.conf import settings
|
|
from django.contrib.auth import get_user_model
|
|
from django.utils.decorators import method_decorator
|
|
from django.views.decorators.cache import cache_page
|
|
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"],
|
|
)
|
|
@method_decorator(cache_page(60 * 60 * 2))
|
|
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"],
|
|
)
|
|
@method_decorator(cache_page(60 * 60 * 2))
|
|
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()
|
|
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", UpcomingRecurringPaymentOutputSerializer
|
|
)
|
|
|
|
@swagger_auto_schema(
|
|
operation_summary="Recurring Payments List",
|
|
responses={
|
|
200: success_response,
|
|
},
|
|
tags=["Recurring Payments"],
|
|
)
|
|
@method_decorator(cache_page(60 * 60 * 2))
|
|
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,
|
|
)
|