中級者向け No.45

DBクエリビルダー

SQLite向けにSQLを編集・実行できるクエリツール。構文ハイライト・スキーマ一覧・サンプルSQL挿入で手早く書けます。

🎯 難易度: ★★★ 📦 ライブラリ: tkinter（標準ライブラリ） ⏱️ 制作時間: 30〜90分

📋 目次

1. アプリ概要
2. 機能一覧
3. 事前準備・環境
4. 完全なソースコード
5. コード解説
6. ステップバイステップガイド
7. カスタマイズアイデア
8. よくある問題と解決法
9. 練習問題

1. アプリ概要

SQLite向けにSQLを編集・実行できるクエリツール。構文ハイライト・スキーマ一覧・サンプルSQL挿入で手早く書けます。

このアプリは中級カテゴリに分類される実践的なGUIアプリです。使用ライブラリは tkinter（標準ライブラリ） で、難易度は ★★★ です。

このアプリは「データベース連携」カテゴリです。データベース連携 GUI は永続化と画面更新を結びつける典型例で、トランザクション・スキーマ設計・UI 反映の三点を一度に学べます。tkinter（標準ライブラリ） を活かして実装するこの構造は、他のアプリにも応用が効きます。

コードを読む前に実行することをおすすめします。動いている挙動を先に把握しておくと、解説で出てくる関数や処理がどこに対応するかが頭に入りやすくなります。

応用のヒントは、機能を 1 つ増やす・見た目を整える・例外を一つでも丁寧に扱う、のいずれかから始めるのがおすすめです。

DBクエリビルダー実行画面（Linux Mint） — 実行画面（Linux Mint）

2. 機能一覧

「📂 開く」ボタン（_open_db()）・「💾 新規作成」ボタン（_new_db()）・「🔌 接続」ボタン（_connect_db()）・「🔄 スキーマ更新」ボタン（_refresh_schema()）・「▶ 実行 (F5)」ボタン（_execute()）・「✨ フォーマット」ボタン（_format_sql()）・「🗑 クリア」ボタン（lambda: (self.editor.delete('1.0', tk.END), self._update_line_nums())）で操作
マウス操作 <Double-1> から _on_schema_click() を実行
イベント <<ComboboxSelected>> から _load_sample() を実行
キー <KeyRelease> から lambda e: (self._update_line_nums(), self._highlight_sql()) を実行
キー <F5> から _execute() を実行
ttk.Treeview による表形式の一覧表示
sqlite3.connect() による SQLite データベース接続と SQL 実行
filedialog.askopenfilename() によるファイル選択ダイアログ
filedialog.asksaveasfilename() によるファイル選択ダイアログ
messagebox.showerror()・messagebox.showwarning() によるダイアログ通知
ウィンドウはタイトル「DBクエリビルダー」・サイズ 1060x1006 で起動

3. 事前準備・環境

ℹ️

動作確認環境

Python 3.10 以上 / Windows 11（実機）で起動・基本操作を確認／ Linux Mint 22.3（仮想マシン）で起動・画面表示を確認（macOS は未検証）

Windows 11（実機）で起動と基本操作を確認しています（全機能の網羅テストではありません）。Linux Mint 22.3（仮想マシン）では起動と画面表示を確認しました。macOS は標準ライブラリの範囲で動作する想定ですが、未検証です。

Python 3.10 以上
OS: Windows 11（実機で起動・基本操作を確認）・Linux Mint 22.3（仮想マシンで起動・画面表示を確認）

4. 完全なソースコード

💡

コードのコピー方法

右上の「コピー」ボタンをクリックするとコードをクリップボードにコピーできます。

✅ 追加インストール不要（標準ライブラリのみ使用）

app45.py

import tkinter as tk
from tkinter import ttk, messagebox, filedialog
import sqlite3
import os


