第8章 PythonからDBを使う

8.1 sqlite3モジュールの基本

PythonからSQLiteを使うのに、追加のインストールは一切不要です。sqlite3 モジュールは標準ライブラリに含まれています(第1章でSQLiteを選んだ理由のひとつがこれでした)。基本の流れは、どのデータベース・どの言語でもほぼ同じで、次の5ステップです。

1. connect — データベースに接続する
2. cursor — SQLを実行するための「カーソル」を得る
3. execute — SQLを実行する
4. fetchall / fetchone — 結果を受け取る(SELECTの場合)
5. close — 接続を閉じる

第3章から使っている company.db を読んでみましょう。まずは対話モードで1行ずつ確かめます。

$ python3
>>> import sqlite3
>>> conn = sqlite3.connect("company.db")
>>> cur = conn.cursor()
>>> cur.execute("SELECT name, salary FROM employees WHERE salary >= 300000")
<sqlite3.Cursor object at 0x7f3a2c1b8d40>
>>> cur.fetchall()
[('佐藤 花子', 380000), ('鈴木 一郎', 320000), ('高橋 美咲', 350000), ('伊藤 健', 300000)]
>>> conn.close()

結果はタプルのリストで返ってきます。1行が1タプル、列の並びはSELECTで書いた順です。スクリプトにするとこうなります。

import sqlite3

conn = sqlite3.connect("company.db")
try:
    cur = conn.cursor()
    cur.execute("SELECT id, name, salary FROM employees ORDER BY salary DESC")
    for emp_id, name, salary in cur.fetchall():
        print(f"{emp_id}: {name} ({salary:,}円)")
finally:
    conn.close()  # 何があっても必ず閉じる

try / finally を毎回書くのは面倒なので、実務では contextlib.closing と with文を組み合わせて「ブロックを抜けたら自動で閉じる」書き方がよく使われます。

import sqlite3
from contextlib import closing

with closing(sqlite3.connect("company.db")) as conn:
    cur = conn.cursor()
    cur.execute("SELECT COUNT(*) FROM employees")
    print(cur.fetchone()[0])  # → 6
# ここに来た時点で conn.close() 済み

8.2 プレースホルダは絶対条件

ここが本章でいちばん重要な節です。SQLに「ユーザーが入力した値」を組み込むとき、f文字列や文字列連結を使ってはいけません。まず、やってはいけない例から見ます。

# ダメな例: f文字列でSQLを組み立てている
name = input("社員名: ")
cur.execute(f"SELECT * FROM employees WHERE name = '{name}'")  # 脆弱!

正しい書き方はプレースホルダです。SQL文の中に ? を置き、値は第2引数のタプルで渡します。

# 正しい例: ? プレースホルダ
name = input("社員名: ")
cur.execute("SELECT * FROM employees WHERE name = ?", (name,))

# 複数の値も同じ。? の数とタプルの要素数を合わせる
cur.execute(
    "SELECT name, salary FROM employees WHERE dept_id = ? AND salary >= ?",
    (20, 300000),
)

プレースホルダを使うと、値は「SQLの一部」ではなく「ただのデータ」としてDBMSに渡されます。入力に ' や OR が含まれていても、それは単なる文字列として検索されるだけで、SQL文の構造を変えることはできません。これがSQLインジェクションの根本的な対策です。Python × セキュリティ入門コースの第10章でSQLインジェクションの攻撃例を学んだ人は、「あの攻撃はプレースホルダで無力化できる」とつなげて覚えてください。攻撃と対策の全体像は次の第9章で改めて扱います。

8.3 実践① CSVをデータベースに取り込む

実務でまず役に立つのが「Excel/CSVで受け取ったデータをDBに入れて、SQLで集計する」パターンです。例として、月次の注文データ orders.csv を取り込んでみます。

$ head -4 orders.csv
order_date,dept_id,amount
2026-06-01,20,150000
2026-06-02,10,32000
2026-06-02,30,98000

標準ライブラリの csv モジュールで読み、executemany で一括登録します。executemany は「同じSQLを、タプルのリストの件数ぶん繰り返す」メソッドで、1件ずつ execute をループするより速く、コードも短くなります。

import csv
import sqlite3

# 1. CSVを読んでタプルのリストにする
with open("orders.csv", newline="", encoding="utf-8") as f:
    rows = [
        (r["order_date"], int(r["dept_id"]), int(r["amount"]))
        for r in csv.DictReader(f)
    ]

conn = sqlite3.connect("company.db")

