flask_plaid_link_token_creation_and_access_token_exchange.py

1import os
2import datetime
3import plaid
4from plaid.model.products import Products
5from plaid.model.country_code import CountryCode
6from plaid.model.item_public_token_exchange_request import ItemPublicTokenExchangeRequest
7from plaid.model.link_token_create_request import LinkTokenCreateRequest
8from plaid.model.link_token_create_request_user import LinkTokenCreateRequestUser
9from plaid.api import plaid_api
10from flask import Flask, jsonify, request
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# Global variable to store access_token for demonstration purposes
37access_token = None
38
39@app.route('/api/create_link_token', methods=['POST'])
40def create_link_token():
41    try:
42        request = LinkTokenCreateRequest(
43            products=[Products('transactions')],
44            client_name="Plaid Quickstart",
45            country_codes=[CountryCode('US')],
46            language='en',
47            user=LinkTokenCreateRequestUser(client_user_id='unique-user-id')
48        )
49        # create link token
50        response = client.link_token_create(request)
51        return jsonify(response.to_dict())
52    except plaid.ApiException as e:
53        return jsonify({'error': e.body})
54
55@app.route('/api/set_access_token', methods=['POST'])
56def get_access_token():
57    global access_token
58    public_token = request.json.get('public_token')
59    try:
60        exchange_request = ItemPublicTokenExchangeRequest(
61            public_token=public_token
62        )
63        exchange_response = client.item_public_token_exchange(exchange_request)
64        access_token = exchange_response['access_token']
65        return jsonify(exchange_response.to_dict())
66    except plaid.ApiException as e:
67        return jsonify({'error': e.body})
68
69if __name__ == '__main__':
70    app.run(port=8000)