class App45:
    """DBクエリビルダー"""

    SAMPLE_SQL = [
        ("全レコード取得", "SELECT * FROM {table} LIMIT 100;"),
        ("件数カウント", "SELECT COUNT(*) FROM {table};"),
        ("列を指定して取得", "SELECT {cols} FROM {table} WHERE 条件 LIMIT 50;"),
        ("グループ集計", "SELECT category, COUNT(*), SUM(amount) FROM {table} GROUP BY category ORDER BY 2 DESC;"),
        ("テーブル一覧", "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name;"),
        ("テーブル定義", "PRAGMA table_info({table});"),
        ("INSERT", "INSERT INTO {table} ({cols}) VALUES ({vals});"),
        ("UPDATE", "UPDATE {table} SET column = 'value' WHERE id = 1;"),
        ("DELETE", "DELETE FROM {table} WHERE id = 1;"),
    ]

    def __init__(self, root):
        self.root = root
        self.root.title("DBクエリビルダー")
        self.root.geometry("1060x1006")
        self.root.minsize(1060, 1006)
        self.root.configure(bg="#1e1e1e")
        self._conn = None
        self._db_path = None
        self._history = []
        self._build_ui()

    def _build_ui(self):
        # ヘッダー
        header = tk.Frame(self.root, bg="#252526", pady=6)
        header.pack(fill=tk.X)
        tk.Label(header, text="🔍 DBクエリビルダー",
                 font=("Noto Sans JP", 12, "bold"),
                 bg="#252526", fg="#4fc3f7").pack(side=tk.LEFT, padx=12)

        # DB接続バー
        conn_f = tk.Frame(header, bg="#252526")
        conn_f.pack(side=tk.LEFT, padx=12)
        self.db_path_var = tk.StringVar(value="新規メモリDB")
        ttk.Entry(conn_f, textvariable=self.db_path_var,
                  width=36, font=("Arial", 10)).pack(side=tk.LEFT, padx=4)
        ttk.Button(conn_f, text="📂 開く",
                   command=self._open_db).pack(side=tk.LEFT, padx=2)
        ttk.Button(conn_f, text="💾 新規作成",
                   command=self._new_db).pack(side=tk.LEFT, padx=2)
        ttk.Button(conn_f, text="🔌 接続",
                   command=self._connect_db).pack(side=tk.LEFT, padx=2)

        # メインエリア
        main_paned = ttk.PanedWindow(self.root, orient=tk.HORIZONTAL)
        main_paned.pack(fill=tk.BOTH, expand=True, padx=4, pady=4)

        # 左パネル: スキーマ
        left = tk.Frame(main_paned, bg="#252526")
        main_paned.add(left, weight=1)
        self._build_schema_panel(left)

        # 中央: エディタ + 結果
        center = ttk.PanedWindow(main_paned, orient=tk.VERTICAL)
        main_paned.add(center, weight=4)
        self._build_editor_panel(center)
        self._build_result_panel(center)

        self.status_var = tk.StringVar(value="DBを開いてください")
        tk.Label(self.root, textvariable=self.status_var,
                 bg="#252526", fg="#858585", font=("Arial", 9),
                 anchor="w", padx=8).pack(fill=tk.X, side=tk.BOTTOM)

        # 自動接続（メモリDB）
        self._connect_memory()

    def _build_schema_panel(self, parent):
        tk.Label(parent, text="スキーマ", bg="#252526", fg="#858585",
                 font=("Arial", 9)).pack(anchor="w", padx=6, pady=2)

        # テーブルツリー
        self.schema_tree = ttk.Treeview(parent, show="tree", height=20)
        sb = ttk.Scrollbar(parent, command=self.schema_tree.yview)
        self.schema_tree.configure(yscrollcommand=sb.set)
        sb.pack(side=tk.RIGHT, fill=tk.Y)
        self.schema_tree.pack(fill=tk.BOTH, expand=True, padx=4)
        self.schema_tree.bind("<Double-1>", self._on_schema_click)

        ttk.Button(parent, text="🔄 スキーマ更新",
                   command=self._refresh_schema).pack(pady=4)

    def _build_editor_panel(self, parent):
        editor_f = tk.Frame(parent, bg="#1e1e1e")
        parent.add(editor_f, weight=2)

        # ツールバー
        bar = tk.Frame(editor_f, bg="#252526")
        bar.pack(fill=tk.X)
        ttk.Button(bar, text="▶ 実行 (F5)",
                   command=self._execute).pack(side=tk.LEFT, padx=4, pady=4)
        ttk.Button(bar, text="✨ フォーマット",
                   command=self._format_sql).pack(side=tk.LEFT, padx=4)
        ttk.Button(bar, text="🗑 クリア",
                   command=lambda: (
                       self.editor.delete("1.0", tk.END),
                       self._update_line_nums())).pack(side=tk.LEFT, padx=4)
        tk.Label(bar, text="サンプル:", bg="#252526", fg="#ccc",
                 font=("Arial", 9)).pack(side=tk.LEFT, padx=(12, 4))
        self.sample_var = tk.StringVar()
        sample_cb = ttk.Combobox(bar, textvariable=self.sample_var,
                                  values=[s[0] for s in self.SAMPLE_SQL],
                                  state="readonly", width=18)
        sample_cb.pack(side=tk.LEFT)
        sample_cb.bind("<<ComboboxSelected>>", self._load_sample)

        # 行番号 + エディタ
        num_f = tk.Frame(editor_f, bg="#1e1e1e")
        num_f.pack(fill=tk.BOTH, expand=True)
        self.line_nums = tk.Text(num_f, width=4, bg="#252526", fg="#858585",
                                  font=("Courier New", 12), state=tk.DISABLED,
                                  relief=tk.FLAT, padx=4, pady=4, takefocus=0)
        self.line_nums.pack(side=tk.LEFT, fill=tk.Y)
        self.editor = tk.Text(num_f, bg="#1e1e1e", fg="#d4d4d4",
                               font=("Courier New", 12), insertbackground="#aeafad",
                               relief=tk.FLAT, undo=True, padx=8, pady=4,
                               wrap=tk.NONE)
        ed_h_sb = ttk.Scrollbar(num_f, orient=tk.HORIZONTAL,
                                command=self.editor.xview)
        ed_v_sb = ttk.Scrollbar(num_f, orient=tk.VERTICAL,
                                command=self.editor.yview)
        self.editor.configure(xscrollcommand=ed_h_sb.set,
                               yscrollcommand=ed_v_sb.set)
        ed_v_sb.pack(side=tk.RIGHT, fill=tk.Y)
        ed_h_sb.pack(side=tk.BOTTOM, fill=tk.X)
        self.editor.pack(fill=tk.BOTH, expand=True)
        self.editor.bind("<KeyRelease>", lambda e: (self._update_line_nums(),
                                                     self._highlight_sql()))
        self.root.bind("<F5>", lambda e: self._execute())
        self._setup_sql_tags()
        self.editor.insert("1.0", "SELECT * FROM sqlite_master WHERE type='table';")
        self._update_line_nums()
        self._highlight_sql()

    def _build_result_panel(self, parent):
        result_f = tk.Frame(parent, bg="#1e1e1e")
        parent.add(result_f, weight=3)

        result_nb = ttk.Notebook(result_f)
        result_nb.pack(fill=tk.BOTH, expand=True)

        # 結果グリッドタブ
        grid_tab = tk.Frame(result_nb, bg="#1e1e1e")
        result_nb.add(grid_tab, text="結果")

        self.result_tree = ttk.Treeview(grid_tab, show="headings")
        h_sb = ttk.Scrollbar(grid_tab, orient=tk.HORIZONTAL,
                             command=self.result_tree.xview)
        v_sb = ttk.Scrollbar(grid_tab, orient=tk.VERTICAL,
                             command=self.result_tree.yview)
        self.result_tree.configure(xscrollcommand=h_sb.set,
                                    yscrollcommand=v_sb.set)
        v_sb.pack(side=tk.RIGHT, fill=tk.Y)
        h_sb.pack(side=tk.BOTTOM, fill=tk.X)
        self.result_tree.pack(fill=tk.BOTH, expand=True)

        # 履歴タブ
        hist_tab = tk.Frame(result_nb, bg="#1e1e1e")
        result_nb.add(hist_tab, text="実行履歴")
        self.history_text = tk.Text(hist_tab, bg="#0d1117", fg="#c9d1d9",
                                     font=("Courier New", 10), state=tk.DISABLED)
        self.history_text.pack(fill=tk.BOTH, expand=True)
        self.history_text.tag_configure("success", foreground="#3fb950")
        self.history_text.tag_configure("error", foreground="#f97583")
        self.history_text.tag_configure("sql", foreground="#569cd6")

        # エラータブ
        err_tab = tk.Frame(result_nb, bg="#1e1e1e")
        result_nb.add(err_tab, text="エラー")
        self.error_text = tk.Text(err_tab, bg="#2b0f0f", fg="#f97583",
                                   font=("Courier New", 11), state=tk.DISABLED)
        self.error_text.pack(fill=tk.BOTH, expand=True)

    def _setup_sql_tags(self):
        keywords = {
            "keyword": "#569cd6",
            "string": "#ce9178",
            "number": "#b5cea8",
            "comment": "#6a9955",
            "function": "#dcdcaa",
        }
        for tag, color in keywords.items():
            self.editor.tag_configure(tag, foreground=color)

    def _highlight_sql(self):
        import re
        content = self.editor.get("1.0", tk.END)
        for tag in ["keyword", "string", "number", "comment", "function"]:
            self.editor.tag_remove(tag, "1.0", tk.END)
        SQL_KEYWORDS = (
            r"\b(SELECT|FROM|WHERE|JOIN|LEFT|RIGHT|INNER|OUTER|ON|AND|OR|NOT|"
            r"INSERT|INTO|VALUES|UPDATE|SET|DELETE|CREATE|TABLE|INDEX|DROP|ALTER|"
            r"ADD|COLUMN|PRIMARY|KEY|FOREIGN|REFERENCES|UNIQUE|NULL|NOT NULL|"
            r"GROUP BY|ORDER BY|HAVING|LIMIT|OFFSET|AS|DISTINCT|IN|LIKE|"
            r"BETWEEN|EXISTS|UNION|ALL|PRAGMA|IF|EXISTS|DEFAULT|INTEGER|TEXT|"
            r"REAL|BLOB|AUTOINCREMENT|COALESCE|CASE|WHEN|THEN|ELSE|END)\b"
        )
        patterns = [
            ("keyword", SQL_KEYWORDS),
            ("string", r"'[^']*'"),
            ("number", r"\b\d+(?:\.\d+)?\b"),
            ("comment", r"--[^\n]*"),
            ("function", r"\b(COUNT|SUM|AVG|MAX|MIN|LENGTH|SUBSTR|TRIM|UPPER|LOWER|"
                          r"DATE|DATETIME|STRFTIME|REPLACE|ROUND|ABS|COALESCE)\b"),
        ]
        for tag, pattern in patterns:
            for m in re.finditer(pattern, content, re.IGNORECASE):
                self.editor.tag_add(tag,
                                     f"1.0+{m.start()}c",
                                     f"1.0+{m.end()}c")

    def _update_line_nums(self):
        content = self.editor.get("1.0", tk.END)
        lines = content.count("\n")
        self.line_nums.config(state=tk.NORMAL)
        self.line_nums.delete("1.0", tk.END)
        self.line_nums.insert("1.0", "\n".join(str(i) for i in range(1, lines + 1)))
        self.line_nums.config(state=tk.DISABLED)

    # ── DB操作 ──────────────────────────────────────────────────

    def _open_db(self):
        path = filedialog.askopenfilename(
            filetypes=[("SQLite DB", "*.db *.sqlite *.sqlite3"),
                       ("すべて", "*.*")])
        if path:
            self.db_path_var.set(path)
            self._connect_db()

    def _new_db(self):
        path = filedialog.asksaveasfilename(
            defaultextension=".db",
            filetypes=[("SQLite DB", "*.db"), ("すべて", "*.*")])
        if path:
            self.db_path_var.set(path)
            self._connect_db()

    def _connect_memory(self):
        try:
            self._conn = sqlite3.connect(":memory:")
            # サンプルテーブル作成
            self._conn.executescript("""
                CREATE TABLE products (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    name TEXT, category TEXT, price REAL, stock INTEGER
                );
                INSERT INTO products (name, category, price, stock) VALUES
                ('リンゴ', '果物', 150, 50),
                ('バナナ', '果物', 80, 100),
                ('キャベツ', '野菜', 120, 30),
                ('トマト', '野菜', 200, 20),
                ('オレンジ', '果物', 180, 45);
                CREATE TABLE sales (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    product_id INTEGER, quantity INTEGER,
                    sale_date TEXT, amount REAL
                );
                INSERT INTO sales (product_id, quantity, sale_date, amount) VALUES
                (1, 5, '2025-04-01', 750),
                (2, 10, '2025-04-01', 800),
                (3, 3, '2025-04-02', 360),
                (1, 2, '2025-04-03', 300);
            """)
            self.status_var.set("インメモリDBに接続 (サンプルデータあり)")
            self._refresh_schema()
        except Exception as e:
            self.status_var.set(f"エラー: {e}")

    def _connect_db(self):
        path = self.db_path_var.get().strip()
        if not path or path == "新規メモリDB":
            self._connect_memory()
            return
        try:
            if self._conn:
                self._conn.close()
            self._conn = sqlite3.connect(path)
            self._db_path = path
            self.status_var.set(f"接続: {path}")
            self._refresh_schema()
        except Exception as e:
            messagebox.showerror("接続エラー", str(e))

    def _refresh_schema(self):
        if not self._conn:
            return
        self.schema_tree.delete(*self.schema_tree.get_children())
        try:
            tables = self._conn.execute(
                "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
            ).fetchall()
            for (tbl,) in tables:
                node = self.schema_tree.insert("", "end", text=f"📋 {tbl}", iid=tbl)
                cols = self._conn.execute(f"PRAGMA table_info({tbl})").fetchall()
                for col in cols:
                    cid, cname, ctype, notnull, dflt, pk = col
                    pk_str = " 🔑" if pk else ""
                    nn_str = " NOT NULL" if notnull else ""
                    self.schema_tree.insert(node, "end",
                                             text=f"  {cname} ({ctype}{nn_str}{pk_str})")
        except Exception as e:
            self.status_var.set(f"スキーマ取得エラー: {e}")

    def _on_schema_click(self, event):
        sel = self.schema_tree.selection()
        if sel:
            iid = sel[0]
            # テーブルノードの場合
            tables = [self.schema_tree.item(n)["text"].strip().lstrip("📋 ")
                      for n in self.schema_tree.get_children()]
            tbl = self.schema_tree.item(iid)["text"].strip().lstrip("📋 ")
            if tbl in tables:
                sql = f"SELECT * FROM {tbl} LIMIT 50;"
                self.editor.delete("1.0", tk.END)
                self.editor.insert("1.0", sql)
                self._update_line_nums()
                self._highlight_sql()
                self._execute()

    def _execute(self):
        if not self._conn:
            messagebox.showwarning("警告", "DBに接続してください")
            return
        sql = self.editor.get("1.0", tk.END).strip()
        if not sql:
            return
        try:
            # 複数のステートメントがある場合は最後のSELECTを実行
            statements = [s.strip() for s in sql.split(";") if s.strip()]
            last_select = None
            for stmt in statements:
                if stmt.upper().startswith("SELECT") or stmt.upper().startswith("PRAGMA"):
                    last_select = stmt
                else:
                    self._conn.execute(stmt)
                    self._conn.commit()

            if last_select:
                cursor = self._conn.execute(last_select)
                rows = cursor.fetchall()
                cols = [d[0] for d in cursor.description] if cursor.description else []
                self._show_results(cols, rows)
                self.status_var.set(f"✅ {len(rows)} 件取得")
                self._add_history(last_select, True, f"{len(rows)} 件")
            else:
                self.status_var.set("✅ 実行完了")
                self._add_history(sql, True, "実行完了")
                self._refresh_schema()

            # エラーテキストをクリア
            self.error_text.config(state=tk.NORMAL)
            self.error_text.delete("1.0", tk.END)
            self.error_text.config(state=tk.DISABLED)
        except Exception as e:
            self.status_var.set(f"❌ エラー: {e}")
            self._add_history(sql, False, str(e))
            self.error_text.config(state=tk.NORMAL)
            self.error_text.delete("1.0", tk.END)
            self.error_text.insert("1.0", str(e))
            self.error_text.config(state=tk.DISABLED)

    def _show_results(self, cols, rows):
        self.result_tree.delete(*self.result_tree.get_children())
        if not cols:
            return
        self.result_tree["columns"] = cols
        for col in cols:
            self.result_tree.heading(col, text=col)
            self.result_tree.column(col, width=max(80, len(col) * 10),
                                     minwidth=40)
        for row in rows:
            self.result_tree.insert("", "end",
                                     values=[str(v) if v is not None else "NULL"
                                             for v in row])

    def _add_history(self, sql, success, msg):
        self.history_text.config(state=tk.NORMAL)
        import datetime
        ts = datetime.datetime.now().strftime("%H:%M:%S")
        tag = "success" if success else "error"
        self.history_text.insert("1.0", f"[{ts}] {'✅' if success else '❌'} {msg}\n", tag)
        self.history_text.insert("1.0", "", "")
        self.history_text.insert("2.0", f"{sql[:200]}\n\n", "sql")
        self.history_text.config(state=tk.DISABLED)

    def _format_sql(self):
        sql = self.editor.get("1.0", tk.END).strip()
        keywords = ["SELECT", "FROM", "WHERE", "AND", "OR", "JOIN", "LEFT",
                    "INNER", "GROUP BY", "ORDER BY", "HAVING", "LIMIT",
                    "INSERT INTO", "VALUES", "UPDATE", "SET", "DELETE FROM"]
        import re
        formatted = sql.upper()
        for kw in sorted(keywords, key=len, reverse=True):
            formatted = re.sub(r'\b' + kw + r'\b', "\n" + kw, formatted,
                               flags=re.IGNORECASE)
        # 元の大小文字を保持（簡易版）
        self.editor.delete("1.0", tk.END)
        self.editor.insert("1.0", formatted.strip())
        self._update_line_nums()
        self._highlight_sql()

    def _load_sample(self, event):
        name = self.sample_var.get()
        sample = next((s for s in self.SAMPLE_SQL if s[0] == name), None)
        if sample:
            sql = sample[1].replace("{table}", "products").replace(
                "{cols}", "name, price").replace("{vals}", "'商品名', 1000")
            self.editor.delete("1.0", tk.END)
            self.editor.insert("1.0", sql)
            self._update_line_nums()
            self._highlight_sql()


if __name__ == "__main__":
    root = tk.Tk()
    app = App45(root)
    root.mainloop()

5. コード解説

DBクエリビルダーのコードを、実際に書かれている実装に沿って解説します。

クラス設計とコンストラクタ

App45 クラスにアプリの全機能をまとめています（メソッド19個・全426行）。__init__ ではタイトル「DBクエリビルダー」とウィンドウサイズ 1060x1006 を設定します。あわせて状態を保持する変数（self._conn・self._db_path）を初期化し、最後に _build_ui() で画面を組み立てます。