前提：API が検索させてくれないことがある
公開 API を使っていると、こういう仕様にときどき出会います。
- 日付での絞り込みができない
- ソート順が固定
- 1 回に取れる件数に上限がある
- ページ送りは
offsetだけ
こうなると「先月の分を条件付きで見たい」という当たり前の要求が、
サーバ側では絶対に叶いません。残る道は 1 つです。
取得したものを手元に貯めて、検索は自分でやる。
このとき、追加インストール無しで使える保存先が SQLite です。
Python には sqlite3 が標準添付されているので、import するだけで始められます。
この記事では、その「貯めて検索する層」の作り方と、
実際に踏んだ落とし穴をまとめます。
まず API の仕様を確定させる
ドキュメントが Swagger UI で JavaScript レンダリングだったりすると読みづらいので、
実際に叩いて確かめるのが早いです。
このとき使える小技があります。パラメータを検証している API なら、
候補を全部まとめて 1 回投げると、無効なものだけ教えてくれることがあります。
params = dict(limit=1, offset=0, order=1, min_scale=10, max_scale=70,
since_date="20260801", until_date="20260810",
start_date="x", end_date="x", sort="x", page="1")
r = requests.get(BASE + "/history", params=params, timeout=30)
print(r.status_code, r.text)
400 {"error":"extra keys found",
"extra_keys":["order","min_scale","max_scale","since_date","until_date",
"start_date","end_date","sort","page"]}
1 リクエストで有効なパラメータ一覧が確定しました。
残ったのは limit と offset だけ。日付検索も並べ替えも絞り込みも無い、と分かります。
（当てずっぽうを 10 回投げると 10 回分の待ち時間と負荷がかかります。
まとめて 1 回にするのは相手のためでもあります。）
テーブル設計：API の生データを残す
SCHEMA = """
CREATE TABLE IF NOT EXISTS quakes (
id TEXT PRIMARY KEY, -- API が振る ID。再取得時の重複を防ぐ鍵
occurred_at TEXT NOT NULL, -- 'YYYY-MM-DD HH:MM:SS'
place TEXT NOT NULL,
latitude REAL, -- 不明なら NULL
longitude REAL,
magnitude REAL,
max_scale INTEGER,
issued_at TEXT,
issue_type TEXT NOT NULL,
raw_json TEXT NOT NULL, -- 取得時の生 JSON
fetched_at TEXT NOT NULL -- こちらが取得した時刻
);
CREATE INDEX IF NOT EXISTS idx_quakes_occurred_at ON quakes(occurred_at);
"""
意識した点が 3 つあります。
raw_json を持つ
最初から全部の項目を列にしなくても、生の JSON を残しておけば
「やっぱりあの項目も使いたい」となったときにデータを取り直さずに済みます。
ディスクは安いです。
fetched_at を持つ
「いつ取ったデータか」が分かると、おかしな値を見つけたときの調査が一気に楽になります。
索引を張る
日付範囲で検索するなら occurred_at に索引が要ります。
無いと毎回の全件走査です。
SQLite には日付型が無い
ここが最初の関門です。SQLite の型は
NULL / INTEGER / REAL / TEXT / BLOB の 5 つだけ。日付型がありません。
ではどう持つか。答えは単純で、YYYY-MM-DD HH:MM:SS という文字列で持つ。
この形式なら辞書順の比較が、そのまま時系列順の比較になるからです。
'2026-08-13 04:27:00' < '2026-08-18 18:36:00' -- 文字列比較だが正しい
なので範囲検索が普通に書けます。
WHERE occurred_at >= '2026-08-01 00:00:00'
AND occurred_at < '2026-08-20 00:00:00'
逆に言えば、2026/08/13 のようなスラッシュ区切りや、
8/13/2026 のような形式で入れてしまうと比較が壊れます。
API から来た文字列は、取り込む時点で必ず正規化します。
from datetime import datetime
def _parse_datetime(text):
"""'2026/08/18 18:36:00' -> '2026-08-18 18:36:00'"""
for fmt in ("%Y/%m/%d %H:%M:%S.%f", "%Y/%m/%d %H:%M:%S", "%Y/%m/%d %H:%M"):
try:
return datetime.strptime(text.strip(), fmt).strftime("%Y-%m-%d %H:%M:%S")
except ValueError:
continue
return None
「終了日を含む」の書き方
2026-08-19 までを含めたいとき、これは間違いです。
WHERE occurred_at <= '2026-08-19' -- 8/19 の 00:00:00 までしか入らない
'2026-08-19 09:30:00' <= '2026-08-19' は偽です。文字列として長いほうが大きい。
その日を丸ごと含めるには、翌日の 0 時未満にします。
from datetime import timedelta
start = f"{start_day.isoformat()} 00:00:00"
end = f"{(end_day + timedelta(days=1)).isoformat()} 00:00:00"
# WHERE occurred_at >= :start AND occurred_at < :end
値は必ずプレースホルダで渡す
条件が動的に増える検索では、SQL を文字列で組み立てたくなります。
組み立ててよいのは「条件を足すかどうか」という構造だけで、値は絶対にプレースホルダです。
sql = "SELECT * FROM quakes WHERE occurred_at >= :start AND occurred_at < :end"
params = {"start": start, "end": end, "limit": limit}
if min_scale is not None:
sql += " AND max_scale >= :min_scale"
params["min_scale"] = min_scale
if keyword:
sql += " AND place LIKE :keyword"
params["keyword"] = f"%{keyword}%" # ワイルドカードは「値」側に入れる
sql += " ORDER BY occurred_at DESC LIMIT :limit"
rows = connection.execute(sql, params).fetchall()
SQL インジェクションの話は有名ですが、普通に使っていても壊れます。
地名に ' が 1 個入っているだけで、文字列連結の SQL は構文エラーになります。
LIKE のワイルドカードは、SQL 側ではなく値の側に入れます。
params["keyword"] = f"%{keyword}%" # ○
sql += " AND place LIKE '%' || :kw || '%'" # これでもよいが読みにくい
sqlite3 は ?（位置指定）と :name（名前指定）の両方に対応しています。
条件が増減する検索では名前指定が圧倒的に楽です。位置指定だと
条件を 1 つ足すたびにタプルの順番を数え直すことになります。
重複データを SQL で 1 件にまとめる
今回いちばん面白かったのがここです。
扱っていた API は、同じ 1 回の地震について複数回レコードを配信していました。
情報が確定するにつれて続報が出る仕組みです。
| issue_type | 内容 | 震源地名 |
|---|---|---|
ScalePrompt |
速報 | まだ「不明」 |
Destination |
震源が確定 | 確定 |
DetailScale |
詳細が確定 | 確定 |
そのまま並べると一覧が重複し、地図には同じ点が重なります。
方針：DB には全部入れる。まとめるのは検索時
「保存する前に重複を除く」ことも考えましたが、やめました。
DB は API の忠実な写しにしておいて、見せ方の方針は検索側で決めるほうが、
後から方針を変えられるからです。捨てたデータは戻ってきません。
ROW_NUMBER() で「グループごとの代表 1 行」を選ぶ
SQLite 3.25 以降ならウィンドウ関数が使えます（Python 3.11 以降なら大抵入っています）。
SELECT id, occurred_at, place, latitude, longitude, magnitude, max_scale
FROM (
SELECT *, ROW_NUMBER() OVER (
PARTITION BY occurred_at
ORDER BY
CASE issue_type
WHEN 'DetailScale' THEN 0
WHEN 'ScaleAndDestination' THEN 1
WHEN 'Destination' THEN 2
WHEN 'ScalePrompt' THEN 3
ELSE 4
END,
(latitude IS NULL),
issued_at DESC
) AS priority
FROM quakes
WHERE occurred_at >= :start AND occurred_at < :end
)
WHERE priority = 1
ORDER BY occurred_at DESC
LIMIT :limit
読み方はこうです。
-
PARTITION BY occurred_at… 発生時刻ごとにグループ分けする -
ORDER BY ...… グループの中での優先順位を決める -
ROW_NUMBER()… その順で 1, 2, 3… と番号を振る - 外側で
WHERE priority = 1… 各グループの 1 位だけ残す
「グループごとに代表 1 行を取る」定石です。
GROUP BY + MAX() では「どの行が選ばれたか」の他の列が取れませんが、
この書き方なら行まるごと取れます。
ORDER BY の中で CASE を使って優先順位を明示しているのがポイントです。
issue_type は文字列なので、そのまま並べても意味のある順になりません。
(latitude IS NULL) は真が 1・偽が 0 になるので、
座標を持っている行が先に来ます（昇順なので 0 が先）。
実データでは 100 レコードが 83 件になりました。
名寄せの鍵をどう選ぶか
最初は PARTITION BY occurred_at, place にしていました。素直な発想です。
ところがこれだと重複が消えませんでした。
速報の段階では震源地名が「不明」だからです。
同じ地震が「不明」と「熊本県天草・芦北地方」に割れてしまう。
そこで鍵から地名を外し、発生時刻だけにしました。
代償として「同じ分に発生した別々の地震」は 1 件に潰れます。
発生時刻が分単位である以上、これは避けられません。
こういう判断は必ずコードに残しておきます。半年後の自分は絶対に覚えていません。
"""
名寄せの鍵に震源地名を含めていないのは、速報の段階では震源が未確定で
地名が「不明」になり、同じ地震が 2 行に割れてしまうため。
代償として「同じ分に発生した別々の地震」は 1 件に潰れるが、
発生時刻は分単位であり、実際にぶつかる頻度は低いと判断した。
"""
with connection: は接続を閉じない
これは知らないと必ず踏みます。
with sqlite3.connect(path) as connection:
connection.execute("INSERT ...")
# ここで接続は閉じていない
ファイルオブジェクトの with open(...) と同じ感覚で書くと裏切られます。
sqlite3 の with connection: は
「トランザクションをコミット / ロールバックする」構文であって、close() ではありません。
放置すると接続が溜まり、Windows では
PermissionError: [WinError 32] プロセスはファイルにアクセスできません。
別のプロセスが使用中です。
で DB ファイルを削除も移動もできなくなります。私はテストの後片付けで踏みました。
正しくは contextlib.closing と併用します。
from contextlib import closing
with closing(self._connect()) as connection, connection:
connection.execute("INSERT ...")
読み方は「closing が閉じる責任、内側の connection がコミットの責任」。
読み取りだけなら closing だけで十分です。
with closing(self._connect()) as connection:
rows = connection.execute(sql, params).fetchall()
接続はスレッドをまたげない
GUI アプリだと通信を別スレッドでやるので、これも効いてきます。
sqlite3.Connection は既定で作ったスレッド以外から使うと例外になります。
ProgrammingError: SQLite objects created in a thread can only be used in that same thread.
check_same_thread=False で黙らせることもできますが、
今度は自分でロックを管理する羽目になります。
操作ごとに開いて閉じるほうが単純で安全です。
ローカルファイルへの接続は十分速いので、実測でも問題になりませんでした。
def _connect(self):
connection = sqlite3.connect(self.db_path, timeout=10.0)
connection.row_factory = sqlite3.Row # 列名でアクセスできるようにする
return connection
row_factory = sqlite3.Row を入れておくと row["place"] と書けます。
row[3] で書いていると、列を 1 つ足した瞬間に全部ずれます。
まとめて書く
1 件ずつ execute + commit すると、件数に比例して遅くなります。
executemany は 1 トランザクションでまとめて流れます。
def save_many(self, quakes):
rows = [(q.id, q.occurred_at, q.place, ...) for q in quakes]
with closing(self._connect()) as connection, connection:
before = connection.execute("SELECT COUNT(*) FROM quakes").fetchone()[0]
connection.executemany(
"""
INSERT OR REPLACE INTO quakes (id, occurred_at, place, ...)
VALUES (?, ?, ?, ...)
""",
rows,
)
after = connection.execute("SELECT COUNT(*) FROM quakes").fetchone()[0]
return after - before # 「新規に増えた件数」
INSERT OR REPLACE は主キーが衝突したら置き換えます。
同じデータを何度取り直しても行が増えないので、
再取得を気軽にやれるようになります。
「何件が新規だったか」は前後の総件数の差で取れます。
cursor.rowcount は REPLACE を含めた件数を返すので、ここでは使えません。
差分取得で通信を減らす
貯める設計にすると、次はこう考えます。
「毎回 3000 件取り直すのは無駄では？」
既知の ID が分かれば打ち切れます。
def known_ids(self):
with closing(self._connect()) as connection:
return {row[0] for row in connection.execute("SELECT id FROM quakes")}
known = repository.known_ids() # 一度だけまとめて読む
for page in range(MAX_PAGES):
items = client.fetch_page(offset=page * PAGE_SIZE)
if not items:
break
collected.update({i.id: i for i in items})
# 1 ページ丸ごと既知 = その先も取得済み
if all(i.id in known for i in items):
break
# 目的の日付より古いところまで到達した
if min(i.occurred_at for i in items) < boundary:
break
ここで大事なのが、known_ids() を一度だけ呼ぶことです。
is_known(id) を 1 件ずつ DB に問い合わせる形にすると、
100 件のページごとに 100 回接続することになります。集合にして持てば in は一瞬です。
まとめ
| 落とし穴 | 対処 |
|---|---|
| SQLite に日付型が無い |
YYYY-MM-DD HH:MM:SS の文字列で持つ（辞書順＝時系列順） |
| 終了日が含まれない | 翌日 0 時 未満 で書く |
| 文字列連結の SQL | 値は必ずプレースホルダ（:name 形式が楽） |
| 重複レコード |
ROW_NUMBER() OVER (PARTITION BY ...) で代表 1 行 |
with connection: が閉じない |
with closing(conn) as conn, conn: |
| スレッドをまたげない | 操作ごとに開いて閉じる |
| 1 件ずつの INSERT が遅い |
executemany + INSERT OR REPLACE
|
| 毎回全件取り直す | 既知 ID の集合で打ち切る |
API に検索機能が無いのは不便ですが、SQLite を挟むと
むしろサーバ側より柔軟な検索ができるようになり、オフラインでも動くという
おまけまで付いてきました。標準ライブラリだけでここまでできるのは、なかなか気持ちがいいです。