engine.execute削除→to_sql追記｜26年10月

スクレイピング結果をSQLiteに貯めるコードを生成AIに書かせたら、engine.execute() で落ちた。直させたら今度はエラーも出ずに1行も保存されていなかった。SQLAlchemy 2.0 で書き方が変わったのに、古い保存コードは今も検索結果や生成AIの回答によく出てきます。この記事では、pandas の to_sql() を with engine.begin() の中で呼び、INSERT ... ON CONFLICT で重複を弾きながら毎日追記する、今の書き方をコード付きで示します。最後に systemd timer で止めずに動かし続ける設定まで扱います。

SQLAlchemy 2.0でengine.execute()が使えない理由と、今の保存コードの形

以前はこう書いていた

数年前の記事や生成AIの回答には、次のような保存コードがよく出てきます。

from sqlalchemy import create_engine

engine = create_engine("sqlite:///news.db")
engine.execute("CREATE TABLE IF NOT EXISTS articles (url TEXT PRIMARY KEY, title TEXT)")
df.to_sql("articles", engine, if_exists="append", index=False)

SQLAlchemy 2.0 では、このコードは2箇所で動きません。

Engine.execute() が削除された：エンジンから直接SQLを流す「コネクションレス実行」は廃止されました。SQLは必ずコネクションを開いて、その上で実行します。実行すると AttributeError: 'Engine' object has no attribute 'execute' が出ます。
生の文字列SQLを渡せない：文字列のSQLは text() で包む必要があります。

このコードには、2.0 とは別の問題もあります。url を主キーにしているので、2日目に同じ記事を取ると to_sql() が一意制約違反で落ちます。その日の新着も含めて、まとめて1件も保存されません。

「engine.connect()に置き換えるだけ」の修正は危ない

エラーを見せて直させると、with engine.connect() as conn: に書き換えるだけの修正がよく返ってきます。SQLAlchemy 2.0 の connect() は自動コミットしません。conn.commit() を書き忘れると、ブロックを抜けたときにエラーも出ずにロールバックされます。スクリプトは正常終了するのに、DBは空のままになります。

今はこう書く

やりたいこと	以前の書き方	今の書き方
SQLを実行する	`engine.execute("...")`	`conn.execute(text("..."))`
トランザクション	暗黙の自動コミット	`with engine.begin() as conn:`（正常終了でコミット、例外でロールバック）
DataFrameを保存	`df.to_sql(..., engine)`	`df.to_sql(..., conn)` を `begin()` の中で呼ぶ
重複の扱い	保存後に DELETE で掃除、または落ちる	`method=` に `ON CONFLICT DO NOTHING` の関数を渡す

engine.begin() を使えば、コミットの書き忘れが起こりません。テーブル作成・件数確認・追記を1つのトランザクションにまとめられるので、途中で落ちても中途半端なデータが残りません。

実行環境とpip install（Python 3.10以上・pandas 3.x系）

Python 3.10以上：この記事のコード自体は requests と pandas で動きます。ただ、2026年10月リリースの Selenium 4.51.0 と Playwright 1.64.0 はどちらも Python 3.10 以上が必須になりました。後からブラウザ自動化を足すことを考えると、ここで揃えておくのが無難です。
pandas 3.x系：調査時点の最新安定版は 3.0.6 です。3.x では文字列カラムの既定の dtype が object から専用の str に変わりましたが、to_sql() で TEXT 列に保存する分には影響しません。
SQLAlchemy は 2.0 以上を指定：この記事は 2.0 の書き方が前提です。古い版が入っていると、旧来のコードが手元では動いてしまい、移行漏れに気づけません。

python3 -m venv .venv
. .venv/bin/activate
pip install "sqlalchemy>=2.0" pandas requests beautifulsoup4
pip show sqlalchemy pandas        # 実際に入ったバージョンを確認
python3 -c "import sqlite3; print(sqlite3.sqlite_version)"   # ON CONFLICT が使えるSQLiteか確認

ON CONFLICT（重複したときの動作を指定する構文）は、古いSQLiteにはありません。Pythonに組み込まれたSQLiteのバージョンは、上の最後のコマンドで確認できます。公式ドキュメントの UPSERT の項に対応バージョンが載っています。

to_sql()とwith engine.begin()でSQLiteへ毎日追記し、ON CONFLICTで重複を弾くコード

一覧ページから記事のURLとタイトルを取り、news.db に追記する完全なスクリプトです。取得先は説明用に example.com にしています。

# collect.py
import sys
from datetime import datetime, timezone
from urllib.parse import urljoin
from urllib.robotparser import RobotFileParser

