blog-fuwari/feedback-api/database.py

import aiosqlite

DB_PATH = "feedback.db"


async def init_db():
    async with aiosqlite.connect(DB_PATH) as db:
        await db.execute("""
            CREATE TABLE IF NOT EXISTS feedbacks (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                url TEXT NOT NULL,
                choice TEXT NOT NULL CHECK(choice IN ('yes', 'no')),
                timestamp TEXT NOT NULL,
                ip_hash TEXT NOT NULL,
                created_at TEXT NOT NULL DEFAULT (datetime('now'))
            )
        """)
        await db.execute("""
            CREATE INDEX IF NOT EXISTS idx_feedbacks_ip_hash
            ON feedbacks(ip_hash, created_at)
        """)
        await db.commit()


async def add_feedback(ip_hash: str, url: str, choice: str, timestamp: str):
    async with aiosqlite.connect(DB_PATH) as db:
        await db.execute(
            "INSERT INTO feedbacks (url, choice, timestamp, ip_hash) VALUES (?, ?, ?, ?)",
            (url, choice, timestamp, ip_hash),
        )
        await db.commit()


async def check_rate_limit(ip_hash: str, limit: int = 5, window_seconds: int = 3600) -> bool:
    async with aiosqlite.connect(DB_PATH) as db:
        cursor = await db.execute(
            "SELECT COUNT(*) FROM feedbacks WHERE ip_hash = ? AND created_at > datetime('now', ?)",
            (ip_hash, f"-{window_seconds} seconds"),
        )
        row = await cursor.fetchone()
        return row[0] >= limit


async def get_stats() -> dict:
    async with aiosqlite.connect(DB_PATH) as db:
        cursor = await db.execute("SELECT COUNT(*) FROM feedbacks")
        total = (await cursor.fetchone())[0]

        cursor = await db.execute("SELECT choice, COUNT(*) FROM feedbacks GROUP BY choice")
        rows = await cursor.fetchall()
        counts = {row[0]: row[1] for row in rows}

        return {
            "total": total,
            "yes": counts.get("yes", 0),
            "no": counts.get("no", 0),
        }