43강. PySide6 앱을 WebSocket으로 연결
43강 구현 중심 ⏱ 약 40분
0. 학습 목표
→ 이번 강의의 핵심 문제와 해결 구조를 먼저 잡습니다.
더보기
0.1 이번 글에서 다룰 내용
이번 글은 구현 중심 강의입니다.
39~42강에서 브라우저로 검증한 FastAPI 서버에 PySide6 데스크톱 앱을 연결합니다. 서버는 전혀 손대지 않습니다. 이번 강의에서 만드는 것은 desktop/main.py, 하나입니다.
이 강의가 이 파트에서 가장 까다로운 이유가 있습니다. PySide6는 Qt 이벤트 루프로 돌아가고, WebSocket 수신은 asyncio 이벤트 루프가 필요합니다. 두 루프는 서로 모르는 채 한 프로세스에서 함께 돌아야 합니다. 수신 루프를 GUI 스레드에서 직접 돌리면 화면이 멈춥니다. 해결책은 QThread 안에서 asyncio 루프를 별도로 돌리고, 메시지가 오면 Qt 시그널로 GUI 스레드에 넘기는 것입니다.
| 구분 | 내용 |
| 이해할 것 | Qt 이벤트 루프와 asyncio 루프가 왜 충돌하는지, QThread + 시그널로 어떻게 분리하는지 |
| 만들 것 | desktop/main.py — ChatClient · ReceiverThread · ChatWindow |
| 확인할 것 | 앱에서 닉네임으로 입장하면 브라우저 채팅창에도 입장 안내가 뜨고, 양쪽에서 메시지를 주고받을 수 있는 것 |
0.2 이번 강의에서 만드는 구조
fastapi-chat/
├── server/
│ ├── app.py ← 유지 (42강 완성본 그대로)
│ └── test_client.html ← 유지 (브라우저 검증용)
└── desktop/
└── main.py ← ➕ 이번 강의에서 생성(➕ 새로 생성 · 표시 없음은 변경 없음)
desktop/main.py 안에서 만드는 세 클래스의 역할은 다음과 같습니다.
ChatClient — WebSocket 연결·송신·수신 루프 (asyncio 담당)
ReceiverThread — QThread 안에서 asyncio 루프 구동, 수신 메시지를 시그널로 전달
ChatWindow — GUI 화면 구성, 시그널 수신 → 화면 갱신, 사용자 입력 → 메시지 전송
1. 실습 준비하기
→ websockets 패키지를 설치하고 desktop/main.py 파일을 만듭니다.
더보기
PySide6는 이미 설치되어 있습니다. WebSocket 클라이언트 라이브러리만 추가합니다.
pip install websockets
fastapi-chat/desktop/ 폴더를 만들고 main.py 파일을 생성합니다.
mkdir desktop
서버는 42강 완성본을 그대로 사용합니다. 터미널 하나를 서버 전용으로 열어 실행해 둡니다.
uvicorn server.app:app --reload
앱 실행은 별도 터미널에서 합니다.
python desktop/main.py
2. ChatClient — WebSocket 연결과 수신 루프 만들기
→ asyncio로 서버에 연결하고, 메시지를 수신할 때마다 콜백을 호출하는 클라이언트를 만듭니다.
더보기
desktop/main.py를 만들고 아래 코드를 작성합니다. 먼저 import와 서버 주소 상수, 그리고 ChatClient부터 시작합니다.
import asyncio
import json
import sys
import websockets # ➕ WebSocket 클라이언트 라이브러리
from PySide6.QtCore import QThread, Signal, QObject
from PySide6.QtWidgets import (
QApplication, QWidget, QVBoxLayout, QHBoxLayout,
QTextEdit, QLineEdit, QPushButton, QLabel,
)
WS_URL = "ws://127.0.0.1:8000/ws" # ➕ 서버 WebSocket 주소
class ChatClient: # ➕ asyncio 담당 — WebSocket 연결·송신·수신
def __init__(self, nickname: str, on_message):
self.nickname = nickname
self.on_message = on_message # ➕ 메시지 수신 시 호출할 콜백 (ReceiverThread가 넘김)
self.ws = None # ➕ WebSocket 연결 객체
async def connect_and_run(self): # ➕ 연결 → join 전송 → 수신 루프
async with websockets.connect(WS_URL) as ws:
self.ws = ws
await ws.send(json.dumps({ # ➕ 서버에 join 메시지 전송
"type": "join",
"sender": self.nickname
}))
async for raw in ws: # ➕ 메시지가 올 때마다 반복
data = json.loads(raw)
self.on_message(data) # ➕ 콜백으로 전달 (ReceiverThread의 시그널 emit)
async def send(self, message: dict): # ➕ GUI 스레드에서 호출할 송신 메서드
if self.ws:
await self.ws.send(json.dumps(message))on_message는 함수 객체를 인자로 받습니다. ChatClient는 메시지가 오면 그 함수를 호출하기만 합니다. 실제로 어떻게 처리할지는 ChatClient가 모릅니다. 이렇게 콜백으로 분리하면 ChatClient를 수정하지 않고도 처리 방식을 바꿀 수 있습니다.
async for raw in ws:는 연결이 열려 있는 동안 메시지를 계속 기다리는 수신 루프입니다. 서버가 메시지를 보낼 때마다 이 루프의 한 바퀴가 돌아갑니다. 연결이 끊기면 루프가 자동으로 종료됩니다.
3. ReceiverThread — asyncio 루프를 QThread 안에서 돌리기
→ QThread 안에서 asyncio.run()으로 ChatClient를 구동하고, 수신 메시지를 시그널로 GUI에 넘깁니다.
더보기
왜 QThread가 필요한지 먼저 이해합니다.
asyncio.run()은 호출한 스레드를 이벤트 루프가 끝날 때까지 점유합니다.
GUI 스레드에서 직접 호출하면 Qt 이벤트 루프가 멈춰서 창이 응답하지 않습니다. 해결책은 별도 스레드를 하나 만들고 그 안에서 asyncio 루프를 돌리는 것입니다.
그런데 수신 메시지를 받아 화면에 표시하려면 GUI 위젯을 직접 건드려야 합니다. Qt에서 다른 스레드가 GUI 위젯을 직접 건드리면 오류가 납니다. 이 문제를 시그널로 풉니다. ReceiverThread는 메시지를 받으면 시그널을 emit하고, GUI 스레드의 슬롯이 그것을 받아 화면을 갱신합니다.
class ReceiverThread(QThread): # ➕ 별도 스레드에서 asyncio 루프 구동
message_received = Signal(dict) # ➕ 메시지 수신 시 GUI 스레드로 전달할 시그널
def __init__(self, nickname: str):
super().__init__()
self.nickname = nickname
self.client = ChatClient(nickname, self._on_message) # ➕ 콜백으로 자신의 메서드 전달
def _on_message(self, data: dict):
self.message_received.emit(data) # ➕ 수신 메시지를 시그널로 emit
def run(self): # ➕ QThread가 시작되면 호출되는 메서드
asyncio.run(self.client.connect_and_run()) # ➕ 이 스레드 안에서 asyncio 루프 실행
def send_message(self, message: dict): # ➕ GUI 스레드에서 메시지 전송 요청
if self.client.ws:
asyncio.run_coroutine_threadsafe( # ➕ 다른 스레드에서 asyncio 코루틴 안전 실행
self.client.send(message),
self._loop
)
def run(self):
self._loop = asyncio.new_event_loop() # ➕ 이 스레드 전용 이벤트 루프 생성
asyncio.set_event_loop(self._loop)
self._loop.run_until_complete( # ➕ 수신 루프가 끝날 때까지 루프 실행
self.client.connect_and_run()
)
asyncio.run() 대신 asyncio.new_event_loop()와 run_until_complete()를 쓰는 이유가 있습니다.
asyncio.run()은 루프를 만들고 실행한 뒤 바로 닫아버려서,
나중에 run_coroutine_threadsafe()로 외부에서 코루틴을 밀어 넣을 수 없습니다.
루프 객체를 self._loop에 저장해두면 GUI 스레드에서 send_message()를 호출할 때
이 루프에 코루틴을 안전하게 전달할 수 있습니다.
GUI 스레드 ReceiverThread (asyncio 루프)
│ │
│ send_message({"type":"chat",...}) │
│──run_coroutine_threadsafe()──────────▶│ client.send() 실행
│ │
│ 수신 메시지 │
│◀──message_received 시그널────────────│ _on_message() → emit()
│ │
on_message() 슬롯 호출 │
→ chat_display.append(...) │두 스레드는 시그널·슬롯과 run_coroutine_threadsafe()를 통해서만 대화합니다. 직접 위젯이나 소켓에 접근하지 않습니다.
4. ChatWindow — GUI 구성과 시그널 연결하기
→ 닉네임 입력 화면과 채팅 화면을 만들고, 시그널을 받아 화면을 갱신하는 슬롯을 연결합니다.
더보기
class ChatWindow(QWidget): # ➕ GUI 화면 담당
def __init__(self):
super().__init__()
self.setWindowTitle("채팅")
self.resize(560, 480)
self.receiver = None # ➕ 아직 ReceiverThread 없음
self._build_join_ui() # ➕ 닉네임 입력 화면 먼저 표시
def _build_join_ui(self): # ➕ 닉네임 입력 화면
self.join_widget = QWidget()
layout = QVBoxLayout(self.join_widget)
self.nickname_input = QLineEdit()
self.nickname_input.setPlaceholderText("닉네임을 입력하세요")
self.join_button = QPushButton("입장")
layout.addStretch()
layout.addWidget(QLabel("닉네임:"))
layout.addWidget(self.nickname_input)
layout.addWidget(self.join_button)
layout.addStretch()
self.join_button.clicked.connect(self._on_join)
self.nickname_input.returnPressed.connect(self._on_join)
main_layout = QVBoxLayout(self)
main_layout.addWidget(self.join_widget)
def _on_join(self): # ➕ 입장 버튼 클릭
nickname = self.nickname_input.text().strip()
if not nickname:
return
self.nickname = nickname
self.receiver = ReceiverThread(nickname) # ➕ ReceiverThread 생성
self.receiver.message_received.connect(self._on_message) # ➕ 시그널 연결
self.receiver.start() # ➕ QThread 시작 → run() 호출
self.join_widget.hide()
self._build_chat_ui() # ➕ 채팅 화면으로 전환
def _build_chat_ui(self): # ➕ 채팅 화면
self.chat_widget = QWidget()
layout = QVBoxLayout(self.chat_widget)
self.status_label = QLabel(f"접속 중: {self.nickname}")
self.chat_display = QTextEdit()
self.chat_display.setReadOnly(True)
self.message_input = QLineEdit()
self.message_input.setPlaceholderText("메시지 입력")
self.send_button = QPushButton("보내기")
input_layout = QHBoxLayout()
input_layout.addWidget(self.message_input)
input_layout.addWidget(self.send_button)
layout.addWidget(self.status_label)
layout.addWidget(self.chat_display)
layout.addLayout(input_layout)
self.send_button.clicked.connect(self._on_send)
self.message_input.returnPressed.connect(self._on_send)
self.layout().addWidget(self.chat_widget) # ➕ 기존 레이아웃에 채팅 위젯 추가
def _on_send(self): # ➕ 보내기 버튼 클릭
text = self.message_input.text().strip()
if not text or not self.receiver:
return
self.receiver.send_message({ # ➕ ReceiverThread를 통해 전송
"type": "chat",
"sender": self.nickname,
"text": text
})
self.message_input.clear()
def _on_message(self, data: dict): # ➕ ReceiverThread 시그널을 받는 슬롯
msg_type = data.get("type", "")
if msg_type == "system":
self.chat_display.append(f"[시스템] {data.get('text', '')}")
elif msg_type == "chat":
self.chat_display.append(f"{data.get('sender', '')}: {data.get('text', '')}")
elif msg_type == "whisper":
self.chat_display.append(f"[귓속말 from {data.get('sender', '')}] {data.get('text', '')}")
app = QApplication(sys.argv)
window = ChatWindow()
window.show()
sys.exit(app.exec())_on_message()는 GUI 스레드에서 실행됩니다.
ReceiverThread가 시그널을 emit하면 Qt가 자동으로 GUI 스레드의 슬롯을 호출합니다.
그래서 chat_display.append()처럼 위젯을 직접 건드려도 안전합니다.
user_list 타입은 이번 강의에서 처리하지 않습니다.
서버가 보내더라도 _on_message()에서 무시됩니다. 44강에서 접속자 목록 패널을 추가할 때 채웁니다.
5. 실행 결과 확인하기
→ 앱과 브라우저가 같은 채팅방에서 메시지를 주고받는지 확인합니다.
더보기
터미널 1에서 서버를 실행하고, 터미널 2에서 앱을 실행합니다. 브라우저에서도 test_client.html을 엽니다.
# 터미널 1
uvicorn server.app:app --reload
# 터미널 2
python desktop/main.py
앱에서 닉네임을 입력하고 입장하면 브라우저 채팅창에 입장 안내가 표시됩니다.
[앱 — 홍길동 입장] [브라우저]
닉네임 입력 → 입장 버튼 클릭
[시스템] 홍길동 님이 입장했습니다. [시스템] 홍길동 님이 입장했습니다.
홍길동: 안녕하세요 → 홍길동: 안녕하세요
← 김철수: 반갑습니다
[시스템] 김철수: 반갑습니다✔ 확인 기준: 앱에서 입장했을 때 브라우저에 입장 안내가 뜨고, 앱과 브라우저 양쪽에서 보낸 메시지가 서로에게 전달되면 완료. 앱 창이 응답하지 않으면 ReceiverThread.run()에서 self._loop가 제대로 설정되었는지 확인하세요.
5.2 실행이 안 될 때 확인할 것
| 증상 | 원인 및 해결 |
| 앱 창이 뜨지 않거나 바로 종료된다 | 서버가 실행 중이지 않습니다. uvicorn server.app:app --reload를 먼저 실행하세요 |
| 입장 후 앱 창이 굳어서 반응이 없다 | asyncio.run()을 GUI 스레드에서 직접 호출한 것입니다. ReceiverThread.run() 안에서만 루프를 실행해야 합니다 |
| 메시지를 보내도 서버에 전달되지 않는다 | send_message()에서 run_coroutine_threadsafe()와 self._loop를 확인하세요. self._loop가 run()에서 할당되기 전에 호출되면 오류가 납니다 |
ModuleNotFoundError: websockets |
pip install websockets가 완료되지 않은 것입니다 |
6. 최종 코드 정리하기
→ desktop/main.py 전체를 한곳에 정리합니다.
더보기
6.1 desktop/main.py
import asyncio
import json
import sys
import websockets
from PySide6.QtCore import QThread, Signal
from PySide6.QtWidgets import (
QApplication, QWidget, QVBoxLayout, QHBoxLayout,
QTextEdit, QLineEdit, QPushButton, QLabel,
)
WS_URL = "ws://127.0.0.1:8000/ws"
class ChatClient:
def __init__(self, nickname: str, on_message):
self.nickname = nickname
self.on_message = on_message
self.ws = None
async def connect_and_run(self):
async with websockets.connect(WS_URL) as ws:
self.ws = ws
await ws.send(json.dumps({
"type": "join",
"sender": self.nickname
}))
async for raw in ws:
data = json.loads(raw)
self.on_message(data)
async def send(self, message: dict):
if self.ws:
await self.ws.send(json.dumps(message))
class ReceiverThread(QThread):
message_received = Signal(dict)
def __init__(self, nickname: str):
super().__init__()
self.nickname = nickname
self.client = ChatClient(nickname, self._on_message)
self._loop = None
def _on_message(self, data: dict):
self.message_received.emit(data)
def run(self):
self._loop = asyncio.new_event_loop()
asyncio.set_event_loop(self._loop)
self._loop.run_until_complete(self.client.connect_and_run())
def send_message(self, message: dict):
if self._loop and self.client.ws:
asyncio.run_coroutine_threadsafe(
self.client.send(message),
self._loop
)
class ChatWindow(QWidget):
def __init__(self):
super().__init__()
self.setWindowTitle("채팅")
self.resize(560, 480)
self.receiver = None
self.nickname = ""
self._build_join_ui()
def _build_join_ui(self):
self.join_widget = QWidget()
layout = QVBoxLayout(self.join_widget)
self.nickname_input = QLineEdit()
self.nickname_input.setPlaceholderText("닉네임을 입력하세요")
self.join_button = QPushButton("입장")
layout.addStretch()
layout.addWidget(QLabel("닉네임:"))
layout.addWidget(self.nickname_input)
layout.addWidget(self.join_button)
layout.addStretch()
self.join_button.clicked.connect(self._on_join)
self.nickname_input.returnPressed.connect(self._on_join)
main_layout = QVBoxLayout(self)
main_layout.addWidget(self.join_widget)
def _on_join(self):
nickname = self.nickname_input.text().strip()
if not nickname:
return
self.nickname = nickname
self.receiver = ReceiverThread(nickname)
self.receiver.message_received.connect(self._on_message)
self.receiver.start()
self.join_widget.hide()
self._build_chat_ui()
def _build_chat_ui(self):
self.chat_widget = QWidget()
layout = QVBoxLayout(self.chat_widget)
self.status_label = QLabel(f"접속 중: {self.nickname}")
self.chat_display = QTextEdit()
self.chat_display.setReadOnly(True)
self.message_input = QLineEdit()
self.message_input.setPlaceholderText("메시지 입력")
self.send_button = QPushButton("보내기")
input_layout = QHBoxLayout()
input_layout.addWidget(self.message_input)
input_layout.addWidget(self.send_button)
layout.addWidget(self.status_label)
layout.addWidget(self.chat_display)
layout.addLayout(input_layout)
self.send_button.clicked.connect(self._on_send)
self.message_input.returnPressed.connect(self._on_send)
self.layout().addWidget(self.chat_widget)
def _on_send(self):
text = self.message_input.text().strip()
if not text or not self.receiver:
return
self.receiver.send_message({
"type": "chat",
"sender": self.nickname,
"text": text
})
self.message_input.clear()
def _on_message(self, data: dict):
msg_type = data.get("type", "")
if msg_type == "system":
self.chat_display.append(f"[시스템] {data.get('text', '')}")
elif msg_type == "chat":
self.chat_display.append(f"{data.get('sender', '')}: {data.get('text', '')}")
elif msg_type == "whisper":
self.chat_display.append(
f"[귓속말 from {data.get('sender', '')}] {data.get('text', '')}"
)
app = QApplication(sys.argv)
window = ChatWindow()
window.show()
sys.exit(app.exec())
6.2 최종 확인 표
| 코드 | 역할 |
ChatClient.on_message |
수신 메시지 처리를 콜백으로 분리 — ChatClient는 전달만 담당 |
async for raw in ws |
연결이 열린 동안 메시지를 계속 기다리는 수신 루프 |
message_received = Signal(dict) |
ReceiverThread → GUI 스레드 단방향 데이터 전달 |
self._loop = asyncio.new_event_loop() |
이 스레드 전용 asyncio 루프 생성 — GUI 스레드 루프와 분리 |
self._loop.run_until_complete(...) |
수신 루프가 끝날 때까지 이 스레드를 점유 |
asyncio.run_coroutine_threadsafe(coro, loop) |
GUI 스레드에서 asyncio 루프로 코루틴을 안전하게 전달 |
self.receiver.message_received.connect(self._on_message) |
시그널을 GUI 슬롯에 연결 — emit 시 GUI 스레드에서 슬롯 실행 |
chat_display.append() (슬롯 안에서) |
시그널 슬롯은 GUI 스레드에서 실행되므로 위젯 접근 안전 |
→ 다음 강의 (44강): 지금 앱은 닉네임만 입력하고 바로 채팅방에 들어갑니다. 44강에서 로그인 화면을 추가하고, 서버에 사용자 등록·인증 처리를 붙입니다. desktop/main.py의 닉네임 입력 화면이 아이디·비밀번호를 받는 로그인 화면으로 교체됩니다.