집 >백엔드 개발 >파이썬 튜토리얼 >FastAPI에서 푸시 알림을 사용하여 실시간으로 데이터를 업데이트하는 방법

FastAPI에서 푸시 알림을 사용하여 실시간으로 데이터를 업데이트하는 방법

王林원래의: 2023-07-29 18:09:141898검색

소개:
인터넷의 지속적인 발전으로 실시간 데이터 업데이트가 점점 더 중요해지고 있습니다. 예를 들어 실시간 거래, 실시간 모니터링, 실시간 게임과 같은 애플리케이션 시나리오에서는 가장 정확한 정보와 최상의 사용자 경험을 제공하기 위해 적시에 데이터를 업데이트해야 합니다. FastAPI는 고성능 웹 애플리케이션을 구축하는 간단하고 효율적인 방법을 제공하는 최신 Python 기반 웹 프레임워크입니다. 이 기사에서는 FastAPI를 사용하여 실시간으로 데이터를 업데이트하는 푸시 알림을 구현하는 방법을 소개합니다.

1단계: 준비
먼저 FastAPI와 해당 종속 라이브러리를 설치해야 합니다. 다음 명령을 사용하여 설치할 수 있습니다:

pip install fastapi
pip install uvicorn

다음으로 Python 파일을 만들고 이름을 main.py로 지정해야 합니다. 여기에 FastAPI 애플리케이션을 작성하겠습니다. main.py。我们将在其中编写我们的FastAPI应用程序。

步骤二：编写推送通知逻辑
我们将使用WebSocket来实现推送通知的功能。WebSocket是一种在客户端和服务器之间实现全双工通信的协议。在FastAPI中，可以使用第三方库fastapi-websocket来轻松地为我们的应用程序添加WebSocket支持。可以使用以下命令来安装该库：

pip install fastapi-websocket

在main.py中，我们首先导入所需的模块：

from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from fastapi import WebSocket, WebSocketDisconnect

from fastapi_websocket_pubsub import PubSubEndpoint
from fastapi_websocket_pubsub import PubSubWebSocketEndpoint

然后，我们创建一个FastAPI应用程序：

app = FastAPI()
app.mount("/static", StaticFiles(directory="static"), name="static")

templates = Jinja2Templates(directory="templates")

接下来，我们需要定义WebSocket的逻辑。我们将创建一个WebSocket的子类，并在其中实现推送通知的功能：

class NotificationsWebSocket(WebSocketEndpoint):
async def on_receive(self, websocket: WebSocket, data: str):
    # 在这里实现推送通知的逻辑
    # 可以根据需要订阅特定的主题或频道

async def on_connect(self, websocket: WebSocket):
    await websocket.accept()

async def on_disconnect(self, websocket: WebSocket, close_code: int):
    await self.pubsub.unsubscribe(self.room_name, websocket)

app.add_websocket_route("/ws", NotificationsWebSocket)

在on_receive方法中，我们可以根据需要订阅特定的主题或频道，并通过WebSocket将新数据发送给客户端。在FastAPI应用程序中，我们可以使用PubSubEndpoint来管理订阅和发布的逻辑。可以使用以下命令来安装fastapi_websocket_pubsub库：

pip install fastapi_websocket_pubsub

在main.py中，我们导入PubSubEndpoint并创建一个实例：

from fastapi_websocket_pubsub import PubSubEndpoint

pubsub = PubSubEndpoint()

然后，我们将pubsub实例传递给WebSocket的子类，以便在其中使用推送通知的功能：

class NotificationsWebSocket(WebSocketEndpoint):
    async def on_receive(self, websocket: WebSocket, data: str):
        # 在这里实现推送通知的逻辑
        await self.pubsub.publish(self.room_name, data)

    async def on_connect(self, websocket: WebSocket):
        await self.pubsub.subscribe(self.room_name, websocket)
        await websocket.accept()

    async def on_disconnect(self, websocket: WebSocket, close_code: int):
        await self.pubsub.unsubscribe(self.room_name, websocket)

步骤三：创建前端页面
我们还需要创建一个前端页面，以便用户可以通过浏览器访问我们的应用程序。可以创建一个名为index.html的文件，并将其放在名为templates的文件夹中。在index.html中，我们可以使用WebSocket来订阅推送通知，并在接收到新数据时更新页面。以下是一个简单的示例：

<!DOCTYPE html>
<html>
<head>
    <title>Notifications</title>
