28. リレーショナルデータベース
テーマ: SQLite で売上データベースを構築し SQL で分析する
学習点: sqlite3, DDL/DML, 外部キー, インデックス, トランザクション, JOIN・GROUP BY・ウィンドウ関数, プレースホルダ（SQLインジェクション対策）
依存: 標準ライブラリのみ ／ 難易度: 上級
実行方法
uv run 28_sqlite_db.py
スクリプト冒頭の PEP 723 メタデータ（# /// script）により、必要なライブラリは
uv が自動的に仮想環境へ導入します。事前の pip install は不要です。
解説
何をするプログラムか
販売管理・会計・在庫管理など、企業の基幹システムのほぼすべてはリレーショナルデータベース（RDB）の上に構築されています。データを顧客・商品・受注という正規化された表に分け、SQL という宣言的な言語で問い合わせるのが RDB の基本です。本スクリプトは、サーバ不要で Python 標準ライブラリだけで使える SQLite を使い、顧客 200 人・商品 7 種・受注 5,000 件の売上データベースをゼロから構築し、集計・順位付け・安全なクエリ・性能改善・トランザクションという実務の主要トピックを一通り実演します。
SQL には集計だけでなく、行の並びを保ったまま累積や順位を計算する「ウィンドウ関数」があり、月次売上の累積・移動平均・地域内顧客ランキングを 1 本のクエリで求められることも確認します。
コードの読みどころ
SCHEMAの DDL では、REFERENCES customers(id)の外部キーとCHECK (price > 0)の制約で「不正なデータをそもそも入れない」設計をしています。SQLite ではPRAGMA foreign_keys = ONを明示しないと外部キーが働かない点に注意してください。build()はcon.executemany(...)とプレースホルダ?で 5,000 件を一括挿入します。値の埋め込みを DB ドライバに任せるこの書き方が、後述のインジェクション対策にもなります。- 「月次売上と累積」のクエリでは
SUM(SUM(...)) OVER (ORDER BY 月)で累積売上を、ROWS BETWEEN 2 PRECEDING AND CURRENT ROWで 3 か月移動平均を計算しています。「地域内順位」では CTE（WITH t AS ...）で集計結果に名前を付け、RANK() OVER (PARTITION BY region ORDER BY amt DESC)で地域ごとの順位を振っています。 - インジェクション対策の実演では、
"東京' OR '1'='1"という悪意ある入力をプレースホルダ経由で渡すとヒット 0 件、つまりただの文字列として扱われることを確認します。f 文字列で SQL を組み立てると条件が常に真になり全件漏えいします。 - インデックスの節では、同じ範囲検索を 200 回実行して
CREATE INDEX前後の時間を比較し、EXPLAIN QUERY PLANで実行計画がSEARCH ... USING INDEXに変わったことを確認します。 - 最後の
with con:ブロックは、正常終了なら commit、例外なら rollback を自動で行います。2 件目の INSERT が外部キー違反（存在しない顧客 99999）で失敗すると、成功していた 1 件目も取り消されます。
実行結果の見方
カテゴリ別売上では、受注件数が最多の周辺機器（2,170 件）よりも単価の高いサービス（受注 1,422 件・売上 7.8 億円）が売上首位で、「件数と金額は別物」という集計の基本が見えます。インデックスの効果は 70.8 ms から 29.0 ms への 2.44 倍の高速化として表れています。トランザクションの節では IntegrityError: FOREIGN KEY constraint failed の後に 2025-12-31 の登録件数が 0 件、すなわち一連の処理が「全部成功か全部取り消しか」になるという原子性（ACID の A）を数字で確認できます。
ソースコード
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""28: リレーショナルデータベース -----------------------------------------
テーマ: SQLite で売上データベースを構築し SQL で分析する
学習点: sqlite3, DDL/DML, 外部キー, インデックス, トランザクション,
JOIN・GROUP BY・ウィンドウ関数, プレースホルダ（SQLインジェクション対策）
"""
import random
import sqlite3
import time
from pathlib import Path
random.seed(11)
DB = Path("out_28") / "sales.db"
DB.parent.mkdir(exist_ok=True)
if DB.exists():
DB.unlink()
SCHEMA = """
PRAGMA foreign_keys = ON;
CREATE TABLE customers (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
region TEXT NOT NULL,
joined TEXT NOT NULL
);
CREATE TABLE products (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL UNIQUE,
category TEXT NOT NULL,
price INTEGER NOT NULL CHECK (price > 0)
);
CREATE TABLE orders (
id INTEGER PRIMARY KEY,
customer_id INTEGER NOT NULL REFERENCES customers(id),
product_id INTEGER NOT NULL REFERENCES products(id),
qty INTEGER NOT NULL CHECK (qty > 0),
order_date TEXT NOT NULL
);
"""
def build(con: sqlite3.Connection) -> None:
con.executescript(SCHEMA)
regions = ["東京", "大阪", "名古屋", "福岡", "札幌"]
con.executemany(
"INSERT INTO customers (name, region, joined) VALUES (?, ?, ?)",
[(f"顧客{i:03d}", random.choice(regions),
f"202{random.randint(2,5)}-{random.randint(1,12):02d}-01")
for i in range(1, 201)])
con.executemany(
"INSERT INTO products (name, category, price) VALUES (?, ?, ?)",
[("ノートPC", "ハード", 148000), ("モニタ", "ハード", 34000),
("キーボード", "周辺機器", 9800), ("マウス", "周辺機器", 4500),
("ドック", "周辺機器", 22000), ("保守契約", "サービス", 60000),
("導入支援", "サービス", 250000)])
con.executemany(
"INSERT INTO orders (customer_id, product_id, qty, order_date) "
"VALUES (?, ?, ?, ?)",
[(random.randint(1, 200), random.randint(1, 7), random.randint(1, 6),
f"2025-{random.randint(1,12):02d}-{random.randint(1,28):02d}")
for _ in range(5000)])
con.commit()
def show(con, title, sql, params=()):
print(f"\n■ {title}")
cur = con.execute(sql, params)
cols = [d[0] for d in cur.description]
rows = cur.fetchall()
widths = [max(len(c), max((len(str(r[i])) for r in rows), default=0)) + 2
for i, c in enumerate(cols)]
print(" " + "".join(f"{c:>{w}}" for c, w in zip(cols, widths)))
print(" " + "-" * sum(widths))
for r in rows:
print(" " + "".join(f"{str(v):>{w}}" for v, w in zip(r, widths)))
def main() -> None:
con = sqlite3.connect(DB)
con.row_factory = sqlite3.Row
build(con)
print(f"DB作成: {DB} ({DB.stat().st_size/1024:.0f} KB)")
show(con, "カテゴリ別売上（JOIN + GROUP BY）", """
SELECT p.category AS カテゴリ,
COUNT(*) AS 受注件数,
SUM(o.qty) AS 数量,
SUM(o.qty * p.price) AS 売上,
ROUND(AVG(o.qty * p.price), 0) AS 平均単価
FROM orders o JOIN products p ON o.product_id = p.id
GROUP BY p.category ORDER BY 売上 DESC
""")
show(con, "地域別 上位顧客（サブクエリ + LIMIT）", """
SELECT c.region AS 地域, c.name AS 顧客,
SUM(o.qty * p.price) AS 売上
FROM orders o
JOIN customers c ON o.customer_id = c.id
JOIN products p ON o.product_id = p.id
GROUP BY c.id ORDER BY 売上 DESC LIMIT 5
""")
show(con, "月次売上と累積（ウィンドウ関数）", """
SELECT substr(o.order_date, 1, 7) AS 月,
SUM(o.qty * p.price) AS 売上,
SUM(SUM(o.qty * p.price)) OVER (
ORDER BY substr(o.order_date, 1, 7)) AS 累積,
ROUND(AVG(SUM(o.qty * p.price)) OVER (
ORDER BY substr(o.order_date,1,7)
ROWS BETWEEN 2 PRECEDING AND CURRENT ROW), 0) AS 移動平均3
FROM orders o JOIN products p ON o.product_id = p.id
GROUP BY 月 ORDER BY 月
""")
show(con, "地域内順位（CTE + ウィンドウ関数 RANK）", """
WITH t AS (
SELECT c.region, c.name, SUM(o.qty*p.price) AS amt
FROM orders o JOIN customers c ON o.customer_id=c.id
JOIN products p ON o.product_id =p.id
GROUP BY c.id),
r AS (SELECT *, RANK() OVER (PARTITION BY region ORDER BY amt DESC) AS rk
FROM t)
SELECT region AS 地域, name AS 顧客, amt AS 売上, rk AS 地域内順位
FROM r WHERE rk <= 2 ORDER BY region, rk
""")
print("\n■ プレースホルダによる安全なクエリ（SQLインジェクション対策）")
evil = "東京' OR '1'='1"
n = con.execute("SELECT COUNT(*) FROM customers WHERE region = ?",
(evil,)).fetchone()[0]
print(f" 悪意ある入力 {evil!r} -> ヒット {n} 件（文字列として扱われる）")
print(" ※ f文字列でSQLを組み立てると全件返ってしまう。必ず ? を使う。")
print("\n■ インデックスの効果")
# 前方一致(LIKE)ではなく範囲条件にするとB木インデックスが効く
q = ("SELECT COUNT(*) FROM orders o JOIN products p ON o.product_id=p.id "
"WHERE o.order_date >= '2025-06-01' AND o.order_date < '2025-07-01'")
t0 = time.perf_counter()
for _ in range(200):
con.execute(q).fetchone()
t1 = time.perf_counter() - t0
con.execute("CREATE INDEX idx_orders_date ON orders(order_date)")
con.execute("ANALYZE")
t0 = time.perf_counter()
for _ in range(200):
con.execute(q).fetchone()
t2 = time.perf_counter() - t0
print(f" インデックスなし {t1*1000:>8.1f} ms / あり {t2*1000:>8.1f} ms "
f"({t1/t2:.2f}倍)")
print(" 実行計画:", con.execute("EXPLAIN QUERY PLAN " + q)
.fetchall()[0]["detail"])
print("\n■ トランザクションとロールバック（制約違反）")
try:
with con: # with 文を抜けるとcommit、例外ならrollback
con.execute("INSERT INTO orders (customer_id, product_id, qty, "
"order_date) VALUES (1, 1, 1, '2025-12-31')")
con.execute("INSERT INTO orders (customer_id, product_id, qty, "
"order_date) VALUES (99999, 1, 1, '2025-12-31')")
except sqlite3.IntegrityError as e:
print(f" IntegrityError: {e}")
cnt = con.execute("SELECT COUNT(*) FROM orders "
"WHERE order_date='2025-12-31'").fetchone()[0]
print(f" 2025-12-31 の登録件数 = {cnt} 件 -> 1件目も取り消された（原子性）")
con.close()
if __name__ == "__main__":
main()
実行結果
DB作成: out_28/sales.db (148 KB)
■ カテゴリ別売上（JOIN + GROUP BY）
カテゴリ 受注件数 数量 売上 平均単価
---------------------------------------
サービス 1422 5006 781060000 549269.0
ハード 1408 4820 428816000 304557.0
周辺機器 2170 7459 91158200 42008.0
■ 地域別 上位顧客（サブクエリ + LIMIT）
地域 顧客 売上
---------------------
福岡 顧客189 17016900
札幌 顧客198 12315200
札幌 顧客092 11702200
福岡 顧客063 11364000
札幌 顧客105 11159400
■ 月次売上と累積（ウィンドウ関数）
月 売上 累積 移動平均3
---------------------------------------------
2025-01 113855000 113855000 113855000.0
2025-02 97693400 211548400 105774200.0
2025-03 122527900 334076300 111358767.0
2025-04 116822600 450898900 112347967.0
2025-05 111904800 562803700 117085100.0
2025-06 107025500 669829200 111917633.0
2025-07 109328500 779157700 109419600.0
2025-08 110508600 889666300 108954200.0
2025-09 101167400 990833700 107001500.0
2025-10 117032300 1107866000 109569433.0
2025-11 84350300 1192216300 100850000.0
2025-12 108817900 1301034200 103400167.0
■ 地域内順位（CTE + ウィンドウ関数 RANK）
地域 顧客 売上 地域内順位
-----------------------------
名古屋 顧客193 10503200 1
名古屋 顧客160 9611300 2
大阪 顧客100 10485000 1
大阪 顧客169 9067500 2
札幌 顧客198 12315200 1
札幌 顧客092 11702200 2
東京 顧客081 10978200 1
東京 顧客150 9700800 2
福岡 顧客189 17016900 1
福岡 顧客063 11364000 2
■ プレースホルダによる安全なクエリ（SQLインジェクション対策）
悪意ある入力 "東京' OR '1'='1" -> ヒット 0 件（文字列として扱われる）
※ f文字列でSQLを組み立てると全件返ってしまう。必ず ? を使う。
■ インデックスの効果
インデックスなし 70.8 ms / あり 29.0 ms (2.44倍)
実行計画: SEARCH o USING INDEX idx_orders_date (order_date>? AND order_date<?)
■ トランザクションとロールバック（制約違反）
IntegrityError: FOREIGN KEY constraint failed
2025-12-31 の登録件数 = 0 件 -> 1件目も取り消された（原子性）