import pandas as pd
import requests
from bs4 import BeautifulSoup
from sqlalchemy import create_engine, text
from sqlalchemy.dialects.sqlite import insert as sqlite_insert

BASE = "https://example.com"
LIST_URL = f"{BASE}/news/"
DB_URL = "sqlite:////home/username/scraper/news.db"
HEADERS = {"User-Agent": "my-collector/1.0 (+https://example.com/contact)"}

SCHEMA = text("""
CREATE TABLE IF NOT EXISTS articles (
    url        TEXT PRIMARY KEY,
    title      TEXT NOT NULL,
    scraped_at TEXT NOT NULL
)
""")


def allowed(url):
    rp = RobotFileParser(f"{BASE}/robots.txt")
    rp.read()
    return rp.can_fetch(HEADERS["User-Agent"], url)


def fetch_articles():
    res = requests.get(LIST_URL, headers=HEADERS, timeout=30)
    res.raise_for_status()
    soup = BeautifulSoup(res.text, "html.parser")
    now = datetime.now(timezone.utc).isoformat(timespec="seconds")
    rows = []
    for a in soup.select("article h2 a"):
        href = a.get("href")
        if not href:
            continue
        rows.append({
            "url": urljoin(LIST_URL, href),
            "title": a.get_text(strip=True),
            "scraped_at": now,
        })
    return pd.DataFrame(rows, columns=["url", "title", "scraped_at"])


def insert_ignore(pd_table, conn, keys, data_iter):
    rows = [dict(zip(keys, row)) for row in data_iter]
    if not rows:
        return
    stmt = sqlite_insert(pd_table.table).on_conflict_do_nothing(
        index_elements=["url"]
    )
    conn.execute(stmt, rows)


def main():
    if not allowed(LIST_URL):
        sys.exit("robots.txt で取得が許可されていません")

    df = fetch_articles()
    if df.empty:
        sys.exit("記事が1件も取れませんでした（HTML構造の変更を疑う）")

    engine = create_engine(DB_URL)
    with engine.begin() as conn:
        conn.execute(SCHEMA)
        before = conn.execute(text("SELECT COUNT(*) FROM articles")).scalar_one()
        df.to_sql("articles", conn, if_exists="append", index=False,
                  method=insert_ignore, chunksize=500)
        after = conn.execute(text("SELECT COUNT(*) FROM articles")).scalar_one()

    print(f"取得 {len(df)} 件 / 新規 {after - before} 件 / 累計 {after} 件")


if __name__ == "__main__":
    main()

取得部分：allowed() と fetch_articles()

allowed() は、取得前に robots.txt を読み、このUser-Agentで一覧ページを取ってよいかを確かめます。fetch_articles() は一覧ページを1回だけ取得し、article h2 a に当たるリンクからURLとタイトルを抜き出します。相対パスのリンクは urljoin() で絶対URLに直しておきます。こうしないと同じ記事が別の文字列として保存され、重複判定をすり抜けます。scraped_at は ISO 形式の文字列で入れています。SQLiteには日時型がないため、文字列にしておくと並べ替えも比較もそのまま効きます。

テーブル定義：SCHEMA を自分で先に作る

to_sql() にテーブルを作らせると、主キーも一意制約も付きません。そうなると ON CONFLICT が判定に使う列がありません。そこで CREATE TABLE IF NOT EXISTS を text() で包み、url を主キーにしたテーブルを先に作っておきます。2日目以降は IF NOT EXISTS によって何も起きません。

重複を弾く関数：insert_ignore()

to_sql() の method= には、(pd_table, conn, keys, data_iter) を受け取る関数を渡せます。pandas は通常の INSERT の代わりにこの関数を呼びます。

pd_table.table：DataFrameの列から作られた SQLAlchemy の Table オブジェクトです。
keys と data_iter：列名と行のデータです。dict(zip(...)) で1行ずつ辞書にします。
sqlite_insert(...).on_conflict_do_nothing(index_elements=["url"])：INSERT ... ON CONFLICT (url) DO NOTHING を組み立てます。既にあるURLは黙って飛ばし、新しいURLだけが入ります。
conn.execute(stmt, rows)：行のリストを渡すと、まとめて実行されます。chunksize=500 により、pandas が500行ずつに分けてこの関数を呼びます。

重複があってもエラーにならないので、その日の新着だけが確実に積み上がります。以前の「保存後に DELETE で掃除する」書き方と比べると、DBに重複が一瞬も入らず、掃除の処理も要りません。

