46강. 데이터 저장: 메모리에서 MySQL로
0. 학습 목표
→ 메모리 딕셔너리로 저장하던 사용자 정보와 로그인 기록을 MySQL 테이블로 옮겨, 서버를 재시작해도 데이터가 유지되게 만듭니다.
이번 글은 리팩토링 강의입니다.
45강까지 완성한 서버에는 한 가지 약점이 있습니다.
users와 login_log가 파이썬 딕셔너리와 리스트로만 저장되기 때문에,
서버를 재시작하면 회원가입한 계정과 로그인 기록이 모두 사라집니다.
이번 강의에서 이 두 저장소를 MySQL 테이블로 교체합니다.
바뀌는 것은 server/app.py의 데이터 저장·조회 코드뿐입니다.
WebSocket 흐름, 로그인·회원가입 API, 관리 페이지 HTML, PySide6 앱 코드는 전혀 손대지 않습니다.
앱 동작도 전혀 달라지지 않습니다. 변화는 데이터가 어디에 저장되느냐뿐입니다.
아래 코드에서 ➕는 45강 코드에 없던 새 줄, ✏️는 형태가 바뀐 줄입니다.
최종 정리 코드에는 마커 없이 깨끗한 코드만 싣습니다.
| 구분 | 내용 |
| 이해할 것 | 메모리 저장과 DB 저장의 차이, pymysql로 INSERT/SELECT를 수행하는 방법 |
| 만들 것 | server/app.py 수정 — users 딕셔너리·login_log 리스트를 MySQL 테이블로 교체 |
| 확인할 것 | 서버 재시작 후에도 회원가입한 계정으로 로그인되는 것, 로그인 기록이 유지되는 것 |
0.2 이번 강의에서 직접 다루는 구조
chat_project/
├── server/
│ └── app.py ← ✏️ 메모리 저장소 → MySQL로 교체
└── desktop/
└── main.py ← 변경 없음(➕ 새로 생성 · ✏️ 수정 · 표시 없음은 변경 없음)
server/app.py 안에서 이번 강의에서 교체되는 부분입니다.
server/app.py (변경 사항)
├── import pymysql ← ➕ 새로 추가
├── DB 연결 함수 get_db() ← ➕ 새로 추가
├── 테이블 생성 함수 init_db() ← ➕ 새로 추가
├── users: Dict[str, dict] = {} ← ✏️ 삭제 → DB 조회로 교체
├── login_log: List[dict] = [] ← ✏️ 삭제 → DB INSERT로 교체
├── /register 엔드포인트 ← ✏️ dict 저장 → INSERT 쿼리로 교체
├── /login 엔드포인트 ← ✏️ dict 조회 → SELECT 쿼리로 교체
└── /admin 엔드포인트 ← ✏️ login_log 리스트 → DB SELECT로 교체
1. 기존 구조 확인하기
→ 메모리 저장 방식의 어느 부분이 문제이고, 무엇을 어떻게 바꿀지 먼저 정리합니다.
1.1 현재 저장 방식의 문제
45강 코드에서 사용자 정보와 로그인 기록은 다음 두 변수에 담겨 있습니다.
users: Dict[str, dict] = {} # 회원가입 정보 — 프로세스 메모리에만 존재
login_log: List[dict] = [] # 로그인 기록 — 프로세스 메모리에만 존재파이썬 변수는 프로그램이 실행되는 동안만 살아 있습니다. uvicorn을
로 종료하는 순간 두 변수의 내용은 사라지고, 다음 실행 때 빈 상태로 시작합니다.
회원가입한 계정으로 로그인하려면 서버를 끄지 않아야 하고, 관리 페이지의 로그인 기록도 재시작하면 리셋됩니다.
이것을 해결하려면 데이터를 프로세스 바깥, 즉 파일이나 데이터베이스에 저장해야 합니다.
1.2 변경 전후 대응표
두 메모리 변수가 각각 어떤 MySQL 테이블로 바뀌는지 먼저 정리합니다.
| 기존 (메모리) | 바뀐 것 (MySQL) | 저장하는 내용 |
users: Dict[str, dict] |
users 테이블 |
아이디, 비밀번호, 닉네임 |
login_log: List[dict] |
login_log 테이블 |
로그인 아이디, 닉네임, 시각 |
테이블 이름을 기존 변수 이름과 같게 맞췄습니다. 코드를 읽을 때 "이 쿼리가 어떤 데이터를 다루는지"가 바로 보입니다.
2. 실습 준비하기
→ MySQL 데이터베이스와 테이블을 만들고 pymysql을 설치합니다.
2.1 데이터베이스 준비
MySQL에 접속해 이번 강의에서 사용할 데이터베이스와 테이블 두 개를 만듭니다. MySQL 클라이언트(mysql -u root -p)나 DBeaver 같은 GUI 도구에서 아래 쿼리를 실행합니다.
-- 데이터베이스 생성 (이미 있으면 건너뜁니다)
CREATE DATABASE IF NOT EXISTS chat_app CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
USE chat_app;
-- 사용자 테이블
CREATE TABLE IF NOT EXISTS users (
username VARCHAR(50) PRIMARY KEY,
password VARCHAR(100) NOT NULL,
nickname VARCHAR(50) NOT NULL
);
-- 로그인 기록 테이블
CREATE TABLE IF NOT EXISTS login_log (
id INT AUTO_INCREMENT PRIMARY KEY,
username VARCHAR(50) NOT NULL,
nickname VARCHAR(50) NOT NULL,
login_time DATETIME NOT NULL
);users 테이블의 PRIMARY KEY는 username입니다. 45강 코드에서 if req.username in users:로 중복 아이디를 확인하던 것을, MySQL에서는 PRIMARY KEY 제약이 대신 처리합니다. 같은 username으로 INSERT를 시도하면 자동으로 오류가 발생합니다.
login_log의 id는 AUTO_INCREMENT로, 기록을 추가할 때 자동으로 번호가 붙습니다.
2.2 pymysql 설치
파이썬에서 MySQL에 접속하려면 드라이버가 필요합니다. 이 강의에서는 pymysql을 사용합니다.
pip install pymysql설치가 끝나면 터미널에서 바로 확인할 수 있습니다.
python -c "import pymysql; print(pymysql.__version__)"버전 번호가 출력되면 준비 완료입니다.
| 항목 | 값 (환경에 맞게 수정) |
| host | "localhost" |
| user | "root" (본인 MySQL 계정) |
| password | 본인 MySQL 비밀번호 |
| database | "chat_app" |
| charset | "utf8mb4" |
✔ 확인 기준: CREATE TABLE 실행 후 MySQL에서 SHOW TABLES;를 입력했을 때 login_log와 users가 나타나면 완료.
3. 서버 - DB 연결 함수 추가하기
→ app.py 상단에 pymysql import와 DB 연결 함수를 추가합니다.
45강 app.py의 import 블록과 메모리 저장소 선언 부분을 다음으로 교체합니다.
from fastapi import FastAPI, WebSocket
from fastapi.responses import JSONResponse, HTMLResponse
from pydantic import BaseModel
from typing import Dict
from datetime import datetime
import pymysql # ➕ MySQL 드라이버
app = FastAPI()
# ── DB 연결 설정 ────────────────────────────────────
DB_CONFIG = { # ➕ 연결 정보를 한 곳에 모아둠
"host": "localhost",
"user": "root",
"password": "본인_비밀번호", # ➕ 실제 비밀번호로 교체
"database": "chat_app",
"charset": "utf8mb4",
}
def get_db(): # ➕ 쿼리가 필요한 함수마다 호출해 연결을 얻는다
return pymysql.connect(
**DB_CONFIG,
cursorclass=pymysql.cursors.DictCursor # ➕ SELECT 결과를 딕셔너리로 받는다
)get_db()는 호출할 때마다 새 연결을 만들고, 사용 후에는 with 블록이 끝날 때 자동으로 닫습니다. 연결 풀(connection pool)을 쓰지 않는 이유는 간결함 때문입니다. 이 프로젝트 규모에서는 충분합니다.
cursorclass=pymysql.cursors.DictCursor를 지정하면 cursor.fetchone()이 튜플 대신 {"username": "alice", "password": "...", "nickname": "앨리스"} 형태의 딕셔너리를 돌려줍니다. 키 이름으로 값을 꺼낼 수 있어 코드가 읽기 쉬워집니다.
45강에서 List를 import하던 것은 login_log: List[dict] 타입 힌트 때문이었습니다. 이 변수가 사라지므로 List import도 함께 제거합니다.
4. 서버 - 회원가입·로그인 엔드포인트 교체하기
→ /register와 /login에서 딕셔너리 읽기·쓰기를 INSERT/SELECT 쿼리로 교체합니다.
4.1 /register
기존 코드는 if req.username in users:로 중복을 확인하고 users[req.username] = {...}로 저장했습니다. MySQL로 바꾸면 INSERT가 그 역할을 모두 합니다.
@app.post("/register")
async def register(req: UserRequest):
try:
with get_db() as conn: # ➕ DB 연결 (with 블록 끝에 자동 닫힘)
with conn.cursor() as cursor:
cursor.execute( # ➕ users 테이블에 INSERT
"INSERT INTO users (username, password, nickname) VALUES (%s, %s, %s)",
(req.username, req.password, req.nickname or req.username)
)
conn.commit() # ➕ INSERT 후 반드시 commit
return {"ok": True, "message": "회원가입 성공"}
except pymysql.err.IntegrityError: # ➕ PRIMARY KEY 중복 시 발생
return JSONResponse(
{"ok": False, "message": "이미 사용 중인 아이디입니다."},
status_code=400
)45강에서는 if req.username in users:로 중복 아이디를 직접 검사했습니다. MySQL에서는 username이 PRIMARY KEY이므로, 중복 INSERT를 시도하면 pymysql.err.IntegrityError가 자동으로 발생합니다. 이 예외를 잡아서 클라이언트에게 안내 메시지를 돌려주면 됩니다.
conn.commit()은 INSERT, UPDATE, DELETE 같은 변경 작업 뒤에 반드시 호출해야 합니다. 호출하지 않으면 트랜잭션이 완료되지 않아 실제로 테이블에 반영되지 않습니다.
4.2 /login
기존 코드는 users.get(req.username)으로 딕셔너리에서 꺼냈습니다. 이것을 SELECT 쿼리로 교체하고, 로그인 기록도 login_log.append() 대신 INSERT로 저장합니다.
@app.post("/login")
async def login(req: UserRequest):
with get_db() as conn:
with conn.cursor() as cursor:
cursor.execute( # ➕ 아이디로 사용자 조회
"SELECT password, nickname FROM users WHERE username = %s",
(req.username,)
)
user = cursor.fetchone() # ➕ 결과 한 행 (없으면 None)
if user is None:
return JSONResponse(
{"ok": False, "message": "아이디를 찾을 수 없습니다."},
status_code=401
)
if user["password"] != req.password: # ✏️ user["password"] — DictCursor 덕분에 키로 접근
return JSONResponse(
{"ok": False, "message": "비밀번호가 틀렸습니다."},
status_code=401
)
with get_db() as conn: # ➕ 로그인 성공 기록 INSERT
with conn.cursor() as cursor:
cursor.execute(
"INSERT INTO login_log (username, nickname, login_time) VALUES (%s, %s, %s)",
(req.username, user["nickname"], datetime.now())
)
conn.commit()
return {"ok": True, "nickname": user["nickname"]}SELECT와 INSERT를 두 개의 with get_db() 블록으로 나눈 이유가 있습니다. SELECT 결과를 먼저 확인해 비밀번호 검증을 통과한 경우에만 INSERT를 수행하기 위해서입니다. 두 작업을 같은 연결 블록에 넣어도 동작하지만, 이렇게 분리하면 "조회 → 검증 → 기록"의 흐름이 코드에서 더 잘 보입니다.
쿼리 파라미터는 %s로 전달합니다. 값을 문자열로 직접 넣는(f"... WHERE username = '{req.username}'") 방식은 SQL 인젝션(injection) 공격에 취약합니다. pymysql이 %s로 전달된 값을 자동으로 이스케이프(escape)해줍니다.
✔ 확인 기준: 서버 실행 후 앱에서 회원가입 → 로그인이 정상적으로 되면 완료. MySQL에서 SELECT * FROM users;와 SELECT * FROM login_log;를 실행해 데이터가 실제로 저장되어 있는지 확인하세요.
5. 서버 - 관리 페이지 교체하기
→ /admin의 로그인 기록 조회를 login_log 리스트에서 DB SELECT로 교체합니다.
45강 관리 페이지에서 로그인 기록을 꺼낼 때는 reversed(login_log)를 사용했습니다. 이제 login_log 리스트가 없으므로 DB에서 직접 조회합니다. 현재 접속자 목록(manager.get_nicknames())은 WebSocket 연결 상태이므로 메모리에 그대로 유지됩니다.
@app.get("/admin", response_class=HTMLResponse)
async def admin_page():
# 현재 접속자 목록 (WebSocket — 메모리 유지)
nicknames = manager.get_nicknames()
if nicknames:
user_rows = "".join(
f"{n}"
for n in nicknames
)
else:
user_rows = "현재 접속자 없음"
# 로그인 기록 (DB에서 최신순 50건 조회) ← ✏️ login_log 리스트 → DB SELECT
with get_db() as conn:
with conn.cursor() as cursor:
cursor.execute(
"SELECT username, nickname, login_time FROM login_log ORDER BY login_time DESC LIMIT 50" # ➕
)
logs = cursor.fetchall() # ➕ 전체 결과를 리스트로 받는다
if logs:
log_rows = "".join(
f""
f"{r['login_time']}"
f"{r['username']}"
f"{r['nickname']}"
f""
for r in logs
)
else:
log_rows = "로그인 기록 없음"
# 전체 로그인 건수도 DB에서 가져옴 ← ✏️ len(login_log) → DB COUNT
with get_db() as conn:
with conn.cursor() as cursor:
cursor.execute("SELECT COUNT(*) AS cnt FROM login_log") # ➕
total = cursor.fetchone()["cnt"]
html = f"""
관리자 페이지
body {{ font-family: sans-serif; padding: 32px; background: #f8f8f8; }}
h2 {{ color: #006dd7; border-left: 5px solid #006dd7; padding-left: 12px; }}
table {{ border-collapse: collapse; width: 100%; margin-bottom: 32px; background: white; }}
th {{ background: #ddeeff; padding: 10px; border: 1px solid #ddd; text-align: left; }}
.refresh {{ font-size: 13px; color: #888; margin-bottom: 24px; }}
채팅 서버 관리 페이지마지막 조회: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")} · 새로고침현재 접속자 ({len(nicknames)}명){user_rows}닉네임로그인 기록 (총 {total}건){log_rows}시각아이디닉네임""" return HTMLResponse(content=html)45강에서는 len(login_log)로 전체 건수를 바로 알 수 있었지만, DB에서는 SELECT COUNT(*)로 별도로 조회해야 합니다. LIMIT 50이 걸린 결과 리스트의 길이를 전체 건수로 쓰면 50건을 넘는 순간 틀린 숫자가 나옵니다.
6. 실행 결과 확인하기
→ 서버를 재시작해도 데이터가 유지되는지, 관리 페이지에서 DB 기록이 표시되는지 확인합니다.
6.1 정상 동작 확인
서버를 실행하고 앱에서 회원가입 후 로그인합니다. 그 다음 서버를
Ctrl+C로 종료한 뒤 다시 실행합니다.
uvicorn server.app:app --reload서버를 재시작한 상태에서 동일한 계정으로 로그인하면 성공해야 합니다. 45강에서는 재시작 후 "아이디를 찾을 수 없습니다" 오류가 났지만, 이제는 MySQL에 계정이 남아 있으므로 정상적으로 로그인됩니다.
브라우저에서 http://127.0.0.1:8000/admin을 열면 이전 로그인 기록이 누적되어 있어야 합니다. 서버를 껐다 켜도 기록이 사라지지 않습니다.
✔ 확인 기준: 서버 재시작 후 기존 계정으로 로그인 성공, /admin에서 이전 로그인 기록이 유지되면 완료. MySQL에서 SELECT * FROM users;로 회원 데이터가 남아 있는지 직접 확인하세요.
6.2 실행이 안 될 때 확인할 것
| 오류 상황 | 원인 및 해결 방법 |
Access denied for user |
DB_CONFIG의 user나 password가 실제 MySQL 계정과 다릅니다. MySQL에 접속해 계정을 확인하세요 |
Unknown database 'chat_app' |
CREATE DATABASE chat_app을 실행하지 않았습니다. 2단계의 SQL을 먼저 실행하세요 |
Table 'chat_app.users' doesn't exist |
CREATE TABLE users를 실행하지 않았습니다. 2단계의 SQL을 다시 확인하세요 |
| 회원가입 후 데이터가 테이블에 없다 | conn.commit()이 빠진 것입니다. INSERT 뒤에 반드시 commit()을 호출해야 합니다 |
Can't connect to MySQL server |
MySQL 서버가 실행 중인지 확인하세요. Windows는 서비스 관리자, macOS/Linux는 mysql.server start로 시작합니다 |
7. 최종 코드 정리하기
→ 이번 강의에서 완성한 server/app.py 전체를 정리합니다.
7.1 server/app.py
from fastapi import FastAPI, WebSocket
from fastapi.responses import JSONResponse, HTMLResponse
from pydantic import BaseModel
from typing import Dict
from datetime import datetime
import pymysql
app = FastAPI()
DB_CONFIG = {
"host": "localhost",
"user": "root",
"password": "본인_비밀번호",
"database": "chat_app",
"charset": "utf8mb4",
}
def get_db():
return pymysql.connect(
**DB_CONFIG,
cursorclass=pymysql.cursors.DictCursor
)
class UserRequest(BaseModel):
username: str
password: str
nickname: str = ""
@app.post("/register")
async def register(req: UserRequest):
try:
with get_db() as conn:
with conn.cursor() as cursor:
cursor.execute(
"INSERT INTO users (username, password, nickname) VALUES (%s, %s, %s)",
(req.username, req.password, req.nickname or req.username)
)
conn.commit()
return {"ok": True, "message": "회원가입 성공"}
except pymysql.err.IntegrityError:
return JSONResponse(
{"ok": False, "message": "이미 사용 중인 아이디입니다."},
status_code=400
)
@app.post("/login")
async def login(req: UserRequest):
with get_db() as conn:
with conn.cursor() as cursor:
cursor.execute(
"SELECT password, nickname FROM users WHERE username = %s",
(req.username,)
)
user = cursor.fetchone()
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
)
with get_db() as conn:
with conn.cursor() as cursor:
cursor.execute(
"INSERT INTO login_log (username, nickname, login_time) VALUES (%s, %s, %s)",
(req.username, user["nickname"], datetime.now())
)
conn.commit()
return {"ok": True, "nickname": user["nickname"]}
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.get("/admin", response_class=HTMLResponse)
async def admin_page():
nicknames = manager.get_nicknames()
if nicknames:
user_rows = "".join(
f"{n}"
for n in nicknames
)
else:
user_rows = "현재 접속자 없음"
with get_db() as conn:
with conn.cursor() as cursor:
cursor.execute(
"SELECT username, nickname, login_time FROM login_log ORDER BY login_time DESC LIMIT 50"
)
logs = cursor.fetchall()
if logs:
log_rows = "".join(
f""
f"{r['login_time']}"
f"{r['username']}"
f"{r['nickname']}"
f""
for r in logs
)
else:
log_rows = "로그인 기록 없음"
with get_db() as conn:
with conn.cursor() as cursor:
cursor.execute("SELECT COUNT(*) AS cnt FROM login_log")
total = cursor.fetchone()["cnt"]
html = f"""
관리자 페이지
body {{ font-family: sans-serif; padding: 32px; background: #f8f8f8; }}
h2 {{ color: #006dd7; border-left: 5px solid #006dd7; padding-left: 12px; }}
table {{ border-collapse: collapse; width: 100%; margin-bottom: 32px; background: white; }}
th {{ background: #ddeeff; padding: 10px; border: 1px solid #ddd; text-align: left; }}
.refresh {{ font-size: 13px; color: #888; margin-bottom: 24px; }}
채팅 서버 관리 페이지마지막 조회: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")} · 새로고침현재 접속자 ({len(nicknames)}명){user_rows}닉네임로그인 기록 (총 {total}건){log_rows}시각아이디닉네임""" return HTMLResponse(content=html) @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() })
7.2 최종 확인 표
| 바뀐 코드 | 이전 코드 (45강) |
import pymysql + DB_CONFIG + get_db() |
users: Dict[str, dict] = {}login_log: List[dict] = [] |
INSERT INTO users ... + IntegrityError 처리 |
if req.username in users: + users[req.username] = {...} |
SELECT password, nickname FROM users WHERE username = %s |
users.get(req.username) |
INSERT INTO login_log ... |
login_log.append({...}) |
SELECT ... FROM login_log ORDER BY login_time DESC LIMIT 50 |
reversed(login_log) |
SELECT COUNT(*) AS cnt FROM login_log |
len(login_log) |
WebSocket 엔드포인트(/ws)와 ConnectionManager 클래스는 45강과 한 줄도 바뀌지 않았습니다. 현재 접속자는 연결이 살아 있는 동안만 유지하면 되는 데이터이고, DB에 저장할 이유가 없기 때문입니다.