44강. 앱 로그인 + 서버 인증 처리
44강 구현 중심 ⏱ 약 30분
0. 학습 목표
→ 서버에 로그인 API를 추가하고, 앱이 채팅방에 들어가기 전에 인증을 거치도록 합니다.
더보기
0.1 이번 글에서 다룰 내용
이번 글은 구현 중심 강의입니다.
43강까지 앱은 닉네임만 입력하면 바로 채팅방에 들어갈 수 있었습니다. 이번 강의에서 그 흐름에 로그인 단계를 추가합니다. 앱은 아이디·비밀번호를 서버에 HTTP로 보내고, 서버가 검증한 뒤 닉네임을 돌려줍니다. 검증에 성공하면 그 닉네임으로 WebSocket에 입장합니다.
이번 강의에서 사용자 정보는 메모리 딕셔너리에 저장합니다. 프로그램을 재시작하면 사라지는 임시 저장 방식입니다. 46강에서 이 부분을 MySQL로 교체합니다. 지금은 로그인 흐름 자체를 만드는 데 집중합니다.
| 구분 | 내용 |
| 이해할 것 | HTTP 로그인 API와 WebSocket 채팅이 같은 FastAPI 서버에서 함께 동작하는 방식 |
| 만들 것 | server/app.py 수정 — POST /register, POST /login 추가desktop/main.py 수정 — 로그인 화면, HTTP 요청 처리 |
| 확인할 것 | 회원가입 후 로그인 성공 시 채팅방에 입장되는 것, 틀린 비밀번호에서 오류 메시지가 표시되는 것 |
0.2 이번 강의에서 수정하는 구조
fastapi-chat/
├── server/
│ ├── app.py ← ✏️ 사용자 딕셔너리, /register, /login 엔드포인트 추가
│ └── test_client.html ← 유지
└── desktop/
└── main.py ← ✏️ 로그인 화면으로 교체, HTTP 요청 함수 추가(✏️ 수정 · 표시 없음은 변경 없음)
로그인 전체 흐름은 다음과 같습니다.
앱 로그인 화면
↓ 아이디·비밀번호 입력 → [로그인] 클릭
HTTP POST /login → 서버가 검증
↓ 성공: {"ok": true, "nickname": "홍길동"}
WebSocket /ws 연결 → join {"type":"join","sender":"홍길동"}
↓
채팅 화면 진입 ← 이번 강의의 성공 지점
1. 서버 - 사용자 저장소와 인증 엔드포인트 만들기
→ app.py에 사용자 딕셔너리와 /register, /login 라우터를 추가합니다.
더보기
43강 app.py의 import 줄과 app = FastAPI() 바로 아래에 사용자 저장소와 요청 스키마를 추가합니다.
from fastapi import FastAPI, WebSocket
from fastapi.responses import JSONResponse # ➕ JSON 응답 객체
from pydantic import BaseModel # ➕ 요청 데이터 검증용
from typing import Dict
app = FastAPI()
# ── 사용자 저장소 (메모리) ──────────────────────────
users: Dict[str, dict] = {} # ➕ {"아이디": {"password": "...", "nickname": "..."}}
class UserRequest(BaseModel): # ➕ /register, /login 공통 요청 스키마
username: str
password: str
nickname: str = "" # ➕ 회원가입 때만 사용, 로그인에선 무시이제 ConnectionManager 클래스 아래에 두 엔드포인트를 추가합니다.
# ── 회원가입 ────────────────────────────────────────
@app.post("/register") # ➕
async def register(req: UserRequest):
if req.username in users: # ➕ 중복 아이디 확인
return JSONResponse(
{"ok": False, "message": "이미 사용 중인 아이디입니다."},
status_code=400
)
users[req.username] = { # ➕ 사용자 저장
"password": req.password,
"nickname": req.nickname or req.username # ➕ 닉네임 없으면 아이디로 대체
}
return {"ok": True, "message": "회원가입 성공"}
# ── 로그인 ──────────────────────────────────────────
@app.post("/login") # ➕
async def login(req: UserRequest):
user = users.get(req.username)
if user is None: # ➕ 아이디 없음
return JSONResponse(
{"ok": False, "message": "아이디를 찾을 수 없습니다."},
status_code=401
)
if user["password"] != req.password: # ➕ 비밀번호 불일치
return JSONResponse(
{"ok": False, "message": "비밀번호가 틀렸습니다."},
status_code=401
)
return {"ok": True, "nickname": user["nickname"]} # ➕ 성공: 닉네임 반환BaseModel을 쓰면 FastAPI가 요청 본문을 자동으로 파싱하고 타입을 검증합니다. 필드가 없거나 타입이 맞지 않으면 FastAPI가 자동으로 422 오류를 돌려줍니다. 직접 파싱 코드를 짤 필요가 없습니다.
비밀번호를 평문으로 저장하는 것은 실습용 단순화입니다. 실제 서비스에서는 bcrypt 같은 해시 함수를 써야 합니다. 46강에서 MySQL로 전환할 때 함께 개선합니다.
서버가 실행 중이라면 --reload 옵션 덕분에 파일 저장만 해도 자동으로 재시작됩니다. FastAPI는 자동 문서를 제공합니다. 브라우저에서 http://127.0.0.1:8000/docs를 열면 방금 추가한 두 엔드포인트를 직접 테스트해볼 수 있습니다.
✔ 확인 기준: http://127.0.0.1:8000/docs에서 POST /register로 계정을 하나 만들고, POST /login으로 로그인하면 {"ok": true, "nickname": "..."}가 반환되면 완료.
2. 앱 - 로그인 화면과 HTTP 요청 추가하기
→ 닉네임 입력 화면을 로그인 화면으로 교체하고, 서버에 HTTP 요청을 보내는 함수를 추가합니다.
더보기
먼저 HTTP 요청에 쓸 requests 라이브러리를 설치합니다.
pip install requestsdesktop/main.py의 import에 requests를 추가하고, 서버 주소 상수를 하나 더 추가합니다.
import asyncio
import json
import sys
import requests # ✏️ HTTP 요청용
import websockets
from PySide6.QtCore import QThread, Signal
from PySide6.QtWidgets import (
QApplication, QWidget, QVBoxLayout, QHBoxLayout,
QTextEdit, QLineEdit, QPushButton, QLabel, QMessageBox, # ✏️ QMessageBox 추가
)
WS_URL = "ws://127.0.0.1:8000/ws"
HTTP_URL = "http://127.0.0.1:8000" # ➕ HTTP API 기본 주소ChatClient와 ReceiverThread는 43강과 동일합니다. ChatWindow에서 _build_join_ui()만 교체합니다.
회원가입과 로그인 버튼, 오류 메시지 레이블을 추가합니다.
class ChatWindow(QWidget):
def __init__(self):
super().__init__()
self.setWindowTitle("채팅")
self.resize(400, 320)
self.receiver = None
self.nickname = ""
self._build_login_ui() # ✏️ _build_join_ui → _build_login_ui
def _build_login_ui(self): # ✏️ 닉네임 화면 → 로그인 화면으로 교체
self.login_widget = QWidget()
layout = QVBoxLayout(self.login_widget)
self.username_input = QLineEdit() # ➕ 아이디 입력
self.username_input.setPlaceholderText("아이디")
self.password_input = QLineEdit() # ➕ 비밀번호 입력
self.password_input.setPlaceholderText("비밀번호")
self.password_input.setEchoMode(QLineEdit.EchoMode.Password) # ➕ 입력 내용 숨김
self.nickname_reg_input = QLineEdit() # ➕ 닉네임 (회원가입 시 사용)
self.nickname_reg_input.setPlaceholderText("닉네임 (회원가입 시 입력)")
self.error_label = QLabel("") # ➕ 오류 메시지 표시
self.error_label.setStyleSheet("color: red;")
self.login_button = QPushButton("로그인") # ➕
self.register_button = QPushButton("회원가입") # ➕
btn_layout = QHBoxLayout()
btn_layout.addWidget(self.login_button)
btn_layout.addWidget(self.register_button)
layout.addStretch()
layout.addWidget(QLabel("아이디:"))
layout.addWidget(self.username_input)
layout.addWidget(QLabel("비밀번호:"))
layout.addWidget(self.password_input)
layout.addWidget(QLabel("닉네임 (회원가입 전용):"))
layout.addWidget(self.nickname_reg_input)
layout.addWidget(self.error_label)
layout.addLayout(btn_layout)
layout.addStretch()
self.login_button.clicked.connect(self._on_login)
self.register_button.clicked.connect(self._on_register)
self.password_input.returnPressed.connect(self._on_login)
main_layout = QVBoxLayout(self)
main_layout.addWidget(self.login_widget)
def _on_register(self): # ➕ 회원가입 버튼 처리
username = self.username_input.text().strip()
password = self.password_input.text()
nickname = self.nickname_reg_input.text().strip()
if not username or not password:
self.error_label.setText("아이디와 비밀번호를 입력하세요.")
return
try:
res = requests.post(f"{HTTP_URL}/register", json={ # ➕ 서버에 회원가입 요청
"username": username,
"password": password,
"nickname": nickname
})
data = res.json()
if data.get("ok"):
self.error_label.setStyleSheet("color: green;")
self.error_label.setText("회원가입 완료. 로그인하세요.")
else:
self.error_label.setStyleSheet("color: red;")
self.error_label.setText(data.get("message", "회원가입 실패"))
except Exception:
self.error_label.setText("서버에 연결할 수 없습니다.")
def _on_login(self): # ✏️ _on_join → _on_login으로 교체
username = self.username_input.text().strip()
password = self.password_input.text()
if not username or not password:
self.error_label.setText("아이디와 비밀번호를 입력하세요.")
return
try:
res = requests.post(f"{HTTP_URL}/login", json={ # ➕ 서버에 로그인 요청
"username": username,
"password": password
})
data = res.json()
except Exception:
self.error_label.setText("서버에 연결할 수 없습니다.")
return
if not data.get("ok"): # ➕ 로그인 실패
self.error_label.setText(data.get("message", "로그인 실패"))
return
# 로그인 성공 — nickname으로 WebSocket 입장
self.nickname = data["nickname"] # ➕ 서버가 돌려준 닉네임 저장
self.receiver = ReceiverThread(self.nickname)
self.receiver.message_received.connect(self._on_message)
self.receiver.start()
self.login_widget.hide()
self._build_chat_ui() # 채팅 화면으로 전환 (43강과 동일)requests.post()는 동기 함수입니다. 버튼을 클릭하면 GUI 스레드에서 바로 호출됩니다. 서버 응답이 돌아오는 짧은 시간 동안 창이 잠깐 굳는 것처럼 보일 수 있지만, 로컬 서버에서는 응답이 매우 빨라 체감되지 않습니다. 느린 네트워크 환경에서는 이 요청도 스레드로 분리하는 것이 좋지만, 지금은 흐름을 먼저 만드는 데 집중합니다.
_build_chat_ui()와 _on_send(), _on_message()는 43강 코드와 동일합니다. 변경 없이 그대로 사용합니다.
3. 실행 결과 확인하기
→ 회원가입 → 로그인 → 채팅 흐름과 실패 케이스를 확인합니다.
더보기
3.1 정상 흐름 확인
서버를 실행하고 앱을 실행합니다. 아이디 hong, 비밀번호 1234, 닉네임 홍길동으로 먼저 회원가입합니다.
아이디: hong
비밀번호: 1234
닉네임: 홍길동
→ [회원가입] 클릭
→ "회원가입 완료. 로그인하세요." (초록 글씨)
같은 아이디·비밀번호로 로그인하면 채팅 화면으로 전환됩니다. 브라우저 test_client.html에 입장 안내가 표시됩니다.
→ [로그인] 클릭
→ 채팅 화면 진입 (상단: "접속 중: 홍길동")
[브라우저]
[시스템] 홍길동 님이 입장했습니다.✔ 확인 기준: 로그인 성공 후 채팅 화면으로 전환되고, 브라우저에 입장 안내가 표시되면 완료. 상단 상태 레이블에 닉네임이 표시되는지 확인하세요.
3.2 실패 케이스 확인
| 시도 | 예상 결과 |
| 틀린 비밀번호로 로그인 | "비밀번호가 틀렸습니다." 표시, 채팅 화면 전환 없음 |
| 없는 아이디로 로그인 | "아이디를 찾을 수 없습니다." 표시 |
| 이미 있는 아이디로 회원가입 | "이미 사용 중인 아이디입니다." 표시 |
| 서버를 끄고 로그인 시도 | "서버에 연결할 수 없습니다." 표시, 앱은 유지 |
서버를 재시작하면 메모리에 저장한 사용자 정보가 사라집니다. 회원가입부터 다시 해야 합니다. 이 문제는 46강에서 MySQL로 전환하면 해결됩니다.
✔ 확인 기준: 실패 케이스마다 채팅 화면으로 넘어가지 않고, 오류 메시지가 빨간 글씨로 표시되면 완료. 앱이 종료되거나 오류 없이 아무것도 표시되지 않는다면 try/except와 error_label.setText() 호출을 확인하세요.
4. 최종 코드 정리하기
→ 이번 강의에서 완성한 두 파일 전체를 정리합니다.
더보기
4.1 server/app.py
from fastapi import FastAPI, WebSocket
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import Dict
app = FastAPI()
# ── 사용자 저장소 (메모리) ──────────────────────────
users: Dict[str, dict] = {}
class UserRequest(BaseModel):
username: str
password: str
nickname: str = ""
# ── 인증 엔드포인트 ─────────────────────────────────
@app.post("/register")
async def register(req: UserRequest):
if req.username in users:
return JSONResponse(
{"ok": False, "message": "이미 사용 중인 아이디입니다."},
status_code=400
)
users[req.username] = {
"password": req.password,
"nickname": req.nickname or req.username
}
return {"ok": True, "message": "회원가입 성공"}
@app.post("/login")
async def login(req: UserRequest):
user = users.get(req.username)
if user is None:
return JSONResponse(
{"ok": False, "message": "아이디를 찾을 수 없습니다."},
status_code=401
)
if user["password"] != req.password:
return JSONResponse(
{"ok": False, "message": "비밀번호가 틀렸습니다."},
status_code=401
)
return {"ok": True, "nickname": user["nickname"]}
# ── WebSocket ───────────────────────────────────────
class ConnectionManager:
def __init__(self):
self.active_connections: Dict[WebSocket, str] = {}
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections[websocket] = ""
def disconnect(self, websocket: WebSocket):
self.active_connections.pop(websocket, None)
def set_nickname(self, websocket: WebSocket, nickname: str):
self.active_connections[websocket] = nickname
def get_nickname(self, websocket: WebSocket) -> str:
return self.active_connections.get(websocket, "")
def get_nicknames(self) -> list:
return [n for n in self.active_connections.values() if n]
def get_socket_by_nickname(self, nickname: str):
for ws, name in self.active_connections.items():
if name == nickname:
return ws
return None
async def broadcast(self, message: dict):
for connection in self.active_connections:
await connection.send_json(message)
manager = ConnectionManager()
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_json()
msg_type = data.get("type", "")
if msg_type == "join":
nickname = data.get("sender", "익명")
manager.set_nickname(websocket, nickname)
await manager.broadcast({
"type": "system",
"text": f"{nickname} 님이 입장했습니다."
})
await manager.broadcast({
"type": "user_list",
"users": manager.get_nicknames()
})
elif msg_type == "chat":
await manager.broadcast({
"type": "chat",
"sender": manager.get_nickname(websocket),
"text": data.get("text", "")
})
elif msg_type == "whisper":
target_name = data.get("target", "")
text = data.get("text", "")
sender_name = manager.get_nickname(websocket)
target_ws = manager.get_socket_by_nickname(target_name)
if target_ws is None:
await websocket.send_json({
"type": "system",
"text": f"'{target_name}' 님은 현재 접속해 있지 않습니다."
})
else:
whisper_msg = {
"type": "whisper",
"sender": sender_name,
"text": text
}
await target_ws.send_json(whisper_msg)
await websocket.send_json(whisper_msg)
except Exception:
nickname = manager.get_nickname(websocket)
manager.disconnect(websocket)
if nickname:
await manager.broadcast({
"type": "system",
"text": f"{nickname} 님이 퇴장했습니다."
})
await manager.broadcast({
"type": "user_list",
"users": manager.get_nicknames()
})
4.2 desktop/main.py
import asyncio
import json
import sys
import requests
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"
HTTP_URL = "http://127.0.0.1:8000"
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(400, 320)
self.receiver = None
self.nickname = ""
self._build_login_ui()
def _build_login_ui(self):
self.login_widget = QWidget()
layout = QVBoxLayout(self.login_widget)
self.username_input = QLineEdit()
self.username_input.setPlaceholderText("아이디")
self.password_input = QLineEdit()
self.password_input.setPlaceholderText("비밀번호")
self.password_input.setEchoMode(QLineEdit.EchoMode.Password)
self.nickname_reg_input = QLineEdit()
self.nickname_reg_input.setPlaceholderText("닉네임 (회원가입 시 입력)")
self.error_label = QLabel("")
self.error_label.setStyleSheet("color: red;")
self.login_button = QPushButton("로그인")
self.register_button = QPushButton("회원가입")
btn_layout = QHBoxLayout()
btn_layout.addWidget(self.login_button)
btn_layout.addWidget(self.register_button)
layout.addStretch()
layout.addWidget(QLabel("아이디:"))
layout.addWidget(self.username_input)
layout.addWidget(QLabel("비밀번호:"))
layout.addWidget(self.password_input)
layout.addWidget(QLabel("닉네임 (회원가입 전용):"))
layout.addWidget(self.nickname_reg_input)
layout.addWidget(self.error_label)
layout.addLayout(btn_layout)
layout.addStretch()
self.login_button.clicked.connect(self._on_login)
self.register_button.clicked.connect(self._on_register)
self.password_input.returnPressed.connect(self._on_login)
main_layout = QVBoxLayout(self)
main_layout.addWidget(self.login_widget)
def _on_register(self):
username = self.username_input.text().strip()
password = self.password_input.text()
nickname = self.nickname_reg_input.text().strip()
if not username or not password:
self.error_label.setText("아이디와 비밀번호를 입력하세요.")
return
try:
res = requests.post(f"{HTTP_URL}/register", json={
"username": username,
"password": password,
"nickname": nickname
})
data = res.json()
if data.get("ok"):
self.error_label.setStyleSheet("color: green;")
self.error_label.setText("회원가입 완료. 로그인하세요.")
else:
self.error_label.setStyleSheet("color: red;")
self.error_label.setText(data.get("message", "회원가입 실패"))
except Exception:
self.error_label.setText("서버에 연결할 수 없습니다.")
def _on_login(self):
username = self.username_input.text().strip()
password = self.password_input.text()
if not username or not password:
self.error_label.setText("아이디와 비밀번호를 입력하세요.")
return
try:
res = requests.post(f"{HTTP_URL}/login", json={
"username": username,
"password": password
})
data = res.json()
except Exception:
self.error_label.setText("서버에 연결할 수 없습니다.")
return
if not data.get("ok"):
self.error_label.setText(data.get("message", "로그인 실패"))
return
self.nickname = data["nickname"]
self.receiver = ReceiverThread(self.nickname)
self.receiver.message_received.connect(self._on_message)
self.receiver.start()
self.login_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())
4.3 최종 확인 표
| 코드 | 역할 |
users: Dict[str, dict] = {} |
메모리 사용자 저장소 — 서버 재시작 시 초기화 |
class UserRequest(BaseModel) |
요청 본문 자동 파싱·검증 |
POST /register |
아이디 중복 확인 후 사용자 저장 |
POST /login |
아이디·비밀번호 검증 후 닉네임 반환 |
JSONResponse(..., status_code=401) |
실패 응답 — HTTP 상태코드와 함께 이유 전달 |
self.password_input.setEchoMode(...Password) |
비밀번호 입력 숨김 처리 |
requests.post(f"{HTTP_URL}/login", json={...}) |
서버 로그인 API 호출 |
self.nickname = data["nickname"] |
서버가 돌려준 닉네임으로 WebSocket 입장 |
error_label.setText(...) |
실패 원인을 화면에 표시 — 앱 종료 없이 재시도 가능 |
→ 다음 강의 (45강): 서버에 GET /admin 페이지를 추가합니다. 브라우저에서 이 주소를 열면 현재 접속자 목록과 로그인 기록을 확인할 수 있는 웹 관리창이 나타납니다. FastAPI의 HTMLResponse로 HTML을 직접 돌려주는 방식을 사용합니다.