保存部分：with engine.begin() の中で全部やる

テーブル作成、件数の確認、to_sql()、もう一度の件数確認を、すべて1つの begin() ブロックの中で行います。to_sql() の第2引数にエンジンではなく conn を渡すのがポイントです。こうすると pandas の書き込みも同じトランザクションに乗り、ブロックを正常に抜けた時点でまとめてコミットされます。途中で例外が出ればすべてロールバックされます。

前後の COUNT(*) の差が、その日に実際に増えた件数です。2回続けて実行すると、2回目は次のように新規0件になるはずです。重複排除が効いているかは、これで確認できます。

$ python3 collect.py
取得 20 件 / 新規 20 件 / 累計 20 件
$ python3 collect.py
取得 20 件 / 新規 0 件 / 累計 20 件

0件のときに異常終了させる理由

サイトのHTML構造が変わると、セレクタが何にも当たらなくなります。そのまま正常終了すると、何日も「新規0件」が続いても気づけません。sys.exit("...") は終了コード1で終わるので、後述の systemd がこの実行を「失敗」として扱います。CSVに書き出していた処理をこの形へ移す場合は、line_terminator削除とCSV文字化け｜26/10 もあわせて確認しておくと、pandas 側の引数変更で同時に詰まらずに済みます。

取得先サイトへの配慮：robots.txt・利用規約・アクセス間隔

robots.txt は取得前に毎回確認する：上の allowed() がその役目です。Googleは2026年9月の更新で、GPTBot などのクローラーを User-agent ごとに制御する例を示しています。サイト運営者がボットの種類ごとに細かく許可・拒否を書く流れが強まっているので、自分のUser-Agentに当たる行も見落とさないようにします。
robots.txt だけで判断しない：スクレイピングの適法性は、利用規約・著作権法・不正競争防止法・不正アクセス禁止法などが重なり合う領域です。利用規約で禁止されていれば対象を変えるか、運営者に許諾を取ります。ログインが必要なページや個人情報を含むページは、特に慎重に判断します。
アクセスは最小限にする：このスクリプトは一覧ページを1日1回取るだけです。詳細ページまで辿る場合は、1件ごとに time.sleep() で数秒以上空けて、並列化はしません。429（リクエストが多すぎる）が返ったら、その日は打ち切ります。
身元を示すUser-Agentを付ける：連絡先URLを入れておけば、運営者が問題を感じたときに止める手段があります。ブロックされた場合も、回避する方向には進まず、公式APIの有無を確認するか取得をやめます。

systemd timerで毎日止めずに動かす設定と、手元PC・共有サーバーcron・VPSの止まる条件

systemd timer のユニット

~/.config/systemd/user/ に、次の2ファイルを置きます。

# ~/.config/systemd/user/collect.service
[Unit]
Description=news collector

[Service]
Type=oneshot
WorkingDirectory=/home/username/scraper
ExecStart=/home/username/scraper/.venv/bin/python /home/username/scraper/collect.py

# ~/.config/systemd/user/collect.timer
[Unit]
Description=run news collector daily

[Timer]
OnCalendar=*-*-* 07:00:00
Persistent=true

[Install]
WantedBy=timers.target

systemctl --user daemon-reload
systemctl --user enable --now collect.timer
loginctl enable-linger username        # ログアウト中もユーザーのtimerを動かす
systemctl --user list-timers           # 次回実行時刻を確認
journalctl --user -u collect.service   # print の出力（取得・新規・累計）を確認

ExecStart は venv 内の python を絶対パスで指定します。timer から起動するプロセスは、ログインシェルの PATH や activate の状態を引き継ぎません。
Type=oneshot のサービスは、実行中に timer が次の起動時刻を迎えても二重には起動しません。同じSQLiteファイルへ2本が同時に書き込む事故を防げます。
Persistent=true を付けると、電源断などで逃した回を、起動時にまとめて1回実行します。ON CONFLICT で重複を弾いているので、遅れて実行されてもデータは壊れません。
enable-linger がないと、ログアウトした時点でユーザーの timer が止まります。

先ほど0件で異常終了させたのは、ここで効いてきます。失敗をSlackに知らせる設定は systemd OnFailureでSlack通知｜26年10月にまとめています。OnFailure= を1行足せば、HTML構造の変更に翌朝気づけます。

どこで動かすか：止まる条件で選ぶ

