django多种支付、并发订单处理实例代码

 ''' #思路 我们希望，通过插拔的方式来实现多方式登录，比如新增一种支付方式，那么只要在项目中新增一个py文件，导入里面的pay方法就可以了，这样在支付业务中支付语句是不发生变化的。 所以就可以使用python的鸭子类型及面向对象的反射方法来实现功能 ''' ##新建一个Pay文件夹，里面放支付方式.py文件 #Alipay.py class Alipay: def pay(self): pass #Visapay.py class Visapay: def pay(self): pass #Wxpay.py（完全按照接口文档来得） import time #记得导入商户号和key哦！ from app01.wx import settings class Wxpay: def pay(self,order_data): self.order_id = order_data["order_id"] self.open_id = order_data['open_id'] self.ip = order_data['ip'] data_body = self.get_body_data() import requests url = "https://api.mch.weixin.qq.com/pay/unifiedorder" response = requests.post(url, data_body.encode("utf-8"), headers={'content-type': "application/xml"}) res_dict = self.xml_to_dic(response.content) timeStamp = str(int(time.time())) paySign = self.get_pay_sign(res_dict, timeStamp) data_dic = { 'timeStamp': timeStamp, 'nonceStr': res_dict['nonce_str'], 'package': f"prepay_id={res_dict['prepay_id']}", 'signType': 'MD5', "paySign": paySign, } return data_dic def get_pay_sign(self, res_dict, timeStamp): print("res_dict", res_dict) data_dic = { 'appId': res_dict['appid'], 'timeStamp': timeStamp, 'nonceStr': res_dict['nonce_str'], 'package': f"prepay_id={res_dict['prepay_id']}", "signType": "MD5" } sign_str = "&".join([f"{k}={data_dic[k]}" for k in sorted(data_dic)]) sign_str = f"{sign_str}&key={settings.pay_apikey}" import hashlib md5 = hashlib.md5() md5.update(sign_str.encode("utf-8")) sign = md5.hexdigest() return sign.upper() def xml_to_dic(self, xml_data): import xml.etree.ElementTree as ET ''' xml to dict :param xml_data: :return: ''' xml_dict = {} root = ET.fromstring(xml_data) for child in root: xml_dict[child.tag] = child.text return xml_dict def get_random(self): import random data = "123456789zxcvbnmasdfghjklqwertyuiopZXCVBNMASDFGHJKLQWERTYUIOP" nonce_str = "".join(random.sample(data, 30)) return nonce_str def get_sign(self): data_dic = { "nonce_str": self.nonce_str, "out_trade_no": self.out_trade_no, "spbill_create_ip": self.spbill_create_ip, "notify_url": self.notify_url, "openid": self.open_id, "body": self.body, "trade_type": "JSAPI", "appid": self.appid, "total_fee": "1", "mch_id": self.mch_id } sign_str = "&".join([f"{k}={data_dic[k]}" for k in sorted(data_dic)]) sign_str = f"{sign_str}&key={settings.pay_apikey}" import hashlib md5 = hashlib.md5() md5.update(sign_str.encode("utf-8")) sign = md5.hexdigest() return sign.upper() def get_body_data(self): self.appid = settings.AppId # openid=self.open_id self.mch_id = str(settings.pay_mchid) self.nonce_str = self.get_random() self.out_trade_no = self.order_id self.spbill_create_ip = self.ip self.notify_url = "https://www.test.com" self.body = "老男孩学费" self.sign = self.get_sign() body_data = f"""  {self.appid}{self.mch_id}{self.nonce_str}{self.sign}{self.body}{self.out_trade_no}1{ self.spbill_create_ip}{self.notify_url}{self.open_id}JSAPI""" return body_data ##调用支付方法的语句（一般支付都是发生在订单创建好之后） import importlib from rest_framework.response import Response pay_method = "Wxpay" #这里是举例子，所以把pay_method写死了，正常情况下，应该是外面传来的支付方式，然后用pay_method接收 try: #用字符串导入支付方式的py文件,例如这里的app01.Pay.{pay_method} pay_field = importlib.import_module(f"app01.Pay.{pay_method}") #用反射拿到该文件下面的类 pay_method_class = getattr(pay_field, pay_method) except: return Response({"code": 205, "msg": "错误的支付方式"}) order_data['ip'] = host_ip order_data['open_id'] = open_id #完成支付，并把支付数据返回 pay_data = pay_method_class().pay(order_data) ''' 这里直接用反射拿到的支付类，然后使用它的pay方法，完成支付 ''' return Response({"code": 200, "msg": "ok", "data": pay_data})

 ''' 几个注意点： 1.a,b两个用户同时买一个库存为1的商品，这样为了保证数据安全（即a买了，库存没更新，b又买了，这样就存在安全问题）,需要在数据库操作时加锁，有两个选择：悲观锁和乐观锁。 悲观锁：冲突比较多的时候，使用悲观锁。悲观锁获取数据时对数据行了锁定，其他事务要想获取锁，必须等原事务结束。 乐观锁：冲突比较少的时候，使用乐观锁。查询时不锁数据，提交更改时进行判断.使用乐观锁前，要先 设置mysql事务的隔离级别transaction-isolation = READ-COMMITTED 2.a用户的订单有两种商品，这样提交订单后，假如第一种成功，第二种失败，很显然在订单一个函数里面写一个逻辑是行不通的，应该要么同时成功，要么同时失败。于是自然而然就用到了事务。 ''' #urls.py from django.urls import path from app01.view import order urlpattern = [ path('order/create', order.Create.as_view()), ] #common文件夹下的func.py文件 import time,random def get_order_id(): str_all="1242356796734534" return time.strftime("%Y%m%d%H%M%S")+"".join(random.sample(str_all,5)) #view文件夹下的order.py文件 from rest_framework.views import APIView from rest_framework.response import Response from django.db import transaction from django import forms from django.core.cache import cache from common import func class OrderForm(): phone = forms.CharField( error_message = { 'required': "手机号不能为空" }, # 调用Form组件中的验证器来校验手机号 # validators=[RegexValidator(r'1[1-9][0-9]{9}', '手机号格式不正确')], ) token = forms.CharField( error_messages={ "required": "token不能为空" }) province=forms.CharField( error_messages={ "required": "省份不能为空" }) city = forms.CharField(error_messages={ "required": "城市不能为空" }) county = forms.CharField(error_messages={ "required": "县/区不能为空" }) address = forms.CharField(error_messages={ "required": "详细地址不能为空" }) name = forms.CharField(error_messages={ "required": "姓名不能为空" }) class Create(APIView): @transaction.atomic def post(self, requset): param = request.data #form表单检验订单数据是否符合规范 for_obj = OrderForm(param) #校验成功，并且买东西了 if for_obj.is_valid() and param.get('buy_list'): buy_list = param.get("buy_list") #固定方法拿到付款用户的id if request.META.get("HTTP_X_FORWARDED_FOR"): host_ip = request.META["HTTP_X_FROWARDED_FOR"] else: host_ip = request.META["REMOTE_ADDR"] #校验token，保证登入状态 cache_data = cache.get(param["token"]) if not cache_data: return Response({"code": 202, "msg": "错误的token"}) openid = cache_data.split("&")[0] #通过openid查找用户 user_data = models.Wxuser.objects.filter(openid=openid).first() #组织部分总订单数据 order_data = {"consignee_mobile": param['phone'], 'consignee_name': param['name'], 'wxuser_id': user_data.id, "memo": param['remark'], "consignee_area": f"{param['province']},{param['city']},{param['county']}", "consignee_address": param['address'], } order_data['order_id']=func.get_order_id() order_data["order_total"]=0 #获取用户购买商品的id all_product_id=list(buy_list.keys()) #通过id来获取商品的信息 product_data=models.Product.objects.filter(product_id__in=all_product_id).all() #开启事务 sid = transaction.savepoint() for product in product_data: product.product_id=str(product.product_id) # num=buy_list[product.id] #获取商品的库存 for i in range(3): product_stock = product.stock.quantity new_product_stock = product_stock-buy_list[product.product_id] if new_product_stock<0: transaction.savepoint_rollback(sid) return response({"code": 204, "msg": f"{product.name}库存不足"}) #乐观锁 res if not : #如果两次都不成功，就让用户重新下单 i==2: 205, "下单失败从新下单"}) else: continue break #组织子订单数据 order_item_data={'order_id': order_data['order_id'], 'product_id': product.product_id,"name": product.name, "image": product.image, "price": product.price, "nums": buy_list[product.product_id], "brief": product.brief} #创建数据 models.order_items.objects.create(**order_item_data) #获取订单总金额 order_data["order_total"] +=buy_list[product.product_id]*product.price #创建总订单 models.order.objects.create(**order_data) transaction.savepoint_commit(sid) #提交延时任务，判断订单再指定时间内，有没有支付，如果没有支付，回滚库存，取消订单 func.check_order(order_data['order_id']) #如果pay_methon是外面传的 pay_methon try: #导入app01.pay.{pay_methon} pay_filed=importlib.import_module(f"app01.Pay.{pay_methon}") #用反射获取，这个文件中的类 pay_methon_class=getattr(pay_filed,pay_methon) except: "错误的支付方式"}) order_data['ip']=host_ip order_data['open_id']=openid #获取小程序所需的支付数据 pay_data=pay_methon_class().pay(order_data) # wxpay().pay elif:... 200, "ok","data":pay_data}) 203, "缺少参数"}) class notify(apiview): def post(self,request,paymethon): importlib.import_module(f"app01.pay.{paymethon}") getattr(pay_filed, paymethon) data=pay_methon_class().notify(request.data) data['stauts']=="suc": models.order.objects.filter(order_id=data['order_id']).update(pay_status="")

 #proj_celery文件夹下的celery.py文件 import celery import time # broker='redis://127.0.0.1:6379/2' 不加密码 backend = 'redis://127.0.0.1:6379/1' broker = 'redis://127.0.0.1:6379/2' cel = celery.Celery('test', backend=backend, broker=broker) import os, sys import django BASE_DIR = os.path.dirname(os.path.dirname(__file__)) # 定位到你的django根目录 # sys.path.append(os.path.join(BASE_DIR, "app01")) sys.path.append(os.path.abspath(BASE_DIR)) os.environ.setdefault("DJANGO_SETTINGS_MODULE", "shop.settings") django.setup() from django.db import transaction @cel.task @transaction.atomic def del_order(order_id): from app01 import models # 查看订单数据 order_data = models.Order.objects.filter(order_id=order_id, pay_status=False).first() # 如果有数据表示没有支付，要进行库存回滚，和取消订单 if order_data: # 获取该订单下的所有子订单 order_items = models.Order_items.objects.filter(order_id=order_id).all() # 将子订单中的数据转变成 {商品id:购买数量，。。。}的格式 product_all_dic = {item.product_id: item.nums for item in order_items} # 获取所有商品的id，成为list格式 product_all_id = list(product_all_dic.keys()) # 获取所有的商品 all_product = models.Product.objects.filter(product_id__in=product_all_id).all() sid = transaction.savepoint() # 把对应的商品进行库存回滚 for product in all_product: for i in range(3): stock = product.stock.quantity new_stock = stock + product_all_dic[product.product_id] #乐观锁 res = models.Stock.objects.filter(stock_id=product.stock.stock_id, quantity=stock).update( quantity=new_stock) if not res: if i == 2: transaction.savepoint_rollback(sid) # 如果这个执行失败了，那我们要从新提交任务，不然库存无法回滚 from app01.common import func func.check_order(order_id, 1) return else: continue else: break # 修改订单状态 res1 = models.Order.objects.filter(order_id=order_id, pay_status=False).update(status="dead") if res1: transaction.savepoint_commit(sid) else: transaction.savepoint_rollback(sid)