# 2. 受け皿のテーブルを用意する(あれば何もしない)
conn.execute("""
    CREATE TABLE IF NOT EXISTS orders (
        id         INTEGER PRIMARY KEY AUTOINCREMENT,
        order_date TEXT    NOT NULL,
        dept_id    INTEGER,
        amount     INTEGER NOT NULL CHECK (amount >= 0)
    )
""")

# 3. 一括INSERT(ここでもプレースホルダ)
conn.executemany(
    "INSERT INTO orders (order_date, dept_id, amount) VALUES (?, ?, ?)", rows
)
conn.commit()
print(f"{len(rows)}件を取り込みました")

# 4. 取り込んだそばからSQLで集計
cur = conn.execute("""
    SELECT d.name, COUNT(*), SUM(o.amount)
    FROM orders o
    JOIN departments d ON o.dept_id = d.id
    GROUP BY d.name
    ORDER BY SUM(o.amount) DESC
""")
for dept, cnt, total in cur.fetchall():
    print(f"{dept}: {cnt}件 / 合計 {total:,}円")
conn.close()

$ python3 import_orders.py
120件を取り込みました
営業部: 58件 / 合計 6,420,000円
情報システム部: 34件 / 合計 2,150,000円
総務部: 28件 / 合計 812,000円

Excelなら「ピボットテーブルを作って、範囲を選び直して……」と手作業だった月次集計が、スクリプト1本の再実行で終わるようになりました。第4章で学んだGROUP BYと第7章のJOINが、Pythonと組み合わさって初めて「自動化」に化けるのを体感してください。

8.4 実践② 問い合わせ管理ツールを作る

仕上げに、第6章でテーブル設計をした「社内問い合わせ管理」を、動くCLIツールにします。機能は登録(add)・一覧(list)・完了への更新(done)・キーワード検索(search)の4つ。完成コードを先に載せます(約100行)。

#!/usr/bin/env python3
"""社内問い合わせ管理ツール ticket.py"""
import sqlite3
import sys
from contextlib import closing
from datetime import datetime

DB_PATH = "company.db"
STATUSES = ("open", "in_progress", "closed")


def get_conn():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row  # 列名でアクセスできるようにする
    return conn


