PySide6 実践講座 #2 SQLite データ層 — リポジトリパターンとマイグレーション

# src/daily/data/paths.py
from pathlib import Path

from PySide6.QtCore import QStandardPaths


def db_path() -> Path:
    base = QStandardPaths.writableLocation(
        QStandardPaths.StandardLocation.AppDataLocation
    )
    folder = Path(base)
    folder.mkdir(parents=True, exist_ok=True)
    return folder / "daily.db"

CREATE TABLE habits (
    id         INTEGER PRIMARY KEY,
    name       TEXT NOT NULL,
    created_at TEXT NOT NULL,          -- ISO 日付文字列
    archived   INTEGER NOT NULL DEFAULT 0
);

CREATE TABLE checks (
    habit_id INTEGER NOT NULL REFERENCES habits(id) ON DELETE CASCADE,
    day      TEXT NOT NULL,            -- 'YYYY-MM-DD'
    PRIMARY KEY (habit_id, day)
);

# src/daily/data/repository.py
import sqlite3
from datetime import date
from pathlib import Path

from daily.core.habits import Habit

_SCHEMA_VERSION = 1


class HabitRepository:
    def __init__(self, path: Path) -> None:
        self._conn = sqlite3.connect(path)
        self._conn.execute("PRAGMA foreign_keys = ON")
        self._migrate()

    # --- 習慣 ---
    def add_habit(self, name: str) -> int:
        cur = self._conn.execute(
            "INSERT INTO habits (name, created_at) VALUES (?, ?)",
            (name, date.today().isoformat()),
        )
        self._conn.commit()
        return cur.lastrowid

    def active_habits(self) -> list[Habit]:
        rows = self._conn.execute(
            "SELECT id, name FROM habits WHERE archived = 0 ORDER BY id"
        ).fetchall()
        return [Habit(id=r[0], name=r[1]) for r in rows]

    def archive_habit(self, habit_id: int) -> None:
        self._conn.execute("UPDATE habits SET archived = 1 WHERE id = ?", (habit_id,))
        self._conn.commit()

    # --- チェック ---
    def set_checked(self, habit_id: int, day: date, checked: bool) -> None:
        if checked:
            self._conn.execute(
                "INSERT OR IGNORE INTO checks (habit_id, day) VALUES (?, ?)",
                (habit_id, day.isoformat()),
            )
        else:
            self._conn.execute(
                "DELETE FROM checks WHERE habit_id = ? AND day = ?",
                (habit_id, day.isoformat()),
            )
        self._conn.commit()

    def checked_days(self, habit_id: int) -> set[str]:
        rows = self._conn.execute(
            "SELECT day FROM checks WHERE habit_id = ?", (habit_id,)
        ).fetchall()
        return {r[0] for r in rows}

    def _migrate(self) -> None:
        version = self._conn.execute("PRAGMA user_version").fetchone()[0]
        if version < 1:
            self._conn.executescript("""
                CREATE TABLE habits (
                    id INTEGER PRIMARY KEY,
                    name TEXT NOT NULL,
                    created_at TEXT NOT NULL,
                    archived INTEGER NOT NULL DEFAULT 0
                );
                CREATE TABLE checks (
                    habit_id INTEGER NOT NULL REFERENCES habits(id) ON DELETE CASCADE,
                    day TEXT NOT NULL,
                    PRIMARY KEY (habit_id, day)
                );
            """)
        # if version < 2: ...  次のスキーマ変更はここに追加
        self._conn.execute(f"PRAGMA user_version = {_SCHEMA_VERSION}")
        self._conn.commit()

# app.py の main() に追加
from daily.data.paths import db_path
from daily.data.repository import HabitRepository

repo = HabitRepository(db_path())
window = MainWindow(repo)   # MainWindow が repo を受け取り、各画面に配ります