디스코드 봇 만들기#23aiosqlite로 비동기 DB 제대로 설계하기! JSON 탈출 (파이썬 예제)

같은 유저에게 동시에 경고 100번 주기
JSON  결과: 1     (99번이 증발!)
SQLite 결과: 100  (정상)

유저 1,000명   → JSON 0.29ms / SQLite 4.11ms   ← JSON이 14배 빠름 😳
유저 50,000명  → JSON 11.37ms / SQLite 3.91ms  ← 역전!

# 이 코드가 왜 위험할까요?
d = json.load(f)              # ① 읽고
d["1"]["7"] += 1              # ② 고치고
json.dump(d, f)               # ③ 쓴다

pip install -U discord.py aiosqlite python-dotenv

python -c "import aiosqlite, discord; print(aiosqlite.__version__, discord.__version__)"

import asyncio
import aiosqlite

async def main():
    async with aiosqlite.connect("data/bot.db") as db:
        await db.execute("CREATE TABLE IF NOT EXISTS memo (text TEXT)")
        await db.execute("INSERT INTO memo VALUES (?)", ("첫 줄!",))
        await db.commit()

        async with db.execute("SELECT text FROM memo") as cursor:
            async for row in cursor:
                print(row[0])

asyncio.run(main())

class Database:
    def __init__(self, path: Path = DB_PATH):
        self.path = path
        self.conn: Optional[aiosqlite.Connection] = None

    async def connect(self) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)   # data/ 폴더 보장
        self.conn = await aiosqlite.connect(self.path)
        self.conn.row_factory = aiosqlite.Row                 # 컬럼명으로 접근
        await self.conn.execute("PRAGMA journal_mode=WAL")
        await self.conn.execute("PRAGMA foreign_keys=ON")
        await self.conn.executescript(SCHEMA)                 # 테이블 자동 생성
        await self.conn.commit()

    async def close(self) -> None:
        if self.conn is not None:
            await self.conn.close()
            self.conn = None

db = Database()      # 모듈 하나에 인스턴스 하나 (Cog 전체가 공유)

class ServerBot(commands.Bot):
    async def setup_hook(self):
        await db.connect()                    # ① Cog보다 먼저!
        for file in sorted(Path("cogs").glob("*.py")):
            await self.load_extension(f"cogs.{file.stem}")

    async def close(self):
        await db.close()                      # ② 끌 때 정리
        await super().close()

CREATE TABLE IF NOT EXISTS warns (
    guild_id   INTEGER NOT NULL,
    user_id    INTEGER NOT NULL,
    count      INTEGER NOT NULL DEFAULT 0,
    updated_at TEXT    NOT NULL DEFAULT (datetime('now')),
    PRIMARY KEY (guild_id, user_id)
);

CREATE TABLE IF NOT EXISTS banned_words (
    guild_id INTEGER NOT NULL,
    word     TEXT    NOT NULL,
    category TEXT,
    added_by INTEGER,
    added_at TEXT    NOT NULL DEFAULT (datetime('now')),
    PRIMARY KEY (guild_id, word)
);

CREATE INDEX IF NOT EXISTS idx_banned_guild ON banned_words (guild_id);

async def add_warn(guild_id: int, user_id: int) -> int:
    await db.execute(
        """
        INSERT INTO warns (guild_id, user_id, count, updated_at)
        VALUES (?, ?, 1, datetime('now'))
        ON CONFLICT(guild_id, user_id)
        DO UPDATE SET count = count + 1, updated_at = datetime('now')
        """,
        (guild_id, user_id),
    )
    row = await db.fetchone(
        "SELECT count FROM warns WHERE guild_id = ? AND user_id = ?",
        (guild_id, user_id),
    )
    return row["count"]

# 변경 전 (#21): JSON 읽고 → 파이썬으로 +1 → 다시 쓰기
# 변경 후
from database import add_warn

count = await add_warn(message.guild.id, message.author.id)
if count >= 3:
    await message.author.timeout(timedelta(minutes=10), reason="경고 3회 누적")

async def main():
    await db.connect()

    wf = Path("data/warns.json")
    if wf.exists():
        data = json.loads(wf.read_text(encoding="utf-8"))
        for guild_id, users in data.items():
            for user_id, count in users.items():
                await db.execute(
                    "INSERT OR REPLACE INTO warns (guild_id, user_id, count) VALUES (?, ?, ?)",
                    (int(guild_id), int(user_id), int(count)),
                )
        wf.rename(wf.with_suffix(".json.bak"))     # 백업으로 이름만 바꿔둬요

