解决ssl.SSLCertVerificationError: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed

问题

原代码：

async def call_wss_api(msg):
    async with websockets.connect('wss://xxx.com/tool/handle') as websocket:
        await websocket.send(msg)
        response = ""
        count = 0
        while websocket.open:
            response = await websocket.recv()
    return response

执行代码时，发现会报错：ssl.SSLCertVerificationError: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:997)

Traceback (most recent call last):
  File "/Users/aaa/xxx.py", line 354, in create_order_by_athena
    res = asyncio.get_event_loop().run_until_complete(call_wss_api(msg))
  File "/Library/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/base_events.py", line 646, in run_until_complete
    return future.result()
  File "/Users/aaa/xxx.py", line 376, in call_wss_api
    async with websockets.connect('wss://xxx.com/tool/handle') as websocket:
  File "/Users/aaa/xxx/venv/lib/python3.10/site-packages/websockets/legacy/client.py", line 633, in __aenter__
    return await self
  File "/Users/aaa/xxx/venv/lib/python3.10/site-packages/websockets/legacy/client.py", line 650, in __await_impl_timeout__
    return await asyncio.wait_for(self.__await_impl__(), self.open_timeout)
  File "/Library/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/tasks.py", line 445, in wait_for
    return fut.result()
  File "/Users/aaa/xxx/venv/lib/python3.10/site-packages/websockets/legacy/client.py", line 654, in __await_impl__
    transport, protocol = await self._create_connection()
  File "/Library/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/base_events.py", line 1089, in create_connection
    transport, protocol = await self._create_connection_transport(
  File "/Library/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/base_events.py", line 1119, in _create_connection_transport
    await waiter
  File "/Library/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/sslproto.py", line 534, in data_received
    ssldata, appdata = self._sslpipe.feed_ssldata(data)
  File "/Library/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/sslproto.py", line 188, in feed_ssldata
    self._sslobj.do_handshake()
  File "/Library/Frameworks/Python.framework/Versions/3.10/lib/python3.10/ssl.py", line 974, in do_handshake
    self._sslobj.do_handshake()
ssl.SSLCertVerificationError: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:997)

进程已结束，退出代码为 1

解决方法

需要在调用函数websockets.connect()时，传递一个关键字参数ssl=ssl_context即可，代码如下：

import ssl
import certifi

ssl_context = ssl.create_default_context()
ssl_context.load_verify_locations(certifi.where())

async def call_wss_api(msg):
    async with websockets.connect('wss://xxx.com/tool/handle', ssl=ssl_context) as websocket:
        await websocket.send(msg)
        response = ""
        count = 0
        while websocket.open:
            response = await websocket.recv()
    return response

参考文档：https://www.cnpython.com/qa/260804