BelvoApp/transactions/views.py

import coreschema
from django.http import Http404
from rest_framework import viewsets, mixins
from rest_framework import permissions
import coreapi
from rest_framework.schemas import AutoSchema
from rest_framework.response import Response
from .mixins import TransactionsMixin
from django.db.models import (Sum, F, Case, FloatField,
                              When)
from .serializers import (CreateTransactionSerializer,
                          BalanceByAccountSerializer)
from .models import Transaction
from django.contrib.auth import get_user_model

User = get_user_model()


class CustomSchema(AutoSchema):
    """
    Custom Scheme for documentation
    """

    def __init__(self, manual_fields=None):
        """
        Parameters:

        Add Manual Fields
        """
        super(AutoSchema, self).__init__()
        self._manual_fields = [
            coreapi.Field(
                name='date_from',
                location='query',
                required=False,
                schema=coreschema.Integer(title='date_from', description='Date Format: `YYYY-MM-DD`')
            ),
            coreapi.Field(
                name='date_to',
                location='query',
                required=False,
                schema=coreschema.Integer(title='date_from', description='Date Format: `YYYY-MM-DD`')
            )
        ]

    def get_path_fields(self, path, method):
        field = coreapi.Field(
            name='user_id',
            location='path',
            required=True,
            schema=coreschema.Integer(title='user_id', description='A unique integer value identifying this user.')
        )
        return [field]


class TransactionViewSet(mixins.CreateModelMixin,
                         viewsets.GenericViewSet):
    """
    API endpoint to create transactions
    Can send one or more transactions by request.
    Query example:

    ```json
        [
          {
            "reference": "000053",
            "account": "C00099",
            "date": "2020-01-03",
            "amount": -51.13,
            "type": "outflow",
            "category": "groceries",
            "user_id": 1
          },
          {
            "reference": "000054",
            "account": "C00099",
            "date": "2020-01-10",
            "amount": 2500.60,
            "type": "inflow",
            "category": "groceries",
            "user_id": 1
          }
        ]
    ```

    """
    queryset = Transaction.objects.all()
    serializer_class = CreateTransactionSerializer
    permission_classes = []

    def create(self, request, *args, **kwargs):
        serializer = self.get_serializer(data=request.data, many=isinstance(request.data, list))
        serializer.is_valid(raise_exception=True)
        self.perform_create(serializer)
        return Response(serializer.data)


class AccountBalanceViewSet(viewsets.ViewSet, TransactionsMixin):
    """
    ### Retrieve summary by accounts of user.

    Optionally restricts the returned account balance to a given user,
    by filtering against a `date_from` and `date_to` query parameter in the URL.

    Date Format: `YYYY-MM-DD`

    Query example:

        /api/transactions/balance/account/1/?date_from=2020-01-10

    """
    permission_classes = []
    lookup_field = 'user_id'
    schema = CustomSchema()

    def retrieve(self, request, format=None, user_id=None, *args, **kwargs):
        try:
            self.user = User.objects.get(pk=user_id)
        except User.DoesNotExist:
            raise Http404
        transactions = self.get_filtered_transactions()
        with_annotations = transactions.values('account').annotate(balance=Sum('amount'),
                                                                   total_inflow=Sum(Case(
                                                                       When(type='inflow', then=F('amount')),
                                                                       output_field=FloatField(),
                                                                   )), total_outflow=Sum(Case(
                When(type='outflow', then=F('amount')),
                output_field=FloatField(),
            )), )
        serializer = BalanceByAccountSerializer(with_annotations, many=True)
        return Response(serializer.data)


class CashflowViewSet(viewsets.ViewSet, TransactionsMixin):
    """
    ### Retrieve summary by category

    Optionally restricts the returned cashflow to a given user,
    by filtering against a `date_from` and `date_to` query parameter in the URL.

    Date Format: `YYYY-MM-DD`

    Query example:

        /api/transactions/cashflow/1/?date_from=2020-01-10

    """
    schema = CustomSchema()
    permission_classes = []
    lookup_field = 'user_id'

    def retrieve(self, request, format=None, user_id=None, *args, **kwargs):
        try:
            self.user = User.objects.get(pk=user_id)
        except User.DoesNotExist:
            raise Http404
        transactions = self.get_filtered_transactions()
        inflows_qs = list(
            transactions.filter(type='inflow').values('category').annotate(amount=Sum('amount')).values('category',
                                                                                                        'amount'))
        outflow_qs = list(
            transactions.filter(type='outflow').values('category').annotate(amount=Sum('amount')).values('category',
                                                                                                         'amount'))
        inflow = {inf['category']: inf['amount'] for inf in inflows_qs}
        outflow = {outf['category']: outf['amount'] for outf in outflow_qs}
        return Response({'inflow': inflow, 'outflow': outflow})