実行場所	スケジューラ	向いている条件	止まる条件
手元のPC	launchd（Mac）／systemd timer（Linux）	試作中、数日だけ集めたい	スリープ中・フタを閉じている間・電源を切った日。launchd は復帰時に発火するので、実行時刻が読めない
共有レンタルサーバー	cron	requests と BeautifulSoup だけで取れる静的なページ	ブラウザを入れられない環境では、Selenium や Playwright に移行した時点で動かせない。実行時間やプロセスの制限に当たると途中で打ち切られる。Pythonのバージョンを選べない場合、3.10 未満だと最新の Selenium と Playwright が入らない
VPS・常時稼働の自宅サーバー	systemd timer	毎日確実に、ブラウザ自動化も含めて動かしたい	電源断やネットワーク断。`Persistent=true` で逃した回は取り戻せるが、ディスクが一杯になると SQLite への書き込みが失敗する

共有サーバーで cron を使う場合も、考え方は同じです。

0 7 * * * cd /home/username/scraper && .venv/bin/python collect.py >> collect.log 2>&1

cron には、逃した回を取り戻す仕組みも、二重起動を防ぐ仕組みもありません。前者は ON CONFLICT があるので翌日の実行で取り戻せます。後者が心配なら flock で包みます。将来ページが動的化して Selenium に切り替えるなら、ブラウザの後始末も含めて Selenium with文とsystemdで後始末｜26年10月の形にしておくと、長く回してもプロセスが溜まりません。

まとめ：engine.execute()削除後のスクレイピング保存は「begin・to_sql・ON CONFLICT」の3点で書く

SQLAlchemy 2.0 では engine.execute() が削除され、文字列のSQLは text() で包む必要があります。engine.connect() に置き換えるだけの修正は、commit を忘れると黙ってロールバックされるので危険です。今の書き方は次の3点です。with engine.begin() でトランザクションを開く。その中で to_sql() にコネクションを渡す。method= に on_conflict_do_nothing() を使った関数を渡し、主キーで重複を弾く。こうすれば、同じページを毎日取っても新着だけが積み上がり、遅れて実行されてもデータは壊れません。取得は robots.txt と利用規約を確認したうえで1日1回に絞ります。0件のときは異常終了させ、systemd timer の Persistent=true と enable-linger で止めずに回します。実行場所は、ブラウザ自動化に移る可能性があるなら、共有サーバーの cron ではなく常時稼働の VPS や自宅サーバーを選ぶのが確実です。

engine.execute削除→to_sql追記｜26年10月

SQLAlchemy 2.0でengine.execute()が使えない理由と、今の保存コードの形

以前はこう書いていた

「engine.connect()に置き換えるだけ」の修正は危ない

今はこう書く

実行環境とpip install（Python 3.10以上・pandas 3.x系）

to_sql()とwith engine.begin()でSQLiteへ毎日追記し、ON CONFLICTで重複を弾くコード

取得部分：allowed() と fetch_articles()

テーブル定義：SCHEMA を自分で先に作る

重複を弾く関数：insert_ignore()

保存部分：with engine.begin() の中で全部やる

0件のときに異常終了させる理由

取得先サイトへの配慮：robots.txt・利用規約・アクセス間隔

systemd timerで毎日止めずに動かす設定と、手元PC・共有サーバーcron・VPSの止まる条件

systemd timer のユニット

どこで動かすか：止まる条件で選ぶ

関連する選択肢

まとめ：engine.execute()削除後のスクレイピング保存は「begin・to_sql・ON CONFLICT」の3点で書く

手を動かして詰まったところまで書いた記事

PC作業効率アップおすすめガジェット

SQLAlchemy 2.0でengine.execute()が使えない理由と、今の保存コードの形

以前はこう書いていた

「engine.connect()に置き換えるだけ」の修正は危ない

今はこう書く

実行環境とpip install（Python 3.10以上・pandas 3.x系）

to_sql()とwith engine.begin()でSQLiteへ毎日追記し、ON CONFLICTで重複を弾くコード

取得部分：allowed() と fetch_articles()

テーブル定義：SCHEMA を自分で先に作る

重複を弾く関数：insert_ignore()

保存部分：with engine.begin() の中で全部やる

0件のときに異常終了させる理由

取得先サイトへの配慮：robots.txt・利用規約・アクセス間隔

systemd timerで毎日止めずに動かす設定と、手元PC・共有サーバーcron・VPSの止まる条件

systemd timer のユニット

どこで動かすか：止まる条件で選ぶ

関連する選択肢

まとめ：engine.execute()削除後のスクレイピング保存は「begin・to_sql・ON CONFLICT」の3点で書く

手を動かして詰まったところまで書いた記事

PC作業効率アップ おすすめガジェット

PC作業効率アップおすすめガジェット