flask_plaid_link_token_and_public_token_exchange.py

1import os
2import datetime
3from flask import Flask, request, jsonify
4import plaid
5from plaid.model.products import Products
6from plaid.model.country_code import CountryCode
7from plaid.model.item_public_token_exchange_request import ItemPublicTokenExchangeRequest
8from plaid.model.link_token_create_request import LinkTokenCreateRequest
9from plaid.model.link_token_create_request_user import LinkTokenCreateRequestUser
10from plaid.api import plaid_api
11
12app = Flask(__name__)
13
14# Configuration for Plaid client
15PLAID_CLIENT_ID = os.getenv('PLAID_CLIENT_ID')
16PLAID_SECRET = os.getenv('PLAID_SECRET')
17PLAID_ENV = os.getenv('PLAID_ENV', 'sandbox')
18
19host = plaid.Environment.Sandbox
20if PLAID_ENV == 'development':
21    host = plaid.Environment.Development
22elif PLAID_ENV == 'production':
23    host = plaid.Environment.Production
24
25configuration = plaid.Configuration(
26    host=host,
27    api_key={
28        'clientId': PLAID_CLIENT_ID,
29        'secret': PLAID_SECRET,
30    }
31)
32
33api_client = plaid.ApiClient(configuration)
34client = plaid_api.PlaidApi(api_client)
35
36# Create a link_token for the given user
37@app.route('/api/create_link_token', methods=['POST'])
38def create_link_token():
39    try:
40        request = LinkTokenCreateRequest(
41            products=[Products('auth'), Products('transactions')],
42            client_name="Plaid Quickstart",
43            country_codes=[CountryCode('US')],
44            language='en',
45            user=LinkTokenCreateRequestUser(
46                client_user_id='unique-user-id'
47            )
48        )
49        response = client.link_token_create(request)
50        return jsonify(response.to_dict())
51    except plaid.ApiException as e:
52        return jsonify({'error': e.body})
53
54# Exchange a public_token for an access_token
55@app.route('/api/set_access_token', methods=['POST'])
56def get_access_token():
57    public_token = request.json.get('public_token')
58    try:
59        exchange_request = ItemPublicTokenExchangeRequest(
60            public_token=public_token
61        )
62        exchange_response = client.item_public_token_exchange(exchange_request)
63        access_token = exchange_response['access_token']
64        item_id = exchange_response['item_id']
65        return jsonify({'access_token': access_token, 'item_id': item_id})
66    except plaid.ApiException as e:
67        return jsonify({'error': e.body})
68
69if __name__ == '__main__':
70    app.run(port=8000)