135 lines
5.2 KiB
Python
Executable File
135 lines
5.2 KiB
Python
Executable File
from .abstract import AbstractConnectionClient
|
|
from django.conf import settings
|
|
import plaid
|
|
from plaid.api import plaid_api
|
|
from plaid.model.item_public_token_exchange_request import ItemPublicTokenExchangeRequest
|
|
from plaid.model.transactions_get_request import TransactionsGetRequest
|
|
from plaid.model.accounts_get_request import AccountsGetRequest
|
|
import os
|
|
import datetime
|
|
|
|
|
|
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):
|
|
print("Plaid Connection Creation Initiated")
|
|
print(credentials)
|
|
self.credentials = credentials.dict()
|
|
|
|
# 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 (username: 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,
|
|
}
|
|
)
|
|
api_client = plaid.ApiClient(configuration)
|
|
self.client = plaid_api.PlaidApi(api_client)
|
|
|
|
# self.client = plaid.Client(
|
|
# client_id=self.PLAID_CLIENT_ID,
|
|
# secret=self.PLAID_SECRET,
|
|
# environment=self.PLAID_ENV,
|
|
# public_key=self.PLAID_PUBLIC_KEY,
|
|
# # api_version='2019-05-29',
|
|
# api_version='2020-09-14',
|
|
# #webhook='https://qrtr-services.herokuapp.com/connection/plaid-webhook/'
|
|
# )
|
|
public_key = self.credentials.get('public_token')
|
|
auth_token = self.credentials.get('auth_token')
|
|
if not auth_token and public_key:
|
|
print("Getting Auth Token From Public Key")
|
|
authorization = self.get_auth_token(public_key)
|
|
print("AUTHORIZATION:")
|
|
print(authorization)
|
|
auth_token = authorization.get("access_token")
|
|
item_id = authorization.get("item_id")
|
|
if "error" in auth_token:
|
|
raise ValueError(f"Unable to generate Auth Token, {auth_token}")
|
|
self.credentials['auth_token'] = auth_token
|
|
self.credentials['item_id'] = item_id
|
|
print("Plaid Connection successful")
|
|
print(self.credentials)
|
|
|
|
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']
|
|
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('auth_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('auth_token')
|
|
if not auth_token:
|
|
raise Exception("Missing Auth Token")
|
|
if not start_date:
|
|
start_date = '{:%Y-%m-%d}'.format(
|
|
datetime.datetime.now() + datetime.timedelta(-30))
|
|
if not end_date:
|
|
end_date = '{:%Y-%m-%d}'.format(datetime.datetime.now())
|
|
try:
|
|
transactions_req = TransactionsGetRequest(
|
|
access_token=auth_token,
|
|
start_date=start_date,
|
|
end_date=end_date
|
|
)
|
|
transactions_resp = self.client.transactions_get(
|
|
transactions_req)
|
|
except plaid.errors.PlaidError as e:
|
|
return format_error(e)
|
|
return transactions_resp.get("transactions")
|