ยินดีต้อนรับสู่โลกอันน่าตื่นเต้นของ FastAPI และ WebSockets! วันนี้ เราจะเจาะลึกถึงกลไกของ WebSockets โดยใช้ FastAPI ไม่ว่าคุณจะเป็นนักพัฒนาที่มีประสบการณ์หรือเพิ่งเริ่มต้น คู่มือนี้จะแนะนำคุณตลอดแนวคิดพื้นฐานและการใช้งานจริงของ WebSockets แต่ก่อนที่เราจะกระโดดเข้าไป อย่าลืมเครื่องมือที่มีประโยชน์ที่จะทำให้การพัฒนา API ของคุณเป็นเรื่องง่าย—Apidog คุณสามารถดาวน์โหลด Apidog ได้ฟรีและปรับปรุงการทดสอบและเอกสาร API ของคุณได้อย่างง่ายดาย
FastAPI คืออะไร?
FastAPI เป็นเว็บเฟรมเวิร์กที่ทันสมัย รวดเร็ว (ประสิทธิภาพสูง) สำหรับสร้าง API ด้วย Python 3.7+ โดยอิงตามคำแนะนำประเภท Python มาตรฐาน ช่วยให้คุณสร้าง API เว็บที่แข็งแกร่งและมีประสิทธิภาพได้อย่างรวดเร็ว หนึ่งในคุณสมบัติเด่นของ FastAPI คือความสามารถในการจัดการการเขียนโปรแกรมแบบอะซิงโครนัสและ WebSockets ได้อย่างง่ายดาย
WebSockets คืออะไร?
WebSockets เป็นโปรโตคอลการสื่อสารที่ให้ช่องทางการสื่อสารแบบฟูลดูเพล็กซ์ผ่านการเชื่อมต่อ TCP เดียวกัน ซึ่งแตกต่างจาก HTTP ซึ่งเป็นโปรโตคอลการร้องขอ-ตอบสนอง WebSockets ช่วยให้สามารถสื่อสารแบบสองทิศทางระหว่างไคลเอนต์และเซิร์ฟเวอร์ได้ สิ่งนี้ทำให้เหมาะสำหรับแอปพลิเคชันแบบเรียลไทม์ เช่น แอปพลิเคชันแชท การแจ้งเตือนสด และเกม
ทำไมต้องใช้ FastAPI กับ WebSockets?
การออกแบบของ FastAPI ทำให้ง่ายต่อการสร้าง API ที่รองรับ WebSockets ด้วย FastAPI คุณสามารถสร้าง API ที่ไม่เพียงแต่รวดเร็วและเชื่อถือได้เท่านั้น แต่ยังรองรับการสื่อสารแบบอะซิงโครนัสโดยใช้ WebSockets การผสมผสานนี้เหมาะสำหรับการพัฒนาแอปพลิเคชันแบบเรียลไทม์ที่ทันสมัย
เริ่มต้นใช้งาน FastAPI WebSockets
ในการเริ่มต้น คุณจะต้องติดตั้ง FastAPI และ Uvicorn ซึ่งเป็นเซิร์ฟเวอร์ ASGI สำหรับให้บริการแอปพลิเคชัน FastAPI ของคุณ
pip install fastapi uvicorn
เมื่อคุณติดตั้งแพ็คเกจที่จำเป็นแล้ว คุณสามารถเริ่มสร้างแอปพลิเคชัน FastAPI ของคุณด้วยการรองรับ WebSocket ได้
การสร้างแอปพลิเคชัน FastAPI WebSocket พื้นฐาน
มาเริ่มต้นด้วยตัวอย่างง่ายๆ เพื่อทำความเข้าใจพื้นฐานของการสื่อสาร WebSocket ใน FastAPI
from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
app = FastAPI()
html = """
<!DOCTYPE html>
<html>
<head>
<title>WebSocket Example</title>
</head>
<body>
<h1>WebSocket Example</h1>
<button onclick="connectWebSocket()">Connect</button>
<script>
function connectWebSocket() {
const ws = new WebSocket("ws://localhost:8000/ws");
ws.onmessage = function(event) {
const message = event.data;
alert("Message from server: " + message);
};
ws.onopen = function() {
ws.send("Hello Server");
};
}
</script>
</body>
</html>
"""
@app.get("/")
async def get():
return HTMLResponse(html)
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
data = await websocket.receive_text()
await websocket.send_text(f"Message text was: {data}")
ในตัวอย่างนี้ เรามีหน้า HTML ง่ายๆ ที่เชื่อมต่อกับเซิร์ฟเวอร์ WebSocket เมื่อคลิกปุ่ม เซิร์ฟเวอร์ได้รับข้อความ "Hello Server" และตอบกลับด้วย "Message text was: Hello Server"
การจัดการการเชื่อมต่อหลายรายการ
หนึ่งในจุดแข็งของ WebSockets คือการจัดการการเชื่อมต่อไคลเอนต์หลายรายการพร้อมกัน มาขยายตัวอย่างของเราเพื่อจัดการการเชื่อมต่อหลายรายการและออกอากาศข้อความไปยังไคลเอนต์ที่เชื่อมต่อทั้งหมด
from fastapi import FastAPI, WebSocket
from typing import List
app = FastAPI()
class ConnectionManager:
def __init__(self):
self.active_connections: List[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
manager = ConnectionManager()
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
await manager.broadcast(f"Client says: {data}")
except:
manager.disconnect(websocket)
ในตัวอย่างที่อัปเดตนี้ เราได้แนะนำคลาส ConnectionManager เพื่อจัดการการเชื่อมต่อ WebSocket หลายรายการ เมธอด broadcast จะส่งข้อความไปยังไคลเอนต์ที่เชื่อมต่อทั้งหมด ด้วยวิธีนี้ ไคลเอนต์ทุกรายจะได้รับข้อความใดๆ ที่ส่งโดยไคลเอนต์รายอื่นๆ
การส่งพารามิเตอร์ไปยังจุดสิ้นสุด WebSocket
บางครั้ง คุณอาจต้องส่งพารามิเตอร์ไปยังจุดสิ้นสุด WebSocket ของคุณ FastAPI ทำให้ง่ายต่อการทำเช่นนี้ นี่คือตัวอย่าง:
from fastapi import FastAPI, WebSocket
app = FastAPI()
@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
await websocket.accept()
await websocket.send_text(f"Hello, Client {client_id}")
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Message from {client_id}: {data}")
ในตัวอย่างนี้ เรากำหนดจุดสิ้นสุด WebSocket ที่ยอมรับพารามิเตอร์ client_id พารามิเตอร์นี้สามารถใช้เพื่อระบุไคลเอนต์และปรับแต่งข้อความ
การรวม WebSockets กับจุดสิ้นสุด API
FastAPI ช่วยให้คุณสามารถรวมการสื่อสาร WebSocket กับจุดสิ้นสุด API ที่มีอยู่ของคุณได้ ซึ่งมีประโยชน์เมื่อคุณต้องการรวมการสื่อสารแบบเรียลไทม์กับการร้องขอ API แบบดั้งเดิม
from fastapi import FastAPI, WebSocket, HTTPException
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id: int):
if item_id == 0:
raise HTTPException(status_code=404, detail="Item not found")
return {"item_id": item_id, "name": "Item name"}
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
await websocket.send_text("Connected to WebSocket")
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Received: {data}")
ในตัวอย่างนี้ เรามีจุดสิ้นสุด API แบบดั้งเดิม /items/{item_id} พร้อมกับจุดสิ้นสุด WebSocket /ws จุดสิ้นสุดทั้งสองสามารถใช้งานได้พร้อมกัน ทำให้คุณสามารถจัดการการสื่อสารแบบซิงโครนัสและอะซิงโครนัสในแอปพลิเคชันของคุณได้
ข้อควรพิจารณาด้านความปลอดภัย
เมื่อทำงานกับ WebSockets สิ่งสำคัญคือต้องพิจารณาด้านความปลอดภัย เช่น การตรวจสอบสิทธิ์และการอนุญาต FastAPI มีเครื่องมือในการรักษาความปลอดภัยการเชื่อมต่อ WebSocket ของคุณ
การรักษาความปลอดภัยการเชื่อมต่อ WebSocket
คุณสามารถใช้ระบบการฉีดพึ่งพาของ FastAPI เพื่อรักษาความปลอดภัยจุดสิ้นสุด WebSocket ของคุณ นี่คือตัวอย่างวิธีรักษาความปลอดภัยจุดสิ้นสุด WebSocket ด้วยการตรวจสอบสิทธิ์ตามโทเค็นอย่างง่าย
from fastapi import FastAPI, WebSocket, Depends, HTTPException, status
app = FastAPI()
def get_token_header(token: str = None):
if token != "mysecrettoken":
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Invalid token")
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket, token: str = Depends(get_token_header)):
await websocket.accept()
await websocket.send_text("Connected to WebSocket")
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Received: {data}")
ในตัวอย่างนี้ เราใช้ฟังก์ชัน Depends เพื่อฉีดการพึ่งพา get_token_header ซึ่งตรวจสอบความถูกต้องของโทเค็น หากโทเค็นไม่ถูกต้อง จะเกิดข้อผิดพลาด HTTP 403 Forbidden
การจัดการข้อมูลไบนารี
WebSockets ยังรองรับการส่งข้อมูลไบนารี ซึ่งมีประโยชน์สำหรับการส่งไฟล์หรือรูปภาพ FastAPI ทำให้ง่ายต่อการจัดการข้อมูลไบนารี
from fastapi import FastAPI, WebSocket
app = FastAPI()
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
await websocket.send_text("Connected to WebSocket")
while True:
data = await websocket.receive_bytes()
await websocket.send_bytes(data)
ในตัวอย่างนี้ จุดสิ้นสุด WebSocket จะรับข้อมูลไบนารีโดยใช้ receive_bytes และส่งกลับไปยังไคลเอนต์โดยใช้ send_bytes
แอปพลิเคชันในโลกแห่งความเป็นจริงของ FastAPI WebSockets
WebSockets มีความหลากหลายอย่างเหลือเชื่อและสามารถใช้ได้ในแอปพลิเคชันในโลกแห่งความเป็นจริงที่หลากหลาย นี่คือตัวอย่างบางส่วน:
แอปพลิเคชันแชทแบบเรียลไทม์
หนึ่งในกรณีการใช้งานทั่วไปที่สุดสำหรับ WebSockets คือการสร้างแอปพลิเคชันแชทแบบเรียลไทม์ ด้วย FastAPI คุณสามารถสร้างเซิร์ฟเวอร์แชทที่จัดการไคลเอนต์หลายรายและออกอากาศข้อความแบบเรียลไทม์ได้อย่างง่ายดาย
การแจ้งเตือนสด
WebSockets เหมาะสำหรับการส่งการแจ้งเตือนสด ตัวอย่างเช่น คุณสามารถใช้ WebSockets เพื่อแจ้งให้ผู้ใช้ทราบถึงข้อความใหม่ การอัปเดต หรือการแจ้งเตือนแบบเรียลไทม์
เกมออนไลน์
WebSockets ถูกนำมาใช้อย่างแพร่หลายในเกมออนไลน์เพื่อเปิดใช้งานการสื่อสารแบบเรียลไทม์ระหว่างผู้เล่น การรองรับการเขียนโปรแกรมแบบอะซิงโครนัสของ FastAPI ทำให้เป็นตัวเลือกที่ดีสำหรับการสร้างเซิร์ฟเวอร์เกม
คุณสามารถใช้ Apidog เพื่อดีบักไคลเอนต์ WebSocket ได้อย่างไร
การจัดการ API และข้อมูลแบบเรียลไทม์อาจเป็นเรื่องท้าทาย นั่นคือที่มาของ Apidog Apidog ช่วยลดความซับซ้อนในการพัฒนา ทดสอบ และจัดการ API ทำให้ง่ายต่อการจัดการทั้งการร้องขอ HTTP ด้วย FastAPI และการอัปเดตแบบเรียลไทม์ด้วย WebSockets
- เปิด Apidog: ขั้นแรก ให้เริ่มแอปพลิเคชัน Apidog แล้วคลิกที่ปุ่ม "+" ทางด้านซ้าย หน้าต่างแบบเลื่อนลงใหม่จะเปิดขึ้น จากนั้นเลือก "New WebSocket API":
2. สร้างการเชื่อมต่อ: เริ่มต้นด้วยการป้อน URL API WebSocket ในแถบที่อยู่ของ Apidog จากนั้นคุณสามารถคลิกปุ่ม "เชื่อมต่อ" เพื่อเริ่มกระบวนการแฮนด์เชคและสร้างการเชื่อมต่อ Apidog ช่วยให้คุณปรับแต่งพารามิเตอร์ เช่น Params, Headers และ Cookies ในระหว่างการแฮนด์เชค
3. ส่งและรับข้อความ: เมื่อสร้างการเชื่อมต่อแล้ว คุณสามารถส่งข้อความภายใต้แท็บ "ข้อความ" คุณสามารถเขียนข้อความ รูปแบบ JSON, XML, HTML และข้อความรูปแบบอื่นๆ รวมถึงข้อความรูปแบบไบนารีโดยใช้ Base64 หรือ Hexadecimal มุมมองไทม์ไลน์ใหม่ของ Apidog แสดงสถานะการเชื่อมต่อ ข้อความที่ส่งและข้อความที่ได้รับตามลำดับเวลา เมื่อคุณคลิกที่ข้อความ คุณสามารถดูรายละเอียดได้อย่างง่ายดาย
4. เอกสาร API: Apidog สืบทอดฟังก์ชันการทำงานของเอกสาร API ที่แข็งแกร่งสำหรับ API WebSocket ทำให้คุณสามารถจัดทำเอกสารการโต้ตอบ WebSocket ของคุณได้อย่างมีประสิทธิภาพ
อย่าลืมตรวจสอบปัญหา WebSocket ทั่วไป เช่น ปัญหาการเชื่อมต่อ ข้อผิดพลาดรูปแบบข้อความ และข้อกังวลด้านความปลอดภัยในระหว่างกระบวนการดีบักของคุณ
การใช้ Apidog เพื่อส่งคำขอ HTTP
Apidog มีคุณสมบัติขั้นสูงหลายอย่างที่ช่วยเพิ่มความสามารถในการทดสอบคำขอ HTTP คุณสมบัติเหล่านี้ช่วยให้คุณปรับแต่งคำขอของคุณและจัดการสถานการณ์ที่ซับซ้อนมากขึ้นได้อย่างง่ายดาย
ขั้นตอนที่ 1: เปิด Apidog และสร้าง คำขอ ใหม่
ขั้นตอนที่ 2: ค้นหาหรือป้อนรายละเอียด API ด้วยตนเองสำหรับคำขอ POST ที่คุณต้องการสร้าง
ขั้นตอนที่ 3: กรอกพารามิเตอร์ที่จำเป็นและข้อมูลใดๆ ที่คุณต้องการรวมไว้ในเนื้อหาคำขอ
บทสรุป
FastAPI และ WebSockets เป็นการผสมผสานที่ทรงพลังสำหรับการสร้างแอปพลิเคชันแบบเรียลไทม์ที่ทันสมัย ไม่ว่าคุณจะสร้างแอปพลิเคชันแชท ระบบการแจ้งเตือนสด หรือเกมออนไลน์ FastAPI มีเครื่องมือที่คุณต้องการในการจัดการการสื่อสาร WebSocket อย่างมีประสิทธิภาพ
อย่าลืมดาวน์โหลด Apidog ได้ฟรีเพื่อปรับปรุงกระบวนการพัฒนา API ของคุณ Apidog ทำให้ง่ายต่อการทดสอบและจัดทำเอกสาร API ของคุณ ช่วยประหยัดเวลาและความพยายาม
