در منظره مدرن معماری نرمافزار، تقاضا برای ارتباطات بلادرنگ از یک ویژگی «خوب است داشته باشیم» به یک نیاز اساسی تبدیل شده است. چه داشبوردهای معاملات زنده سهام، چه ابزارهای ویرایش مشارکتی، و چه برنامههای بازی با فرکانس بالا، سیستمها اکنون باید بلافاصله به تغییرات وضعیت واکنش نشان دهند. روشهای سنتی polling در HTTP برای این سناریوها ناکارآمد هستند و تأخیر غیرضروری و بار اضافی بر سرور تحمیل میکنند. اینجاست که FastAPI، در ترکیب با پروتکلهای WebSocket، به عنوان انتخاب برتر برای ساخت APIهای رویدادمحور و مستحکم در پایتون درخشش میکند.
قدرت FastAPI در چارچوب ناهمگام (asynchronous) آن نهفته است که بر پایه Starlette و Pydantic ساخته شده است. اگرچه این فریمورک به دلیل نقاط پایانی REST با عملکرد بالا از طریق HTTP شهرت دارد، اما قدرت واقعی آن در فضای رویدادمحور از طریق WebSockets آزاد میشود. با بهرهگیری از اکوسیستم `asyncio` پایتون، میتوانیم اتصالات پایدار و دوطرفهای ایجاد کنیم که به سرور اجازه میدهد به محض وقوع یک رویداد، دادهها را به سمت کلاینتها ارسال کند و نیاز به پرسوجوی مکرر کلاینت برای دریافت بروزرسانیها را از بین میبرد.
درک معماری: اولویت با ناهمگامسازی
پایه و اساس هر برنامه بلادرنگ، توانایی آن در مدیریت اتصالات همزمان بدون مسدود کردن نخ اصلی است. در دنیای همگام (synchronous)، یک اتصال WebSocket با عمر طولانی میتواند در صورت مدیریت دقیق، کل سرور را متوقف کند. FastAPI این مسئله را به صورت بومی حل میکند. با استفاده از کلمات کلیدی `async` و `await`، سرور میتواند هزاران اتصال WebSocket همزمان را روی یک نخ تکمخاطبسازی (multiplex) کند و منتظر رویدادهای ورودی/خروجی (مانند پیامهای ورودی یا پرسوجوهای پایگاه داده) بماند بدون اینکه اجرای برنامه را متوقف کند.
هنگام طراحی یک API رویدادمحور، جریان کار معمولاً شامل سه جزء است: کلاینت، بروکر/سرور و بوس رویداد (Event Bus). در یک پیکربندی خالص FastAPI، ما اغلب از یک دیکشنری رویداد مشترک در حافظه یا یک الگوی تولیدکننده-مصرفکننده با استفاده از صفهای `asyncio` برای مدیریت وضعیت در سراسر کلاینتهای متصل استفاده میکنیم.
پیادهسازی نقطه پایانی WebSocket
برای شروع، یک مسیر WebSocket را با استفاده از دکوراتور `@app.websocket` تعریف میکنیم. برخلاف مسیرهای استاندارد REST، WebSockets یک دستتکانی (handshake) را مدیریت میکنند، یک اتصال پایدار را برقرار میکنند و سپس وارد حلقهای میشوند که در آن سرور میتواند به صورت نامحدود پیامها را هم بخواند و هم بنویسد.
در زیر یک مثال عملی از یک نقطه پایانی WebSocket ساده که پیامها را به کلاینت بازتاب میدهد، آورده شده است. این مثال پایهای برای درک چرخه حیات اتصال فراهم میکند.
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
app = FastAPI()
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
try:
while True:
data = await websocket.receive_text()
# پیام را پردازش کنید یا یک رویداد را اینجا راه بیندازید
await websocket.send_text(f"متن پیام این بود: {data}")
except WebSocketDisconnect:
print("کلاینت قطع شد")مقیاسپذیری برای چندین کلاینت: الگوی پخش (Broadcast)
مثال بازتابی که در بالا دیدیم، ساده است. قدرت واقعی APIهای رویدادمحور زمانی ظهور میکند که بخواهیم یک رویداد واحد را به تمام کلاینتهای متصل پخش کنیم. برای دستیابی به این هدف، باید یک ثبت (registry) از اتصالات فعال WebSocket را نگهداری کنیم. در یک محیط تولید، این اغلب یک کانال pub/sub Redis خواهد بود، اما برای یادگیری در سطح متوسط، یک لیست در حافظه کاملاً مناسب است.
بیایید برنامه خود را ارتقا دهیم تا اعلانها را از یک کلاینت بپذیرد و آنها را به همه کلاینتهای دیگر پخش کند.
from typing import List
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
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"در حال پخش: {data}")
except WebSocketDisconnect:
manager.disconnect(websocket)
await manager.broadcast("یک کلاینت قطع شده است")یکپارچهسازی رویدادهای خارجی و وظایف پسزمینه
در یک معماری واقعی رویدادمحور، رویدادها اغلب از لایه WebSocket خود سرچشمه نمیگیرند. این میتواند یک تغییر در پایگاه داده، یک پیام از یک API خارجی یا یک وظیفه زمانبندی شده باشد. FastAPI به ما اجازه میدهد تا وظایف پسزمینه را با استفاده از `BackgroundTasks` یا الگوی `asyncio.create_task` اجرا کنیم تا به دنبال این رویدادهای خارجی گوش دهند و آنها را از طریق اتصالات WebSocket خود ارسال کنیم.
برای مثال، میتوانید یک سرویس پسزمینه ایجاد کنید که هر دقیقه به API آبوهوای یک طرف ثالث دسترسی پیدا کند و دادههای جدید را به تمام کلاینتهای داشبورد متصل پخش کند. این کار تولید رویداد را از تحویل به کلاینت جدا میکند و به اصل تفکیک نگرشها (separation of concerns) پایبند است.
نتیجهگیری
ساخت APIهای بلادرنگ و رویدادمحور با FastAPI و WebSockets، راهحلی قدرتمند و مقیاسپذیر برای برنامههای وب مدرن فراهم میکند. با بهرهگیری از قابلیتهای ناهمگام پایتون، توسعهدهندگان میتوانند سیستمهایی ایجاد کنند که پاسخگو، کارآمد و قادر به مدیریت بارهای کاری با همزمانی بالا هستند. چه در حال ساخت یک برنامه چت، یک سیستم اعلان زنده، یا یک داشبورد نظارتی پیچیده باشید، ترکیب عملکرد FastAPI و اتصال پایدار WebSockets، پایه فنی ایدهآلی را ارائه میدهد. هنگامی که به پیشرفت در پروژههای خود ادامه میدهید، به مقیاسپذیری توجه کنید؛ در حالی که لیستهای در حافظه برای توسعه کار میکنند، گذار به Redis برای مدیریت وضعیت برای استقرارهای سطح تولید حیاتی خواهد بود.