async def registered_autocomplete(self, interaction, current: str):
    rows = await db.fetchall(
        """
        SELECT word FROM banned_words
        WHERE guild_id = ? AND word LIKE ?
        ORDER BY added_at LIMIT ?
        """,
        (interaction.guild_id or 0, f"%{current}%", 25),
    )
    return to_choices([row["word"] for row in rows])

# aiosqlite 0.22.x / discord.py 2.x / Python 3.9+
# 디스코드 봇 만들기 #23 - 비동기 DB 계층

import os
from pathlib import Path
from typing import Any, Iterable, List, Optional

import aiosqlite
from dotenv import load_dotenv

load_dotenv()
DB_PATH = Path(os.getenv("DB_PATH", "data/bot.db"))

SCHEMA = """
CREATE TABLE IF NOT EXISTS warns (
    guild_id   INTEGER NOT NULL,
    user_id    INTEGER NOT NULL,
    count      INTEGER NOT NULL DEFAULT 0,
    updated_at TEXT    NOT NULL DEFAULT (datetime('now')),
    PRIMARY KEY (guild_id, user_id)
);

CREATE TABLE IF NOT EXISTS banned_words (
    guild_id INTEGER NOT NULL,
    word     TEXT    NOT NULL,
    category TEXT,
    added_by INTEGER,
    added_at TEXT    NOT NULL DEFAULT (datetime('now')),
    PRIMARY KEY (guild_id, word)
);

CREATE INDEX IF NOT EXISTS idx_banned_guild ON banned_words (guild_id);
"""

class Database:
    """커넥션 하나를 봇 전체가 공유하는 얇은 래퍼"""

    def __init__(self, path: Path = DB_PATH):
        self.path = path
        self.conn: Optional[aiosqlite.Connection] = None

    async def connect(self) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self.conn = await aiosqlite.connect(self.path)
        self.conn.row_factory = aiosqlite.Row      # row["컬럼명"] 으로 접근
        await self.conn.execute("PRAGMA journal_mode=WAL")
        await self.conn.execute("PRAGMA foreign_keys=ON")
        await self.conn.executescript(SCHEMA)
        await self.conn.commit()

    async def close(self) -> None:
        if self.conn is not None:
            await self.conn.close()
            self.conn = None

    async def execute(self, sql: str, params: Iterable[Any] = ()) -> None:
        await self.conn.execute(sql, tuple(params))
        await self.conn.commit()

    async def fetchone(self, sql: str, params: Iterable[Any] = ()):
        async with self.conn.execute(sql, tuple(params)) as cursor:
            return await cursor.fetchone()

    async def fetchall(self, sql: str, params: Iterable[Any] = ()) -> List[aiosqlite.Row]:
        async with self.conn.execute(sql, tuple(params)) as cursor:
            return await cursor.fetchall()

db = Database()

# ---------- 경고 (#21 automod 에서 사용) ----------
async def add_warn(guild_id: int, user_id: int) -> int:
    """경고를 1 올리고 누적 횟수를 돌려줘요. UPSERT라 동시 호출에도 안전해요."""
    await db.execute(
        """
        INSERT INTO warns (guild_id, user_id, count, updated_at)
        VALUES (?, ?, 1, datetime('now'))
        ON CONFLICT(guild_id, user_id)
        DO UPDATE SET count = count + 1, updated_at = datetime('now')
        """,
        (guild_id, user_id),
    )
    row = await db.fetchone(
        "SELECT count FROM warns WHERE guild_id = ? AND user_id = ?",
        (guild_id, user_id),
    )
    return row["count"]

async def reset_warn(guild_id: int, user_id: int) -> None:
    await db.execute(
        "DELETE FROM warns WHERE guild_id = ? AND user_id = ?", (guild_id, user_id)
    )

# ---------- 금지어 (#22 wordadmin 에서 사용) ----------
async def get_words(guild_id: int) -> List[str]:
    rows = await db.fetchall(
        "SELECT word FROM banned_words WHERE guild_id = ? ORDER BY added_at",
        (guild_id,),
    )
    return [row["word"] for row in rows]

