wechatpayv3 Python SDK

The SDK uses a single unified callback method for all notification types. To handle specific business logic, inspect the event_type field in the returned result.

Example logic for a successful transaction:

1. Call wxpay.callback(headers, body).
2. Check if result.get('event_type') == 'TRANSACTION.SUCCESS'.
3. Extract details from the resource object (e.g., appid, mchid, out_trade_no, transaction_id, amount).
4. Important: After processing, your endpoint must return a 200 or 204 status code to WeChat Pay to acknowledge receipt.

@app.route('/notify', methods=['POST'])
def notify():
    result = wxpay.callback(request.headers, request.data)
    if result and result.get('event_type') == 'TRANSACTION.SUCCESS':
        resource = result.get('resource')
        # ... extract fields like appid, mchid, amount ...
        # TODO: Perform business logic
        return jsonify({'code': 'SUCCESS', 'message': '成功'}) # Returns 200
    else:
        return jsonify({'code': 'FAILED', 'message': '失败'}), 500

The asynchronous client must be managed using the async with syntax to ensure proper connection lifecycle management and resource cleanup. All API method calls must be prefixed with await.

import asyncio
from wechatpayv3.async_ import AsyncWeChatPay, WeChatPayType

async def main():
    with open('/path/to/apiclient_key.pem') as f:
        private_key = f.read()
    
    async with AsyncWeChatPay(
        wechatpay_type=WeChatPayType.NATIVE,
        mchid='1230000109',
        private_key=private_key,
        cert_serial_no='444F4864EA9B34415...',
        apiv3_key='MIIEvwIBADANBgkqhkiG9w0BAQE...',
        appid='wxd678efh567hg6787',
        notify_url='https://www.xxxx.com/notify',
        cert_dir='./cert'
    ) as wxpay:
        # Example: Native payment
        code, message = await wxpay.pay(
            description='Async test',
            out_trade_no='async_demo_001',
            amount={'total': 100},
            pay_type=WeChatPayType.NATIVE
        )
        print(f"Code: {code}, Message: {message}")

asyncio.run(main())

When using FastAPI, it is recommended to initialize the AsyncWeChatPay client within a lifespan context manager. This allows the client to start and clean up resources (like connection pools) when the server starts and stops.

from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from wechatpayv3.async_ import AsyncWeChatPay, WeChatPayType

wxpay = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global wxpay
    with open('/path/to/apiclient_key.pem') as f:
        private_key = f.read()
    
    wxpay = AsyncWeChatPay(
        wechatpay_type=WeChatPayType.NATIVE,
        mchid='1230000109',
        private_key=private_key,
        cert_serial_no='444F4864EA9B34415...',
        apiv3_key='MIIEvwIBADANBgkqhkiG9w0BAQE...',
        appid='wxd678efh567hg6787',
        notify_url='https://www.xxxx.com/notify',
        cert_dir='./cert'
    )
    await wxpay.__aenter__()
    yield
    if wxpay:
        await wxpay.__aexit__(None, None, None)

app = FastAPI(lifespan=lifespan)

class PaymentRequest(BaseModel):
    description: str
    out_trade_no: str
    total: int

@app.post("/pay")
async def create_payment(payment: PaymentRequest):
    code, message = await wxpay.pay(
        description=payment.description,
        out_trade_no=payment.out_trade_no,
        amount={'total': payment.total},
        pay_type=WeChatPayType.NATIVE
    )
    if code == 200:
        return {"code": code, "message": message}
    else:
        raise HTTPException(status_code=code, detail=message)

In Tornado, you can either create a new AsyncWeChatPay instance per request using async with or manage a global instance. The following example demonstrates the per-request pattern.

import tornado.web
import tornado.ioloop
import json
from wechatpayv3.async_ import AsyncWeChatPay, WeChatPayType

# Configuration (simplified for example)
WECHATPAY_CONFIG = {
    'wechatpay_type': WeChatPayType.NATIVE,
    'mchid': '1230000109',
    'private_key': '...', 
    'cert_serial_no': '...',
    'apiv3_key': '...',
    'appid': '...',
    'notify_url': '...',
    'cert_dir': './cert'
}

class PaymentHandler(tornado.web.RequestHandler):
    async def post(self):
        async with AsyncWeChatPay(**WECHATPAY_CONFIG) as wxpay:
            data = json.loads(self.request.body)
            code, message = await wxpay.pay(
                description=data['description'],
                out_trade_no=data['out_trade_no'],
                amount={'total': data['total']},
                pay_type=WeChatPayType.NATIVE
            )
            self.write({"code": code, "message": message})

class QueryHandler(tornado.web.RequestHandler):
    async def get(self, out_trade_no):
        async with AsyncWeChatPay(**WECHATPAY_CONFIG) as wxpay:
            code, message = await wxpay.query(out_trade_no=out_trade_no)
            self.write({"code": code, "message": message})

def make_app():
    return tornado.web.Application([
        (r"/pay", PaymentHandler),
        (r"/query/([^/]+)", QueryHandler),
    ])

if __name__ == "__main__":
    app = make_app()
    app.listen(8888)
    tornado.ioloop.IOLoop.current().start()