Python websockets庫是用於在Python中構建WebSocket伺服器和用戶端的庫。
如果可能，應該使用最新版本的Python。如果使用的是舊版本，請注意，對於每個次要版本(3.x)，僅官方支援最新的錯誤修復版本(3.x.y)。
為了獲得最佳體驗，應該從Python≥3.6以上版本。asyncio在Python 3.4和3.6之間做了很大的改進。
注意：本文件是為
Python≥3.6編寫的。
$ pip install websockets
下面是一個WebSocket伺服器範例。它從用戶端讀取名稱，傳送問候語，然後關閉連線。參考以下實現程式碼 -
#!/usr/bin/env python
# WS server example
import asyncio
import websockets
async def hello(websocket, path):
name = await websocket.recv()
print(f"< {name}")
greeting = f"Hello {name}!"
await websocket.send(greeting)
print(f"> {greeting}")
start_server = websockets.serve(hello, 'localhost', 8765)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()
在伺服器端，websockets為每個WebSocket連線執行一次處理協同程式hello。它在處理協同程式返回時關閉連線。
下面是一個相應的WebSocket用戶端範例。
#!/usr/bin/env python
# WS client example
import asyncio
import websockets
async def hello():
async with websockets.connect('ws://localhost:8765') as websocket:
name = input("What's your name? ")
await websocket.send(name)
print(f"> {name}")
greeting = await websocket.recv()
print(f"< {greeting}")
asyncio.get_event_loop().run_until_complete(hello())
使用connect()作為非同步上下文管理器可確保在退出hello協同程式之前關閉連線。
安全的WebSocket連線可以提高機密性和可靠性，因為它們可以降低不良代理干擾的風險。WSS協定是WS到HTTPS的HTTP：連線是用TLS加密的。WSS需要像HTTPS這樣的TLS證書。以下是使用Python≥3.6中提供的API來調整伺服器範例以提供安全連線的方法。
請參閱ssl模組的文件以安全地組態上下文或將程式碼調整為較舊的Python版本。
#!/usr/bin/env python
# WSS (WS over TLS) server example, with a self-signed certificate
import asyncio
import pathlib
import ssl
import websockets
async def hello(websocket, path):
name = await websocket.recv()
print(f"< {name}")
greeting = f"Hello {name}!"
await websocket.send(greeting)
print(f"> {greeting}")
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
ssl_context.load_cert_chain(
pathlib.Path(__file__).with_name('localhost.pem'))
start_server = websockets.serve(
hello, 'localhost', 8765, ssl=ssl_context)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()
以下是調整用戶端程式碼，也適用於Python≥3.6。
#!/usr/bin/env python
# WSS (WS over TLS) client example, with a self-signed certificate
import asyncio
import pathlib
import ssl
import websockets
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ssl_context.load_verify_locations(
pathlib.Path(__file__).with_name('localhost.pem'))
async def hello():
async with websockets.connect(
'wss://localhost:8765', ssl=ssl_context) as websocket:
name = input("What's your name? ")
await websocket.send(name)
print(f"> {name}")
greeting = await websocket.recv()
print(f"< {greeting}")
asyncio.get_event_loop().run_until_complete(hello())
此用戶端需要上下文，因為伺服器使用自簽名證書。
連線到具有有效證書的安全WebSocket伺服器的用戶端(即由您的Python安裝信任的CA簽名)可以簡單地將ssl = True傳遞給connect()而不是構建上下文。
這是一個如何執行WebSocket伺服器並從瀏覽器連線的範例。
在控制台中執行此指令碼：
#!/usr/bin/env python
# WS server that sends messages at random intervals
import asyncio
import datetime
import random
import websockets
async def time(websocket, path):
while True:
now = datetime.datetime.utcnow().isoformat() + 'Z'
await websocket.send(now)
await asyncio.sleep(random.random() * 3)
start_server = websockets.serve(time, '127.0.0.1', 5678)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()
然後在瀏覽器中開啟此HTML檔案。
<!DOCTYPE html>
<html>
<head>
<title>WebSocket demo</title>
</head>
<body>
<script>
var ws = new WebSocket("ws://127.0.0.1:5678/"),
messages = document.createElement('ul');
ws.onmessage = function (event) {
var messages = document.getElementsByTagName('ul')[0],
message = document.createElement('li'),
content = document.createTextNode(event.data);
message.appendChild(content);
messages.appendChild(message);
};
document.body.appendChild(messages);
</script>
</body>
</html>
WebSocket伺服器可以從用戶端接收事件，處理它們以更新應用程式狀態，並在用戶端之間同步結果狀態。
這是一個範例，任何用戶端都可以遞增或遞減計數器。更新將傳播到所有連線的用戶端。
asyncio的並行模型保證更新被序列化。
在控制台中執行此指令碼：
#!/usr/bin/env python
# WS server example that synchronizes state across clients
import asyncio
import json
import logging
import websockets
logging.basicConfig()
STATE = {'value': 0}
USERS = set()
def state_event():
return json.dumps({'type': 'state', **STATE})
def users_event():
return json.dumps({'type': 'users', 'count': len(USERS)})
async def notify_state():
if USERS: # asyncio.wait doesn't accept an empty list
message = state_event()
await asyncio.wait([user.send(message) for user in USERS])
async def notify_users():
if USERS: # asyncio.wait doesn't accept an empty list
message = users_event()
await asyncio.wait([user.send(message) for user in USERS])
async def register(websocket):
USERS.add(websocket)
await notify_users()
async def unregister(websocket):
USERS.remove(websocket)
await notify_users()
async def counter(websocket, path):
# register(websocket) sends user_event() to websocket
await register(websocket)
try:
await websocket.send(state_event())
async for message in websocket:
data = json.loads(message)
if data['action'] == 'minus':
STATE['value'] -= 1
await notify_state()
elif data['action'] == 'plus':
STATE['value'] += 1
await notify_state()
else:
logging.error(
"unsupported event: {}", data)
finally:
await unregister(websocket)
asyncio.get_event_loop().run_until_complete(
websockets.serve(counter, 'localhost', 6789))
asyncio.get_event_loop().run_forever()
然後在幾個瀏覽器中開啟此HTML檔案。
<!DOCTYPE html>
<html>
<head>
<title>WebSocket demo</title>
<style type="text/css">
body {
font-family: "Courier New", sans-serif;
text-align: center;
}
.buttons {
font-size: 4em;
display: flex;
justify-content: center;
}
.button, .value {
line-height: 1;
padding: 2rem;
margin: 2rem;
border: medium solid;
min-height: 1em;
min-width: 1em;
}
.button {
cursor: pointer;
user-select: none;
}
.minus {
color: red;
}
.plus {
color: green;
}
.value {
min-width: 2em;
}
.state {
font-size: 2em;
}
</style>
</head>
<body>
<div class="buttons">
<div class="minus button">-</div>
<div class="value">?</div>
<div class="plus button">+</div>
</div>
<div class="state">
<span class="users">?</span> online
</div>
<script>
var minus = document.querySelector('.minus'),
plus = document.querySelector('.plus'),
value = document.querySelector('.value'),
users = document.querySelector('.users'),
websocket = new WebSocket("ws://127.0.0.1:6789/");
minus.onclick = function (event) {
websocket.send(JSON.stringify({action: 'minus'}));
}
plus.onclick = function (event) {
websocket.send(JSON.stringify({action: 'plus'}));
}
websocket.onmessage = function (event) {
data = JSON.parse(event.data);
switch (data.type) {
case 'state':
value.textContent = data.value;
break;
case 'users':
users.textContent = (
data.count.toString() + " user" +
(data.count == 1 ? "" : "s"));
break;
default:
console.error(
"unsupported event", data);
}
};
</script>
</body>
</html>