格子快速入门 - 无法获取 link_token (Python)
Posted
技术标签:
【中文标题】格子快速入门 - 无法获取 link_token (Python)【英文标题】：Plaid Quickstart - Unable to fetch link_token (Python) 【发布时间】：2021-10-04 09:10:00 【问题描述】：按照https://dashboard.plaid.com/overview/sandbox 上的说明进行操作后。
我明白了
following message on the frontend 和来自后端的以下error on my terminal 尝试启动并运行格子沙箱时。如何摆脱前端的 Unable to fetch link_token 消息？我只能假设它与后端的端点 @app.route('/api/auth', methods=['GET']) 有关。
* Environment: production
WARNING: This is a development server. Do not use it in a production deployment.
Use a production WSGI server instead.
* Debug mode: off
* Running on http://127.0.0.1:8000/ (Press CTRL+C to quit)
127.0.0.1 - - [28/Jul/2021 10:58:55] "POST /api/info HTTP/1.1" 200 -
127.0.0.1 - - [28/Jul/2021 10:58:55] "GET /favicon.ico HTTP/1.1" 404 -
[2021-07-28 10:58:56,017] ERROR in app: Exception on /api/create_link_token [POST]
Traceback (most recent call last):
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/urllib3/connectionpool.py", line 699, in urlopen
httplib_response = self._make_request(
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/urllib3/connectionpool.py", line 382, in _make_request
self._validate_conn(conn)
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/urllib3/connectionpool.py", line 1010, in _validate_conn
conn.connect()
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/urllib3/connection.py", line 411, in connect
self.sock = ssl_wrap_socket(
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/urllib3/util/ssl_.py", line 449, in ssl_wrap_socket
ssl_sock = _ssl_wrap_socket_impl(
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/urllib3/util/ssl_.py", line 493, in _ssl_wrap_socket_impl
return ssl_context.wrap_socket(sock, server_hostname=server_hostname)
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/ssl.py", line 500, in wrap_socket
return self.sslsocket_class._create(
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/ssl.py", line 1040, in _create
self.do_handshake()
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/ssl.py", line 1309, in do_handshake
self._sslobj.do_handshake()
ssl.SSLCertVerificationError: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1129)
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/flask/app.py", line 2447, in wsgi_app
response = self.full_dispatch_request()
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/flask/app.py", line 1952, in full_dispatch_request
rv = self.handle_user_exception(e)
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/flask/app.py", line 1821, in handle_user_exception
reraise(exc_type, exc_value, tb)
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/flask/_compat.py", line 39, in reraise
raise value
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/flask/app.py", line 1950, in full_dispatch_request
rv = self.dispatch_request()
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/flask/app.py", line 1936, in dispatch_request
return self.view_functions[rule.endpoint](**req.view_args)
File "/Users/cruzgonzalez/quickstart/python/server.py", line 196, in create_link_token
response = client.link_token_create(request)
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/plaid/api_client.py", line 769, in __call__
return self.callable(self, *args, **kwargs)
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/plaid/api/plaid_api.py", line 5843, in __link_token_create
return self.call_with_http_info(**kwargs)
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/plaid/api_client.py", line 831, in call_with_http_info
return self.api_client.call_api(
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/plaid/api_client.py", line 406, in call_api
return self.__call_api(resource_path, method,
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/plaid/api_client.py", line 193, in __call_api
response_data = self.request(
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/plaid/api_client.py", line 452, in request
return self.rest_client.POST(url,
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/plaid/rest.py", line 264, in POST
return self.request("POST", url,
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/plaid/rest.py", line 150, in request
r = self.pool_manager.request(
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/urllib3/request.py", line 78, in request
return self.request_encode_body(
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/urllib3/request.py", line 170, in request_encode_body
return self.urlopen(method, url, **extra_kw)
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/urllib3/poolmanager.py", line 375, in urlopen
response = conn.urlopen(method, u.request_uri, **kw)
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/urllib3/connectionpool.py", line 783, in urlopen
return self.urlopen(
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/urllib3/connectionpool.py", line 783, in urlopen
return self.urlopen(
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/urllib3/connectionpool.py", line 783, in urlopen
return self.urlopen(
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/urllib3/connectionpool.py", line 755, in urlopen
retries = retries.increment(
File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/urllib3/util/retry.py", line 574, in increment
raise MaxRetryError(_pool, url, error or ResponseError(cause))
urllib3.exceptions.MaxRetryError: HTTPSConnectionPool(host='sandbox.plaid.com', port=443): Max retries exceeded with url: /link/token/create (Caused by SSLError(SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1129)')))
127.0.0.1 - - [28/Jul/2021 10:58:56] "POST /api/create_link_token HTTP/1.1" 500 -
这是我的 .env 文件
PLAID_CLIENT_ID='XXXXXXXXXXX'
PLAID_SECRET='XXXXXXXXXXXXX'
PLAID_ENV='sandbox'
PLAID_PRODUCTS=auth,transactions
PLAID_COUNTRY_CODES=US,CA
# Only required for OAuth:
# Set PLAID_REDIRECT_URI to 'http://localhost:3000'
# The OAuth redirect flow requires an endpoint on the developer's website
# that the bank website should redirect to. You will need to configure
# this redirect URI for your client ID through the Plaid developer dashboard
# at https://dashboard.plaid.com/team/api.
PLAID_REDIRECT_URI='http://localhost:3000'
还有我的python文件：server.py
# Read env vars from .env file
from plaid.model.payment_amount import PaymentAmount
from plaid.model.products import Products
from plaid.model.country_code import CountryCode
from plaid.model.nullable_recipient_bacs import NullableRecipientBACS
from plaid.model.payment_initiation_address import PaymentInitiationAddress
from plaid.model.payment_initiation_recipient_create_request import PaymentInitiationRecipientCreateRequest
from plaid.model.payment_initiation_payment_create_request import PaymentInitiationPaymentCreateRequest
from plaid.model.payment_initiation_payment_get_request import PaymentInitiationPaymentGetRequest
from plaid.model.link_token_create_request_payment_initiation import LinkTokenCreateRequestPaymentInitiation
from plaid.model.item_public_token_exchange_request import ItemPublicTokenExchangeRequest
from plaid.model.link_token_create_request import LinkTokenCreateRequest
from plaid.model.link_token_create_request_user import LinkTokenCreateRequestUser
from plaid.model.asset_report_create_request import AssetReportCreateRequest
from plaid.model.asset_report_create_request_options import AssetReportCreateRequestOptions
from plaid.model.asset_report_user import AssetReportUser
from plaid.model.asset_report_get_request import AssetReportGetRequest
from plaid.model.asset_report_pdf_get_request import AssetReportPDFGetRequest
from plaid.model.auth_get_request import AuthGetRequest
from plaid.model.transactions_get_request import TransactionsGetRequest
from plaid.model.transactions_get_request_options import TransactionsGetRequestOptions
from plaid.model.identity_get_request import IdentityGetRequest
from plaid.model.investments_transactions_get_request_options import InvestmentsTransactionsGetRequestOptions
from plaid.model.investments_transactions_get_request import InvestmentsTransactionsGetRequest
from plaid.model.accounts_balance_get_request import AccountsBalanceGetRequest
from plaid.model.accounts_get_request import AccountsGetRequest
from plaid.model.investments_holdings_get_request import InvestmentsHoldingsGetRequest
from plaid.model.item_get_request import ItemGetRequest
from plaid.model.institutions_get_by_id_request import InstitutionsGetByIdRequest
from plaid.api import plaid_api
from flask import Flask
from flask import render_template
from flask import request
from flask import jsonify
from datetime import datetime
from datetime import timedelta
import plaid
import base64
import os
import datetime
import json
import time
from dotenv import load_dotenv
load_dotenv()
app = Flask(__name__)
# Fill in your Plaid API keys - https://dashboard.plaid.com/account/keys
PLAID_CLIENT_ID = os.getenv('PLAID_CLIENT_ID')
PLAID_SECRET = os.getenv('PLAID_SECRET')
# 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
PLAID_ENV = os.getenv('PLAID_ENV', 'sandbox')
# 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.
PLAID_PRODUCTS = os.getenv('PLAID_PRODUCTS', 'transactions').split(',')
# PLAID_COUNTRY_CODES is a comma-separated list of countries for which users
# will be able to select institutions from.
PLAID_COUNTRY_CODES = os.getenv('PLAID_COUNTRY_CODES', 'US').split(',')
def empty_to_none(field):
value = os.getenv(field)
if value is None or len(value) == 0:
return None
return value
host = plaid.Environment.Sandbox
if PLAID_ENV == 'sandbox':
host = plaid.Environment.Sandbox
if PLAID_ENV == 'development':
host = plaid.Environment.Development
if PLAID_ENV == 'production':
host = plaid.Environment.Production
# Parameters used for the OAuth redirect Link flow.
#
# Set PLAID_REDIRECT_URI to 'http://localhost:3000/'
# The OAuth redirect flow requires an endpoint on the developer's website
# that the bank website should redirect to. You will need to configure
# this redirect URI for your client ID through the Plaid developer dashboard
# at https://dashboard.plaid.com/team/api.
PLAID_REDIRECT_URI = empty_to_none('PLAID_REDIRECT_URI')
configuration = plaid.Configuration(
host=host,
api_key=
'clientId': PLAID_CLIENT_ID,
'secret': PLAID_SECRET,
'plaidVersion': '2020-09-14'
)
api_client = plaid.ApiClient(configuration)
client = plaid_api.PlaidApi(api_client)
products = []
for product in PLAID_PRODUCTS:
products.append(Products(product))
# We store the access_token in memory - in production, store it in a secure
# persistent data store.
access_token = None
# The payment_id is only relevant for the UK Payment Initiation product.
# We store the payment_id in memory - in production, store it in a secure
# persistent data store.
payment_id = None
item_id = None
@app.route('/api/info', methods=['POST'])
def info():
global access_token
global item_id
return jsonify(
'item_id': item_id,
'access_token': access_token,
'products': PLAID_PRODUCTS
)
@app.route('/api/create_link_token_for_payment', methods=['POST'])
def create_link_token_for_payment():
global payment_id
try:
request = PaymentInitiationRecipientCreateRequest(
name='John Doe',
bacs=NullableRecipientBACS(account='26207729', sort_code='560029'),
address=PaymentInitiationAddress(
street=['street name 999'],
city='city',
postal_code='99999',
country='GB'
)
)
response = client.payment_initiation_recipient_create(
request)
recipient_id = response['recipient_id']
request = PaymentInitiationPaymentCreateRequest(
recipient_id=recipient_id,
reference='TestPayment',
amount=PaymentAmount(
currency='GBP',
value=100.00
)
)
response = client.payment_initiation_payment_create(
request
)
pretty_print_response(response.to_dict())
payment_id = response['payment_id']
request = LinkTokenCreateRequest(
products=[Products('payment_initiation')],
client_name='Plaid Test',
country_codes=list(map(lambda x: CountryCode(x), PLAID_COUNTRY_CODES)),
language='en',
user=LinkTokenCreateRequestUser(
client_user_id=str(time.time())
),
payment_initiation=LinkTokenCreateRequestPaymentInitiation(
payment_id=payment_id
)
)
response = client.link_token_create(request)
pretty_print_response(response.to_dict())
return jsonify(response.to_dict())
except plaid.ApiException as e:
return json.loads(e.body)
@app.route('/api/create_link_token', methods=['POST'])
def create_link_token():
try:
request = LinkTokenCreateRequest(
products=products,
client_name="Plaid Quickstart",
country_codes=list(map(lambda x: CountryCode(x), PLAID_COUNTRY_CODES)),
language='en',
user=LinkTokenCreateRequestUser(
client_user_id=str(time.time())
)
)
# create link token
response = client.link_token_create(request)
return jsonify(response.to_dict())
except plaid.ApiException as e:
return json.loads(e.body)
# Exchange token flow - exchange a Link public_token for
# an API access_token
# https://plaid.com/docs/#exchange-token-flow
@app.route('/api/set_access_token', methods=['POST'])
def get_access_token():
global access_token
global item_id
public_token = request.form['public_token']
try:
exchange_request = ItemPublicTokenExchangeRequest(
public_token=public_token)
exchange_response = client.item_public_token_exchange(exchange_request)
access_token = exchange_response['access_token']
item_id = exchange_response['item_id']
return jsonify(exchange_response.to_dict())
except plaid.ApiException as e:
return json.loads(e.body)
# Retrieve ACH or ETF account numbers for an Item
# https://plaid.com/docs/#auth
@app.route('/api/auth', methods=['GET'])
def get_auth():
try:
request = AuthGetRequest(
access_token=access_token
)
response = client.auth_get(request)
pretty_print_response(response.to_dict())
return jsonify(response.to_dict())
except plaid.ApiException as e:
error_response = format_error(e)
return jsonify(error_response)
# Retrieve Transactions for an Item
# https://plaid.com/docs/#transactions
@app.route('/api/transactions', methods=['GET'])
def get_transactions():
# Pull transactions for the last 30 days
start_date = (datetime.datetime.now() - timedelta(days=30))
end_date = datetime.datetime.now()
try:
options = TransactionsGetRequestOptions()
request = TransactionsGetRequest(
access_token=access_token,
start_date=start_date.date(),
end_date=end_date.date(),
options=options
)
response = client.transactions_get(request)
pretty_print_response(response.to_dict())
return jsonify(response.to_dict())
except plaid.ApiException as e:
error_response = format_error(e)
return jsonify(error_response)
# Retrieve Identity data for an Item
# https://plaid.com/docs/#identity
@app.route('/api/identity', methods=['GET'])
def get_identity():
try:
request = IdentityGetRequest(
access_token=access_token
)
response = client.identity_get(request)
pretty_print_response(response.to_dict())
return jsonify(
'error': None, 'identity': response.to_dict()['accounts'])
except plaid.ApiException as e:
error_response = format_error(e)
return jsonify(error_response)
# Retrieve real-time balance data for each of an Item's accounts
# https://plaid.com/docs/#balance
@app.route('/api/balance', methods=['GET'])
def get_balance():
try:
request = AccountsBalanceGetRequest(
access_token=access_token
)
response = client.accounts_balance_get(request)
pretty_print_response(response.to_dict())
return jsonify(response.to_dict())
except plaid.ApiException as e:
error_response = format_error(e)
return jsonify(error_response)
# Retrieve an Item's accounts
# https://plaid.com/docs/#accounts
@app.route('/api/accounts', methods=['GET'])
def get_accounts():
try:
request = AccountsGetRequest(
access_token=access_token
)
response = client.accounts_get(request)
pretty_print_response(response.to_dict())
return jsonify(response.to_dict())
except plaid.ApiException as e:
error_response = format_error(e)
return jsonify(error_response)
# Create and then retrieve an Asset Report for one or more Items. Note that an
# Asset Report can contain up to 100 items, but for simplicity we're only
# including one Item here.
# https://plaid.com/docs/#assets
@app.route('/api/assets', methods=['GET'])
def get_assets():
try:
request = AssetReportCreateRequest(
access_tokens=[access_token],
days_requested=60,
options=AssetReportCreateRequestOptions(
webhook='https://www.example.com',
client_report_id='123',
user=AssetReportUser(
client_user_id='789',
first_name='Jane',
middle_name='Leah',
last_name='Doe',
ssn='123-45-6789',
phone_number='(555) 123-4567',
email='jane.doe@example.com',
)
)
)
response = client.asset_report_create(request)
pretty_print_response(response.to_dict())
asset_report_token = response['asset_report_token']
except plaid.ApiException as e:
error_response = format_error(e)
return jsonify(error_response)
# Poll for the completion of the Asset Report.
num_retries_remaining = 20
asset_report_json = None
while num_retries_remaining > 0:
try:
request = AssetReportGetRequest(
asset_report_token=asset_report_token,
)
response = client.asset_report_get(request)
asset_report_json = response['report']
break
except plaid.ApiException as e:
response = json.loads(e.body)
if response['error_code'] == 'PRODUCT_NOT_READY':
num_retries_remaining -= 1
time.sleep(1)
continue
error_response = format_error(e)
return jsonify(error_response)
if asset_report_json is None:
return jsonify('error': 'status_code': e.status, 'display_message':
'Timed out when polling for Asset Report', 'error_code': '', 'error_type': '')
asset_report_pdf = None
try:
request = AssetReportPDFGetRequest(
asset_report_token=asset_report_token,
)
pdf = client.asset_report_pdf_get(request)
return jsonify(
'error': None,
'json': asset_report_json.to_dict(),
'pdf': base64.b64encode(pdf.read()).decode('utf-8'),
)
except plaid.ApiException as e:
error_response = format_error(e)
return jsonify(error_response)
# Retrieve investment holdings data for an Item
# https://plaid.com/docs/#investments
@app.route('/api/holdings', methods=['GET'])
def get_holdings():
try:
request = InvestmentsHoldingsGetRequest(access_token=access_token)
response = client.investments_holdings_get(request)
pretty_print_response(response.to_dict())
return jsonify('error': None, 'holdings': response.to_dict())
except plaid.ApiException as e:
error_response = format_error(e)
return jsonify(error_response)
# Retrieve Investment Transactions for an Item
# https://plaid.com/docs/#investments
@app.route('/api/investment_transactions', methods=['GET'])
def get_investment_transactions():
# Pull transactions for the last 30 days
start_date = (datetime.datetime.now() - timedelta(days=(30)))
end_date = datetime.datetime.now()
try:
options = InvestmentsTransactionsGetRequestOptions()
request = InvestmentsTransactionsGetRequest(
access_token=access_token,
start_date=start_date.date(),
end_date=end_date.date(),
options=options
)
response = client.investment_transactions_get(
request)
pretty_print_response(response.to_dict())
return jsonify(
'error': None, 'investment_transactions': response.to_dict())
except plaid.ApiException as e:
error_response = format_error(e)
return jsonify(error_response)
# This functionality is only relevant for the UK Payment Initiation product.
# Retrieve Payment for a specified Payment ID
@app.route('/api/payment', methods=['GET'])
def payment():
global payment_id
try:
request = PaymentInitiationPaymentGetRequest(payment_id=payment_id)
response = client.payment_initiation_payment_get(request)
pretty_print_response(response.to_dict())
return jsonify('error': None, 'payment': response.to_dict())
except plaid.ApiException as e:
error_response = format_error(e)
return jsonify(error_response)
# Retrieve high-level information about an Item
# https://plaid.com/docs/#retrieve-item
@app.route('/api/item', methods=['GET'])
def item():
try:
request = ItemGetRequest(access_token=access_token)
response = client.item_get(request)
request = InstitutionsGetByIdRequest(
institution_id=response['item']['institution_id'],
country_codes=list(map(lambda x: CountryCode(x), PLAID_COUNTRY_CODES))
)
institution_response = client.institutions_get_by_id(request)
pretty_print_response(response.to_dict())
pretty_print_response(institution_response.to_dict())
return jsonify('error': None, 'item': response.to_dict()[
'item'], 'institution': institution_response.to_dict()['institution'])
except plaid.ApiException as e:
error_response = format_error(e)
return jsonify(error_response)
def pretty_print_response(response):
print(json.dumps(response, indent=2, sort_keys=True, default=str))
def format_error(e):
response = json.loads(e.body)
return 'error': 'status_code': e.status, 'display_message':
response['error_message'], 'error_code': response['error_code'], 'error_type': response['error_type']
if __name__ == '__main__':
app.run(port=os.getenv('PORT', 8000))
【问题讨论】：
【参考方案1】：实际上，我们已经弄清楚为什么 python 3.9 不能与快速入门一起使用。您的 SSLCertVerificationError 已解决 here
在您的终端中运行它以安装根证书：
open /Applications/Python\ 3.9/Install\ Certificates.command
【讨论】：
【参考方案2】：这也可能是由过时的 requests 包引起的。我的团队遇到了同样的问题，将requests 升级到最新版本（当前为 2.27.1）解决了它。
【讨论】：
以上是关于格子快速入门 - 无法获取 link_token (Python)的主要内容，如果未能解决你的问题，请参考以下文章