はじめに
はいどうも、たねだです。
相変わらずPythonの勉強をしています。
さて、今回はFastAPIを使って最小限の機能を備えたリアルタイムチャットの仕組みを作りました。
引き続きQiitaに掲載していくかどうかはさておき、今後さらに内容をアップデートしていきたいと思った仕組みになったので
ちょっとしたプロトタイプとして記事に残そうと思います。
ただ、実際に中身を作ったのは少し前のことなので忘れているところもあるかもしれませんがあしからず。
また、いつも通り今回の内容も公開リポジトリを置いてますので、
よかったらこちらもご覧いただければと思います。
さっそく作っていく
今回もPython3.12.4でつくっていきます。
私の記憶が正しければ、特別に構築したような環境はなかったと思います。
特に今回はほぼ最小単位でつくりますのでディレクトリ構成も下記の通り非常にコンパクトです。
では早速作成にあたって必要なライブラリをインストールしていきます。
必要なライブラリはざっくりと以下の通り。
- fastapi
- asyncpg
- jinja2
- python-multipart
- uvicorn[standard]
ポイントはASGIに関わるuvicornのインストールですが、今回は単体ではなくwebsocketsも利用するのであわせてuvloopなどを含むuvicorn[standard]をインストールしておきます。
続いて、main.pyを記述していきます。
from typing import List
import json
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from fastapi.staticfiles import StaticFiles
app = FastAPI()
# staticとtemplatesの設定
app.mount("/static", StaticFiles(directory="static"), name="static")
templates = Jinja2Templates(directory="templates")
clients: List[WebSocket] = []
message_history: List[dict] = []
# エンドポイントの設定
@app.get("/", response_class=HTMLResponse)
async def get(request: Request):
return templates.TemplateResponse("chat.html", {"request": request})
# チャットシステムの処理
@app.websocket("/ws/chat")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
if message_history:
await websocket.send_text(json.dumps({"type": "history",
"messages": message_history}))
clients.append(websocket)
try:
user_name = await websocket.receive_text()
while True:
data = await websocket.receive_text()
data_json = json.loads(data)
if data_json.get("type") == "message":
message = {"user": user_name, "message": data_json["text"]}
message_history.append(message)
for client in clients:
await client.send_text(json.dumps({"type": "message",
"message": message}))
elif data_json.get("type") == "change-username":
new_user_name = data_json["newUserName"]
old_user_name = user_name
user_name = new_user_name
for client in clients:
await client.send_text(
json.dumps({"type": "username-change",
"oldUserName": old_user_name,
"newUserName": new_user_name}))
# 切断時の処理
except WebSocketDisconnect:
clients.remove(websocket)
except Exception as e:
print(f"An error occurred: {e}")
チャットシステムの処理はjavascriptで行っています。
javascriptはPython以上にまだまだなので
今回は下記のように記述してみましたが、おそらく要改善です。。
const localStorageKey = "chat_user_name";
let userName = localStorage.getItem(localStorageKey);
if (!userName) {
userName = prompt("あなたの名前を入力してください:");
if (!userName) {
alert("ユーザー名が必要です。ページをリロードしてください。");
window.location.reload();
}
localStorage.setItem(localStorageKey, userName);
}
const ws = new WebSocket("ws://localhost:8000/ws/chat");
ws.onopen = function() {
ws.send(userName);
};
const messagesDiv = document.getElementById("messages");
const input = document.getElementById("message-input");
const button = document.getElementById("send-button");
const changeUsernameButton = document.getElementById("change-username-button");
const newUsernameInput = document.getElementById("new-username-input");
ws.onmessage = function(event) {
const data = JSON.parse(event.data);
if (data.type === "history") {
data.messages.forEach(message => {
const messageDiv = document.createElement("div");
messageDiv.textContent = `${message.user}: ${message.message}`;
messagesDiv.appendChild(messageDiv);
});
} else if (data.type === "message") {
const messageDiv = document.createElement("div");
messageDiv.textContent = `${data.message.user}: ${data.message.message}`;
messagesDiv.appendChild(messageDiv);
}
messagesDiv.scrollTop = messagesDiv.scrollHeight;
};
button.onclick = function() {
const message = input.value;
if (message.trim() !== "") {
ws.send(JSON.stringify({type: "message", text: message}));
input.value = "";
}
};
input.addEventListener("keypress", function(event) {
if (event.key === "Enter") {
button.click();
}
});
changeUsernameButton.onclick = function() {
newUsernameInput.style.display = "block";
newUsernameInput.focus();
};
newUsernameInput.addEventListener("blur", function() {
const newUserName = newUsernameInput.value.trim();
if (newUserName) {
ws.send(JSON.stringify({type: "change-username", newUserName}));
localStorage.setItem(localStorageKey, newUserName);
userName = newUserName;
newUsernameInput.value = "";
newUsernameInput.style.display = "none";
}
});
ざっくりと上から
- 初回の名前入力処理
- チャットの表示に関する処理
- ユーザーの名前変更に関する処理
といったところですね。
今回はひとまずローカルストレージを利用して名前を保管していますが、
チャットの履歴についてはwebsocketの接続が切れると消えるようにしています。
では、最後にchat.htmlを作成します。
<!DOCTYPE html>
<html lang="ja">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>realtime-chat-system</title>
<link rel="stylesheet" href="static\css\style.css">
</head>
<body>
<div id="chat-container">
<div id="messages"></div>
<input id="message-input" type="text" placeholder="メッセージを入力">
<button id="send-button">送信</button>
<button id="change-username-button">ユーザー名変更</button>
<input id="new-username-input" type="text" placeholder="新しいユーザー名" style="display: none;">
</div>
<script src="/static/js/chat.js"></script>
</body>
</html>
以上で仕組みは完成です！
動かしてみる
では、早速動かしてみましょう。
FastAPIはmain.pyのあるディレクトリにて下記を行うことで実行が可能です。
uvicorn main:app --reload
--reloadはコードの変更時にuvicornサーバーの再起動を自動で行うようにしてくれます。
開発段階では手動でサーバーを再起動する必要がなくなるので便利だと思いますがつけるかどうかはお好みで。
上記を実行して下記のような内容が出力されたら
http://127.0.0.1:8000/ にアクセスします。
問題なく動作していれば、最初に名前の入力が求められるはずです。
このときキャンセルなどをして名前の入力をせずに進もうとすると、
以下のようなエラーが出ます。
名前を入力してメッセージ欄に入力した文字を送信すると、
入力欄の上部に送信したメッセージが出力されます。
また、ユーザー名変更を押すとメッセージ入力欄の下に名前の入力欄が表示されるので、ここに新たなユーザー名を入力することで次のメッセージから新しい名前でメッセージが出力されます。
おわりに
今回作成したチャットシステムはFastAPIの勉強ついでにほぼ最低限の機能で作成したものなので、不完全な部分もたくさんあると思います。
それでも、インフラ部分を整えればチャットサービスとして、さらに機能や見栄えを整えていけばより発展した何かへつながるような、そんな仕組みの基礎部分を作成できたと感じました。
今回の開発をきっかけにjavascriptにも触れ始めましたが、徐々に様々な言語を使って自分の作りたいものを作れるようになれたらいいなと思います。