def init_db():
    with closing(get_conn()) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS tickets (
                id         INTEGER PRIMARY KEY AUTOINCREMENT,
                title      TEXT NOT NULL,
                requester  TEXT NOT NULL,
                status     TEXT NOT NULL DEFAULT 'open'
                           CHECK (status IN ('open', 'in_progress', 'closed')),
                created_at TEXT NOT NULL
            )
        """)
        conn.commit()


def add_ticket(title, requester):
    now = datetime.now().strftime("%Y-%m-%d %H:%M")
    with closing(get_conn()) as conn:
        cur = conn.execute(
            "INSERT INTO tickets (title, requester, created_at) VALUES (?, ?, ?)",
            (title, requester, now),
        )
        conn.commit()
        print(f"登録しました (id={cur.lastrowid})")


def list_tickets(status=None):
    sql = "SELECT id, title, requester, status, created_at FROM tickets"
    params = ()
    if status is not None:
        sql += " WHERE status = ?"
        params = (status,)
    sql += " ORDER BY id"
    with closing(get_conn()) as conn:
        rows = conn.execute(sql, params).fetchall()
    if not rows:
        print("該当するチケットはありません")
        return
    for r in rows:
        print(f"[{r['id']:3d}] {r['status']:<11} {r['created_at']}  "
              f"{r['title']} ({r['requester']})")


def update_status(ticket_id, status):
    if status not in STATUSES:
        print(f"statusは {STATUSES} のいずれかです")
        return
    with closing(get_conn()) as conn:
        cur = conn.execute(
            "UPDATE tickets SET status = ? WHERE id = ?", (status, ticket_id)
        )
        conn.commit()
        if cur.rowcount == 0:
            print(f"id={ticket_id} のチケットは存在しません")
        else:
            print(f"id={ticket_id} を {status} に更新しました")


def search_tickets(keyword):
    pattern = f"%{keyword}%"  # LIKE用のパターンは「値」なのでこれはOK
    with closing(get_conn()) as conn:
        rows = conn.execute(
            "SELECT id, title, status FROM tickets "
            "WHERE title LIKE ? ORDER BY id",
            (pattern,),
        ).fetchall()
    for r in rows:
        print(f"[{r['id']:3d}] {r['status']:<11} {r['title']}")


def usage():
    print("使い方: python3 ticket.py add <件名> <依頼者>")
    print("        python3 ticket.py list [open|in_progress|closed]")
    print("        python3 ticket.py done <id>")
    print("        python3 ticket.py search <キーワード>")


def main():
    init_db()
    args = sys.argv[1:]
    if len(args) == 3 and args[0] == "add":
        add_ticket(args[1], args[2])
    elif args and args[0] == "list":
        list_tickets(args[1] if len(args) >= 2 else None)
    elif len(args) == 2 and args[0] == "done":
        update_status(int(args[1]), "closed")
    elif len(args) == 2 and args[0] == "search":
        search_tickets(args[1])
    else:
        usage()


if __name__ == "__main__":
    main()

動かすとこうなります。

$ python3 ticket.py add "プリンタが印刷できない" "鈴木 一郎"
登録しました (id=1)
$ python3 ticket.py add "VPNに接続できない" "高橋 美咲"
登録しました (id=2)
$ python3 ticket.py list
[  1] open        2026-07-09 09:12  プリンタが印刷できない (鈴木 一郎)
[  2] open        2026-07-09 09:15  VPNに接続できない (高橋 美咲)
$ python3 ticket.py done 1
id=1 を closed に更新しました
$ python3 ticket.py search VPN
[  2] open        VPNに接続できない

読みどころを3つ挙げます。

• SQLはすべてプレースホルダ — 件名・依頼者・キーワード、外から来る値は例外なく ? で渡しています。search_tickets の %{keyword}% はf文字列ですが、組み立てているのは「LIKEに渡す値」であってSQL文ではないので安全です。
• CHECK制約とPython側チェックの二重防御 — statusの値は第6章で学んだCHECK制約がDB側で守り、Python側でも STATUSES で検証しています。「アプリのチェックはすり抜けられることがある。最後の砦はDBの制約」という役割分担です。
• row_factoryで可読性を上げる — sqlite3.Row を設定すると r['title'] のように列名で読めます。タプルの添字 r[1] より、列を増やしたときの事故が減ります。

8.5 ORMという選択肢

実務のWeb開発では、SQLを直接書く代わりにORM(Object-Relational Mapper)というライブラリを使うことがよくあります。Pythonでの定番はSQLAlchemyで、テーブルをクラス、行をオブジェクトとして扱えます。雰囲気だけ見てください。

# SQLAlchemyの雰囲気(このコースでは深入りしません)
from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session

engine = create_engine("sqlite:///company.db")
with Session(engine) as session:
    stmt = select(Employee).where(Employee.salary >= 300000)
    for emp in session.scalars(stmt):
        print(emp.name)   # SQLを書かずに済んでいる

便利そうに見えますし、実際に便利です。しかし学ぶ順序は「生SQLが先、ORMが後」を強く勧めます。ORMは裏で結局SQLを生成しており、生SQLを知らないままORMだけ使うと、遅いクエリが生成されていることや、第7章で学んだN+1問題が起きていることに気づくことすらできないからです。逆に、SQLと実行計画を読める人にとって、ORMは安心して使える時短ツールになります。

まとめ

• Pythonの sqlite3 は標準ライブラリ。connect → cursor → execute → fetchall/fetchone → close が基本の流れで、変更系は commit() を忘れない
• with conn: はコミット/ロールバックの自動化であってcloseではない。閉じるのは contextlib.closing か明示的な close()
• ユーザー入力をSQLに文字列連結した瞬間にSQLインジェクション脆弱性になる。値は例外なく ? プレースホルダで渡す
• csv + executemany でCSVを取り込めば、Excelで手作業だった集計がSQL1本になる
• ORM(SQLAlchemy)は便利だが、生SQLを知らずに使うと遅いクエリやN+1問題に気づけない。学ぶ順序は生SQLが先
• 接続情報はコードに直書きせず、環境変数や.gitignore済みの設定ファイルで管理する

練習問題

dept = input("部署ID: ")
min_salary = input("最低給与: ")
cur.execute(
    f"SELECT name FROM employees "
    f"WHERE dept_id = {dept} AND salary >= {min_salary}"
)

dept = input("部署ID: ")
min_salary = input("最低給与: ")
cur.execute(
    "SELECT name FROM employees WHERE dept_id = ? AND salary >= ?",
    (int(dept), int(min_salary)),
)

import sqlite3
from contextlib import closing

with closing(sqlite3.connect("company.db")) as conn:
    cur = conn.execute("""
        SELECT d.name, COUNT(*), AVG(e.salary)
        FROM employees e
        JOIN departments d ON e.dept_id = d.id
        GROUP BY d.name
        ORDER BY AVG(e.salary) DESC
    """)
    for dept, cnt, avg_salary in cur.fetchall():
        print(f"{dept}: {cnt}名 / 平均 {avg_salary:,.0f}円")

$ python3 dept_report.py
総務部: 2名 / 平均 327,500円
情報システム部: 1名 / 平均 350,000円
営業部: 2名 / 平均 310,000円