146 lines
6.1 KiB
Python
146 lines
6.1 KiB
Python
from django.shortcuts import render
|
|
from rest_framework import status, viewsets
|
|
from rest_framework.response import Response
|
|
from .models import Connection, ConnectionType
|
|
from .serializers import ConnectionSerializer
|
|
from rest_framework.decorators import action, authentication_classes
|
|
from rest_framework.decorators import permission_classes
|
|
from qrtr_account.models import Account
|
|
from rest_framework.permissions import IsAuthenticated, AllowAny
|
|
import importlib
|
|
import json
|
|
from .serializers import ConnectionSerializer, ConnectionTypeSerializer
|
|
from django.db import transaction
|
|
from qrtr_account.mixins import OwnedAccountsMixin
|
|
|
|
# Create your views here.
|
|
|
|
|
|
class ConnectionTypeViewSet(viewsets.ModelViewSet):
|
|
queryset = ConnectionType.objects.all()
|
|
serializer_class = ConnectionTypeSerializer
|
|
|
|
|
|
class ConnectionViewSet(viewsets.ModelViewSet, OwnedAccountsMixin):
|
|
"""API endpoint that allows connections to be seen or created
|
|
"""
|
|
permission_classes = [IsAuthenticated]
|
|
queryset = Connection.objects.all()
|
|
serializer_class = ConnectionSerializer
|
|
# Make connections somewhat immutable from the users perspective
|
|
http_method_names = [
|
|
'get',
|
|
'post',
|
|
'delete',
|
|
'options']
|
|
|
|
def get_queryset(self):
|
|
return Connection.objects.filter(
|
|
account__in=self.accessible_accounts().values_list('id'))
|
|
|
|
@action(detail=False, methods=['post'], url_path='plaid/exchange_public_token')
|
|
def exchange_public_token(self, request):
|
|
print(f"REQUEST: {request.data}")
|
|
name = request.data.get("name", "dummyName")
|
|
account_id = request.data.get("account")
|
|
public_token = request.data.get("public_token")
|
|
user = request.user
|
|
|
|
if request.user.is_anonymous:
|
|
accounts = (Account.objects.filter(pk=1))
|
|
else:
|
|
accounts = (Account.objects.filter(pk=account_id, owner=user) |
|
|
Account.objects.filter(pk=account_id,
|
|
admin_users__in=[user]))
|
|
if not accounts:
|
|
return Response(
|
|
status=status.HTTP_400_BAD_REQUEST,
|
|
data="ERROR: Account ID not found")
|
|
else:
|
|
print(f"Account Found: {accounts[0]}")
|
|
account = accounts[0]
|
|
print(request)
|
|
plaid_conn = importlib.import_module(f"connection.connections.plaid_client_v2")
|
|
conn_type = ConnectionType.objects.get(name="Plaid")
|
|
try:
|
|
plaid_client = plaid_conn.Connection(request.data.dict(), account_id=account_id)
|
|
token = plaid_client.get_auth_token(public_token)
|
|
except ValueError:
|
|
return Response(status=status.HTTP_503,
|
|
data="ERROR: Invalid public_token")
|
|
with transaction.atomic():
|
|
conn, created = Connection.objects \
|
|
.get_or_create(name=name, type=conn_type,
|
|
defaults={
|
|
"credentials": request.data,
|
|
"account": account
|
|
})
|
|
conn.credentials = plaid_client.credentials
|
|
print(f"CREDS: {plaid_client.credentials}")
|
|
conn.save()
|
|
return Response(plaid_client.get_accounts())
|
|
|
|
@action(detail=False, methods=['post'], url_path='plaid')
|
|
def authenticate(self, request):
|
|
print(request.data)
|
|
print(request.data.keys())
|
|
# public_token = request.data.get("public_token")
|
|
name = request.data.get("name", "dummyName")
|
|
account_id = request.data.get("account")
|
|
print(f"Account ID Detected: {account_id}")
|
|
# if public_token is None:
|
|
# return Response(
|
|
# status=status.HTTP_400_BAD_REQUEST,
|
|
# data="ERROR: missing public_token")
|
|
if account_id is None:
|
|
return Response(
|
|
status=status.HTTP_400_BAD_REQUEST,
|
|
data="ERROR: missing account_id")
|
|
user = request.user
|
|
# Filter out any accounts with the right id, but the given user
|
|
# is not an owner or admin on that account.
|
|
if request.user.is_anonymous:
|
|
accounts = (Account.objects.filter(pk=1))
|
|
else:
|
|
accounts = (Account.objects.filter(pk=account_id, owner=user) |
|
|
Account.objects.filter(pk=account_id,
|
|
admin_users__in=[user]))
|
|
if not accounts:
|
|
return Response(
|
|
status=status.HTTP_400_BAD_REQUEST,
|
|
data="ERROR: Account ID not found")
|
|
else:
|
|
print(f"Account Found: {accounts[0]}")
|
|
account = accounts[0]
|
|
print(request)
|
|
plaid_conn = importlib.import_module(f"connection.connections.plaid_client_v2")
|
|
conn_type = ConnectionType.objects.get(name="Plaid")
|
|
try:
|
|
plaid_client = plaid_conn.Connection(request.data.dict(), account_id=account_id)
|
|
plaid_client.generate_auth_request()
|
|
except ValueError:
|
|
return Response(status=status.HTTP_503,
|
|
data="ERROR: Invalid public_token")
|
|
except Exception as e:
|
|
print(e)
|
|
return Response(status=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
data="ERROR: Unable to contact Plaid")
|
|
with transaction.atomic():
|
|
conn, created = Connection.objects \
|
|
.get_or_create(name=name, type=conn_type,
|
|
defaults={
|
|
"credentials": request.data,
|
|
"account": account
|
|
})
|
|
conn.credentials = plaid_client.credentials
|
|
print(f"CREDS: {plaid_client.credentials}")
|
|
conn.save()
|
|
return Response(plaid_client.credentials)
|
|
|
|
@action(detail=False, methods=['post'], url_path='plaid-webhook',
|
|
permission_classes=[AllowAny])
|
|
def plaid_webhook(self, request):
|
|
print("Plaid Webhook Received!")
|
|
print(request.data)
|
|
return Response(200)
|