async def add_word(guild_id: int, word: str, category=None, added_by=None) -> bool:
    """추가 성공하면 True, 이미 있으면 False (중복 검사는 DB가 대신 해줘요)"""
    try:
        await db.execute(
            "INSERT INTO banned_words (guild_id, word, category, added_by) "
            "VALUES (?, ?, ?, ?)",
            (guild_id, word, category, added_by),
        )
        return True
    except aiosqlite.IntegrityError:
        return False

async def remove_word(guild_id: int, word: str) -> bool:
    before = await get_words(guild_id)
    await db.execute(
        "DELETE FROM banned_words WHERE guild_id = ? AND word = ?", (guild_id, word)
    )
    return word in before

# discord.py 2.x / aiosqlite 0.22.x / Python 3.9+
import os
from pathlib import Path

import discord
from discord.ext import commands
from dotenv import load_dotenv

from database import db

load_dotenv()
TOKEN = os.getenv("token")          # .env 에서 봇 토큰 로딩 (하드코딩 금지!)

intents = discord.Intents.default()
intents.message_content = True
intents.members = True

class ServerBot(commands.Bot):
    def __init__(self):
        super().__init__(command_prefix="!", intents=intents)

    async def setup_hook(self):
        await db.connect()                       # ① DB 먼저
        print("🗄️ DB 연결 완료")
        for file in sorted(Path("cogs").glob("*.py")):
            await self.load_extension(f"cogs.{file.stem}")
            print(f"✅ cogs.{file.stem} 불러옴")

    async def close(self):
        await db.close()                         # ② 끌 때 정리
        print("🗄️ DB 연결 종료")
        await super().close()

bot = ServerBot()

@bot.event
async def on_ready():
    print(f"{bot.user} 로그인 성공!")

if __name__ == "__main__":
    bot.run(TOKEN)

token=여기에_봇_토큰
DB_PATH=data/bot.db

# 기존 JSON 데이터를 SQLite로 옮기는 1회용 스크립트
# 실행: python migrate.py

import asyncio
import json
from pathlib import Path

from database import db, add_word

async def main():
    await db.connect()

    # 1) 경고 기록 (#21)
    warn_file = Path("data/warns.json")
    if warn_file.exists():
        data = json.loads(warn_file.read_text(encoding="utf-8"))
        for guild_id, users in data.items():
            for user_id, count in users.items():
                await db.execute(
                    "INSERT OR REPLACE INTO warns (guild_id, user_id, count) "
                    "VALUES (?, ?, ?)",
                    (int(guild_id), int(user_id), int(count)),
                )
        warn_file.rename(warn_file.with_suffix(".json.bak"))
        print("✅ warns.json 옮김")

    # 2) 금지어 목록 (#22)
    word_file = Path("data/banned_words.json")
    if word_file.exists():
        data = json.loads(word_file.read_text(encoding="utf-8"))
        for guild_id, words in data.items():
            for word in words:
                await add_word(int(guild_id), word, category="이전됨")
        word_file.rename(word_file.with_suffix(".json.bak"))
        print("✅ banned_words.json 옮김")

    rows = await db.fetchall("SELECT * FROM warns")
    print(f"📊 경고 {len(rows)}건 이전 완료")
    await db.close()

if __name__ == "__main__":
    asyncio.run(main())

server-bot/
├─ .env                     # token, DB_PATH
├─ .gitignore               # data/*.db, *.db-wal, *.db-shm, .env
├─ bot.py                   # setup_hook 에서 DB 연결 (오늘 수정)
├─ database.py              # ← 오늘 추가!
├─ migrate.py               # ← 오늘 추가 (1회용)
├─ data/
│  ├─ bot.db                # ← 자동 생성
│  ├─ warns.json.bak        # 백업본
│  └─ banned_words.json.bak
└─ cogs/
   ├─ poll.py               # #18
   ├─ onboarding.py         # #20
   ├─ automod.py            # #21 (add_warn 으로 교체)
   └─ wordadmin.py          # #22 (DB LIKE 검색으로 교체)

$ python migrate.py
✅ warns.json 옮김
✅ banned_words.json 옮김
📊 경고 1건 이전 완료

$ python bot.py
🗄️ DB 연결 완료
✅ cogs.automod 불러옴
✅ cogs.onboarding 불러옴
✅ cogs.poll 불러옴
✅ cogs.wordadmin 불러옴
MojangBot#1234 로그인 성공!

JSON  결과 (기대 100): 1
SQLite 결과 (기대 100): 100