</head>
<body>
    <h1>Notifications</h1>
    <ul id="messages"></ul>

    <script>
    const socket = new WebSocket("ws://localhost:8000/ws");
    
    socket.onopen = () => {
        console.log("WebSocket connection is open.");
    };
    
    socket.onmessage = (event) => {
        const message = document.createElement("li");
        message.textContent = event.data;
        document.getElementById("messages").appendChild(message);
    };
    
    socket.onclose = () => {
        console.log("WebSocket connection is closed.");
    };
    
    </script>
</body>
</html>

步骤四：启动应用程序
在main.py的末尾，我们需要添加以下代码以启动FastAPI应用程序：

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

现在，我们可以使用以下命令启动我们的FastAPI应用程序：

python main.py

完成！现在，我们可以通过浏览器访问http://localhost:8000

2단계: 푸시 알림 로직 작성

WebSocket을 사용하여 푸시 알림 기능을 구현하겠습니다. WebSocket은 클라이언트와 서버 간의 전이중 통신을 가능하게 하는 프로토콜입니다. FastAPI에서는 타사 라이브러리 fastapi-websocket를 사용하여 애플리케이션에 WebSocket 지원을 쉽게 추가할 수 있습니다. 다음 명령을 사용하여 라이브러리를 설치할 수 있습니다.
rrreee

main.py에서 먼저 필요한 모듈을 가져옵니다. 🎜rrreee🎜그런 다음 FastAPI 애플리케이션을 만듭니다. 🎜rrreee🎜다음으로 필요합니다. WebSocket의 논리를 정의합니다. WebSocket의 하위 클래스를 생성하고 그 안에 푸시 알림 기능을 구현하겠습니다. 🎜rrreee🎜 on_receive 메서드에서 필요에 따라 특정 주제나 채널을 구독하고 WebSocket을 통해 새 데이터를 클라이언트에 보낼 수 있습니다. . FastAPI 애플리케이션에서는 PubSubEndpoint를 사용하여 구독 및 게시 로직을 관리할 수 있습니다. fastapi_websocket_pubsub 라이브러리는 다음 명령을 사용하여 설치할 수 있습니다. 🎜rrreee🎜main.py에서 PubSubEndpoint를 가져오고 인스턴스를 생성합니다. 🎜rrreee🎜그런 다음 pubsub를 전달합니다. 인스턴스 푸시 알림 기능을 사용하려면 WebSocket의 하위 클래스를 제공하세요. 🎜rrreee🎜 3단계: 프런트 엔드 페이지 생성 🎜 또한 사용자가 애플리케이션을 통해 애플리케이션에 액세스할 수 있도록 프런트 엔드 페이지도 생성해야 합니다. 브라우저. index.html이라는 파일을 만들어 templates라는 폴더에 넣을 수 있습니다. index.html에서 WebSocket을 사용하여 푸시 알림을 구독하고 새 데이터가 수신되면 페이지를 업데이트할 수 있습니다. 다음은 간단한 예입니다. 🎜rrreee🎜 4단계: 애플리케이션 시작 🎜 main.py 끝에 FastAPI 애플리케이션을 시작하려면 다음 코드를 추가해야 합니다. 🎜rrreee🎜이제, 다음 명령은 FastAPI 애플리케이션을 시작합니다: 🎜rrreee🎜Done! 이제 브라우저를 통해 http://localhost:8000에 접속하여 푸시 알림 기능을 확인할 수 있습니다. 새로운 데이터가 푸시되면 페이지가 자동으로 업데이트되어 최신 정보가 표시됩니다. 🎜🎜결론: 🎜이 글에서는 FastAPI에서 푸시 알림을 사용하여 실시간으로 데이터를 업데이트하는 방법을 설명합니다. WebSocket 및 PubSubEndpoint 라이브러리를 사용하면 푸시 알림 기능을 쉽게 구현할 수 있습니다. 실시간으로 데이터를 업데이트하는 이 방법은 실시간 거래, 실시간 모니터링, 실시간 게임 등 다양한 애플리케이션 시나리오에 적용될 수 있습니다. 이 글이 여러분에게 도움이 되기를 바라며, FastAPI를 사용하여 보다 효율적인 실시간 웹 애플리케이션을 구축할 수 있기를 바랍니다. 🎜

위 내용은 FastAPI에서 푸시 알림을 사용하여 실시간으로 데이터를 업데이트하는 방법의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!

Python html fastapi 子类 http websocket

성명：

이전 기사：Python 및 Baidu Map API를 사용하여 지도 그리기 및 표시 지점 기능을 구현하는 단계다음 기사：Python 및 Baidu Map API를 사용하여 지도 그리기 및 표시 지점 기능을 구현하는 단계