要用支付宝的沙箱环境测试。
首先肯定要安装flask和alipay-sdk-python：
pip install flask alipay-sdk-python
简单地写一个alipay model：
class Base:
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
def save(self):
db.session.add(self)
db.session.commit()
class Alipay(db.Model, Base):
id = db.Column(db.String(32), primary_key=True)
amount = db.Column(db.Integer, nullable=False)
status = db.Column(db.Boolean, default=False)
写views函数：
first = Blueprint('first_blue', __name__)
@first.route('/alipay', methods=['GET', 'POST'])
def alipay():
if request.method == 'GET':
return render_template('alipay.html') # 可以简单写一个能够传输amount(金额)的html表单
elif request.method == 'POST':
amount = request.form.get('amount') # 从alipay.html获取到的amount
alipay_client_config = AlipayClientConfig() # 初始化支付宝客户端配置
alipay_client_config.server_url = 'https://openapi.alipaydev.com/gateway.do' # 支付宝开发者服务器
alipay_client_config.app_id = APPID # 支付宝开发者appid
alipay_client_config.app_private_key = APP_PRIVATE_KEY # 支付宝app私钥
alipay_client_config.alipay_public_key = ALIPAY_PUBLIC_KEY # 支付宝公钥
client = DefaultAlipayClient(alipay_client_config) # 用上面的配置生成一个客户端实例
now_raw = datetime.datetime.now() # 这个时间是模拟的支付宝订单id
now = datetime.datetime.strftime(now_raw, '%Y%m%d%H%M%S%f') # 微秒级
model = AlipayTradePagePayModel() # 初始化一个订单模型
model.out_trade_no = now # 商户订单号（这个flask所有者用的订单号）
model.total_amount = amount # 订单总价
model.subject = "支付宝测试" # 订单标题
model.body = "多商品订单-支付宝测试" # 订单详细内容
model.product_code = "FAST_INSTANT_TRADE_PAY" # 销售产品码 https://opensupport.alipay.com/support/knowledge/20582/201602373525?ant_source=zsearch
model.timeout_express = '20m' # 支付宝订单不支付的超时时间
req = AlipayTradePagePayRequest(biz_model=model) # 用上面的订单model生成一个订单支付请求req
req.return_url = SERVER_HOST + 'paid/' # GET方法，支付完成后跳转到这个页面进行验签并修改订单状态
req.notify_url = SERVER_HOST + 'paid/' # POST方法是支付宝服务器通知flask服务器支付状态，即时快速不可篡改。notify_url只有在可访问的服务器上有效，本地无法测试。notify_url生效后，不仅会调用下面的paid的POST方法，完成后还会调用GET方法返回。
# 得到构造的请求，如果http_method是GET，则是一个带完成请求参数的url，如果http_method是POST，则是一段HTML表单片段
response = client.page_execute(req, http_method="GET") # GET方法会生成一个支付链接
return jsonify({'status': 200, 'response': response, 'order_id': now})
@first.route('/paid/')
def paid():
if request.method == 'GET':
args = request.args.to_dict() # request.args转换为dict才能修改
status = alipay_verify_sign(args) # 对支付宝反回的url里的参数进行验签，验签函数在下面
if not status:
return Response('你在开玩笑吗？你真的支付了吗？')
order_id = args['out_trade_no'] # 因为POST后还会访问GET，为防止重复生一个订单引发主键重复错误，需要判断一下是否已存在订单id
order = Alipay.query.get(order_id)
if order:
return Response('已支付')
ali = Alipay()
ali.id = args['out_trade_no']
ali.amount = args['total_amount']
ali.status = True
try:
ali.save()
except Exception as e:
return Response(f'支付失败，原因：\n{e}')
return Response('已支付')
elif request.method == 'POST':
args = request.form.to_dict() # request.form转换为dict才能修改
status = alipay_verify_sign(args)
if not status:
return Response('failure')
ali = Alipay()
ali.id = args['out_trade_no']
ali.amount = args['total_amount']
ali.status = True
ali.save()
return Response('success') # 支付宝文档要求服务器支付成功后要返回一个纯字符串'success'，但我并没有发现有什么卵用
@first.route('/verifypaid')
def verify_paid(): # 验证支付状态
order_id = request.args.get('order_id', '')
if not order_id:
return Response('你在开玩笑吗？你真的支付了吗？')
order = Alipay.query.get(order_id)
if not order:
return Response('你在开玩笑吗？你真的支付了吗？')
if not order.status:
return Response('你在开玩笑吗？你真的支付了吗？')
return Response(f'订单id：{order_id}，订单状态：已支付')
支付宝验签函数：
def alipay_verify_sign(args):
"""
支付宝验签
"""
sign = args.pop('sign', None) # 删除签名
args.pop('sign_type') # 删除签名类型
params = sorted(args.items(), key=lambda e: e[0], reverse=False) # 取出字典元素按key的字母升序排序形成列表
message = '&'.join(u"{}={}".format(k, v) for k, v in params).encode() # 将列表转为二进制参数字符串
try:
status = verify_with_rsa(ALIPAY_PUBLIC_KEY.encode('utf-8').decode('utf-8'), message, sign) # 通过签名和支付宝公钥验证消息主体的有效性
except: # 无效会抛异常，捕获异常后返回false
status = False
return status
上面就是支付宝支付的主体代码了，如果没有互联网服务器进行测试，其中的notify_url不能用，需要注释掉。
网友评论