qrtr-services/connection/connections/plaid_client_v2.py

from .abstract import AbstractConnectionClient
from django.conf import settings
import os
import datetime

import plaid
from plaid.api import plaid_api
from plaid.model.link_token_create_request import LinkTokenCreateRequest
from plaid.model.link_token_create_request_user import LinkTokenCreateRequestUser
from plaid.model.item_public_token_exchange_request import ItemPublicTokenExchangeRequest
from plaid.model.transactions_sync_request import TransactionsSyncRequest
from plaid.model.accounts_get_request import AccountsGetRequest
from plaid.model.products import Products
from plaid.model.country_code import CountryCode


def format_error(e):
    return {
        'error': {
            'display_message': e.display_message,
            'error_code': e.code,
            'error_type': e.type,
            'error_message': e.message}}


class Connection(AbstractConnectionClient):

    def __init__(self, credentials, account_id=None):
        self.credentials = credentials
        self.account_id = account_id

        # Fill in your Plaid API keys -
        # https://dashboard.plaid.com/account/keys
        self.PLAID_CLIENT_ID = settings.PLAID_CLIENT_ID
        self.PLAID_SECRET = settings.PLAID_SECRET
        self.PLAID_PUBLIC_KEY = settings.PLAID_PUBLIC_KEY
        # Use 'sandbox' to test with Plaid's Sandbox environment (usplaid-python==9.2.0ername: user_good,
        # password: pass_good)
        # Use `development` to test with live users and credentials and `production`
        # to go live
        self.PLAID_ENV = settings.PLAID_ENV
        # PLAID_PRODUCTS is a comma-separated list of products to use when initializing
        # Link. Note that this list must contain 'assets' in order for the app to be
        # able to create and retrieve asset reports.
        self.PLAID_PRODUCTS = settings.PLAID_PRODUCTS

        # PLAID_COUNTRY_CODES is a comma-separated list of countries for which users
        # will be able to select institutions from.
        self.PLAID_COUNTRY_CODES = settings.PLAID_COUNTRY_CODES

        configuration = plaid.Configuration(
            host=self.PLAID_ENV,
            api_key={
                'clientId': self.PLAID_CLIENT_ID,
                'secret': self.PLAID_SECRET,
            }
        )
        self.api_client = plaid.ApiClient(configuration)
        self.client = plaid_api.PlaidApi(self.api_client)

        # # Create a link_token for the given user
        # request = LinkTokenCreateRequest(
        #         products=[Products("auth")],
        #         client_name="Qrtr Plaid",
        #         country_codes=[CountryCode('US')],
        #         #redirect_uri='https://domainname.com/oauth-page.html',
        #         language='en',
        #         webhook='https://webhook.example.com',
        #         user=LinkTokenCreateRequestUser(
        #             client_user_id=self.account_id
        #         )
        #     )
        # response = client.link_token_create(request)
        # resp_dict = response.to_dict()
        # resp_dict['expiration'] = resp_dict['expiration'].strftime('%s')

        # self.credentials.update(resp_dict)
        # return self.credentials

    def generate_auth_request(self):
        # Create a link_token for the given user
        request = LinkTokenCreateRequest(
                products=[Products("auth")],
                client_name="Qrtr Plaid",
                country_codes=[CountryCode('US')],
                # redirect_uri='https://domainname.com/oauth-page.html',
                language='en',
                webhook='https://webhook.example.com',
                user=LinkTokenCreateRequestUser(
                    client_user_id=self.account_id
                )
            )
        response = self.client.link_token_create(request)
        resp_dict = response.to_dict()
        resp_dict['expiration'] = resp_dict['expiration'].strftime('%s')

        self.credentials.update(resp_dict)

    def get_auth_token(self, public_token):
        try:
            exchange_request = ItemPublicTokenExchangeRequest(
                public_token=public_token
            )
            exchange_response = self.client.item_public_token_exchange(
                exchange_request)
        except Exception as e:
            print("Error Occurred")
            print(e)
            return format_error(e)
        access_token = exchange_response['access_token']
        item_id = exchange_response['item_id']
        self.credentials.update({"access_token": access_token, "item_id": item_id})
        return {"access_token": access_token, "item_id": item_id}

    def get_accounts(self, auth_token=None):
        if not auth_token:
            auth_token = self.credentials.get('access_token')
        if not auth_token:
            raise Exception("Missing Auth Token")
        try:
            acc_request = AccountsGetRequest(access_token=auth_token)
            accounts = self.client.accounts_get(acc_request).to_dict()
        except Exception as e:
            print(e)
            accounts = None
        return accounts

    def get_transactions(
            self,
            start_date=None,
            end_date=None,
            auth_token=None):
        if not auth_token:
            auth_token = self.credentials.get('access_token')
        request = TransactionsSyncRequest(
            access_token=auth_token,
        )
        response = self.client.transactions_sync(request)
        transactions = response['added']

        # the transactions in the response are paginated, so make multiple calls while incrementing the cursor to
        # retrieve all transactions
        while (response['has_more']):
            request = TransactionsSyncRequest(
                access_token=auth_token,
                cursor=response['next_cursor']
            )
            response = self.client.transactions_sync(request)
            transactions += response['added']
        return transactions