【保存版】SQLite×Pythonで作る“ローカルDWH”——ETL・集計・レポート自動化の最短手順

retail_etl/
  ├─ data/
  │   ├─ raw/           # 受領データ（CSV等）
  │   └─ warehouse/     # SQLite DB配置
  ├─ src/
  │   ├─ etl.py         # 取り込み/整形
  │   ├─ sql/           # SQLテンプレ
  │   │   ├─ create_tables.sql
  │   │   └─ queries.sql
  │   └─ report.py      # 集計→Excel/画像出力
  ├─ reports/
  ├─ requirements.txt
  └─ README.md

PRAGMA foreign_keys=ON;
CREATE TABLE IF NOT EXISTS sales (
  id         INTEGER PRIMARY KEY AUTOINCREMENT,
  order_date TEXT NOT NULL,
  store      TEXT NOT NULL,
  product    TEXT NOT NULL,
  qty        INTEGER NOT NULL,
  price      REAL    NOT NULL,
  UNIQUE(order_date, store, product, qty, price)
);
CREATE TABLE IF NOT EXISTS stores (
  store     TEXT PRIMARY KEY,
  area      TEXT,
  open_date TEXT
);
CREATE INDEX IF NOT EXISTS idx_sales_store_date ON sales(store, order_date);

from pathlib import Path
import sqlite3

ROOT = Path(**file**).resolve().parents\[1]
DB = ROOT / "data/warehouse/retail.sqlite"
SQL\_DIR = ROOT / "src/sql"

def init\_db() -> None:
DB.parent.mkdir(parents=True, exist\_ok=True)
con = sqlite3.connect(DB)
try:
con.execute("PRAGMA journal\_mode=WAL;")  # 高速/同時読み取り
con.execute("PRAGMA synchronous=NORMAL;")
con.execute("PRAGMA foreign\_keys=ON;")
con.executescript((SQL\_DIR / "create\_tables.sql").read\_text(encoding="utf-8"))
finally:
con.close()

import pandas as pd, numpy as np, sqlite3
from pathlib import Path

ROOT = Path(**file**).resolve().parents\[1]
DB = ROOT / "data/warehouse/retail.sqlite"

def load\_csv\_to\_db(csv\_path: str) -> None:
con = sqlite3.connect(DB)
try:
df = pd.read\_csv(csv\_path)
\# バリデーション（最小限）
req = {"order\_date","store","product","qty","price"}
assert req.issubset(df.columns)
df = df.dropna(subset=\["order\_date","store","product"])
df\["qty"] = pd.to\_numeric(df\["qty"], errors="coerce").fillna(0).astype(int)
df\["price"] = pd.to\_numeric(df\["price"], errors="coerce").astype(float)
df = df\[df\["qty"] > 0]
\# 追記
df.to\_sql("sales", con, if\_exists="append", index=False)
finally:
con.close()

import sqlite3
from pathlib import Path

DB = Path(**file**).resolve().parents\[1] / "data/warehouse/retail.sqlite"

def insert\_store(store: str, area: str, open\_date: str) -> None:
con = sqlite3.connect(DB)
try:
con.execute("PRAGMA foreign\_keys=ON;")
with con:  # トランザクション
con.execute(
"INSERT OR REPLACE INTO stores(store, area, open\_date) VALUES(?,?,?)",
(store, area, open\_date)
)
finally:
con.close()

import sqlite3, pandas as pd
from pathlib import Path

DB = Path(**file**).resolve().parents\[1] / "data/warehouse/retail.sqlite"

Q\_MONTHLY = """
WITH monthly AS (
SELECT strftime('%Y-%m', order\_date) AS ym, store,
SUM(qty\*price) AS sales, SUM(qty) AS qty
FROM sales
WHERE qty>0 AND price>0
GROUP BY ym, store
)
SELECT \* FROM monthly;
"""

def fetch\_monthly() -> pd.DataFrame:
con = sqlite3.connect(DB)
try:
return pd.read\_sql\_query(Q\_MONTHLY, con)
finally:
con.close()

from pathlib import Path
import matplotlib.pyplot as plt
from queries import fetch_monthly

ROOT = Path(**file**).resolve().parents\[1]

def export\_report() -> None:
df = fetch\_monthly()
\# 可視化（例：月次売上推移）
g = df.groupby("ym", as\_index=False)\["sales"].sum()
plt.figure(); plt.plot(g\["ym"], g\["sales"], marker="o")
plt.title("月次売上の推移"); plt.xticks(rotation=45); plt.tight\_layout()
fig\_path = ROOT / "reports/monthly\_sales.png"
plt.savefig(fig\_path, dpi=200, bbox\_inches="tight")

```
# Excel書き出し
xlsx = ROOT / "reports/report.xlsx"
xlsx.parent.mkdir(parents=True, exist_ok=True)
with pd.ExcelWriter(xlsx) as xw:
    df.to_excel(xw, sheet_name="monthly", index=False)
print("Saved:", xlsx, fig_path)

import sqlite3, pandas as pd
from pathlib import Path

DB = Path(**file**).resolve().parents\[1] / "data/warehouse/retail.sqlite"

def incremental\_load(csv\_path: str) -> None:
con = sqlite3.connect(DB)
try:
cur = con.execute("SELECT IFNULL(MAX(order\_date),'0000-00-00') FROM sales")
last = cur.fetchone()\[0]
df = pd.read\_csv(csv\_path)
df = df\[df\["order\_date"] > last]
if not df.empty:
df.to\_sql("sales", con, if\_exists="append", index=False)
finally:
con.close()

import pandas as pd, sqlite3
from pathlib import Path

DB = Path(**file**).resolve().parents\[1] / "data/warehouse/retail.sqlite"

def check\_sales() -> None:
con = sqlite3.connect(DB)
try:
df = pd.read\_sql\_query("SELECT qty, price FROM sales", con)
assert (df\["qty"] >= 0).all(), "qty に負値あり"
assert (df\["price"] >= 0).all(), "price に負値あり"
finally:
con.close()

import logging, sqlite3
from pathlib import Path

logging.basicConfig(level=logging.INFO)
DB = Path(**file**).resolve().parents\[1] / "data/warehouse/retail.sqlite"

def safe\_insert(rows: list\[tuple]) -> None:
con = sqlite3.connect(DB)
try:
with con:
con.executemany(
"INSERT INTO sales(order\_date,store,product,qty,price) VALUES(?,?,?,?,?)",
rows
)
except Exception as e:
logging.exception("insert failed: %s", e)  # 自動でROLLBACK
finally:
con.close()

-- Top-N per group
SELECT * FROM (
  SELECT *, ROW_NUMBER() OVER(PARTITION BY store ORDER BY sales DESC) rn
  FROM monthly
) WHERE rn <= 3;

\-- IFNULL/COALESCE
SELECT COALESCE(area, 'Unknown') FROM stores;

\-- 日付操作（文字列→月）
SELECT strftime('%Y-%m', order\_date) AS ym FROM sales;

import sqlite3, pandas as pd
con = sqlite3.connect(":memory:")
con.executescript("CREATE TABLE t(a INT); INSERT INTO t VALUES(1),(2);")
print(pd.read_sql_query("SELECT COUNT(*) AS n FROM t", con))
con.close()