create_allは黙って列を無視する — データが入った後からAlembicを導入する crossbar_telepath 開発記 (4)

通話の心理をリアルタイム分析するプロダクトを作っている。前回までで、呼ごとの録音と文字起こしが貯まるようになった。次は感情判定だ——という段になって、こんな迷いが出た。

「テナントIDとユーザーIDを、今のうちにDBに付けておくか。まだ早いかな?」

まだ売れてもいないプロダクトに多テナント用の列を足すのは、どう考えても早い。それでも足したくなったのは、後から列を足すのが怖かったからだ。そしてこの「怖さ」の正体を掘ったら、そもそもスキーマ変更ができない状態で開発していたことが判明した。本稿はその修復記である。

発端: その列は追加されなかった

DBはSQLAlchemyで組んでいて、起動時にこうしていた。

def init_db() -> None:
    """テーブルを作る(既にあれば何もしない)。"""
    metadata.create_all(engine())

「既にあれば何もしない」——ここが罠だった。念のため実験してみる。既存の calls テーブルに無い列をモデルに足して、create_all を呼ぶ。

db.calls.append_column(Column('tenant_id', Integer))
db.metadata.create_all(db.engine())
print([c['name'] for c in inspect(db.engine()).get_columns('calls')])

結果:

create_all後の実テーブルの列: ['contact_id', 'label', 'customer_number',
                              'started_at', 'ended_at', 'max_anger', 'summary']
tenant_id は追加された? False

エラーも警告も出ない。黙って無視される。 create_all は「無いテーブルを作る」だけの道具で、既存テーブルを変更する機能を持っていない。

つまりこの時点の私は、スキーマを変えるには「テーブルを消して作り直す(データを捨てる)」か「手でALTER TABLEを打つ」しか手段がなかった。しかも手元には、国際電話代を払って架けて、うち2本はKVSの保持期限と競争しながら救出した通話録音が入っている。捨てられるわけがない。

「怖いから今のうちに列を足しておこう」という発想は、この制約が生んだものだった。実際、私は既に max_anger や anger_score といったまだ使わない列を、後で足すのが怖いという理由だけで先回りして入れていた。

ALTERをテキストで積むのでは駄目なのか

素朴な代案として、ALTER文をSQLファイルに書き溜めていく運用がある。実際よく見る。だがこれには決定的な欠陥がある。

そのファイルは「やるべきことのリスト」であって「やったことの記録」ではない。

ファイルを見ても「この変更を当てたのか?」に答えられない。開発機では実行した、ステージングでは誰かが手動で流した、本番は忘れた——という状態になった瞬間、真のスキーマが誰にも分からなくなる。いわゆるスキーマドリフトで、管理しているつもりの状態が一番危ない。

必要なのは手順書だけではなく、実施記録と切り戻し手順を含めた三点セットだ。電話網で局データを変更するときと同じで、稼働中の設備を全消去して全ロードし直す人はいない。差分だけを当て、記録を残し、駄目なら戻す。

これをやるのがマイグレーションツールで、Pythonなら Alembic になる。

本題: データが入った後から導入する

Alembicの入門記事はたいてい「まっさらなDBから始める」前提で書かれている。だが現実には、動いているものに後から入れたい場面のほうが多い。手順を残しておく。

1. 初期化とenv.pyの調整

uv add alembic
uv run alembic init migrations

生成される migrations/env.py は、接続先を alembic.ini から読む。だがアプリ側は .env の DATABASE_URL を見ている。設定の出どころが二重になるのは事故のもとなので、env.py がアプリの設定を読むように書き換えた。

import config as app_config
from db import metadata

alembic_config.set_main_option("sqlalchemy.url", app_config.DATABASE_URL)
target_metadata = metadata

これで alembic.ini に接続情報を書かずに済む。iniはgit管理下なので、秘密を置かない構造にしておくのは大事だ。

2. 基準線を作る

現在のモデル定義から、最初のマイグレーションを自動生成する。

uv run alembic revision --autogenerate -m "baseline: calls and utterances"

ここで生成されるのは「calls と utterances を作る」マイグレーションだ。だがDBにはもう両方存在する。このまま upgrade すると「テーブルが既にある」と怒られる。

3. stamp——ここが肝

そこで使うのが stamp だ。

uv run alembic stamp head

これはマイグレーションを実行せずに「適用済みである」とだけ記録するコマンドである。DBには alembic_version というテーブルが作られ、現在位置が書き込まれる。

SELECT * FROM alembic_version;
 version_num
--------------
 d32f4d43158a

つまり「いまのDBの状態は、baselineマイグレーションを適用した結果と等しい」と宣言したわけだ。既存データには一切触っていない。これが後付け導入の全てで、あとは通常運転に合流できる。

4. 2本目で動作を確かめる

基準線ができたら、実際の変更を1つ流してみるのが確実だ。ちょうど保存したい列があった——コールフローから呼ばれるLambdaが instance_arn(どのConnectインスタンスから来た呼か)を送ってきているのに、受信側が捨てていた。

モデルに列を足して、生成する。

uv run alembic revision --autogenerate -m "add instance_arn to calls"

INFO  [alembic.autogenerate.compare.tables] Detected added column 'calls.instance_arn'
INFO  [alembic.autogenerate.compare.constraints] Detected added index 'ix_calls_instance_arn'

差分だけが検出された。中身はこうなる。

def upgrade():
    op.add_column('calls', sa.Column('instance_arn', sa.Text(), nullable=True))
    op.create_index(op.f('ix_calls_instance_arn'), 'calls', ['instance_arn'])

def downgrade():
    op.drop_index(op.f('ix_calls_instance_arn'), table_name='calls')
    op.drop_column('calls', 'instance_arn')

適用すると、既存の4呼のデータを保ったまま列が増えた。連なりも記録される。

<base> -> d32f4d43158a, baseline: calls and utterances
d32f4d43158a -> 63d81632375d (head), add instance_arn to calls

5. 起動時に自動で追いつかせる

開発の手軽さのため、init_db() をマイグレーション実行に置き換えた。

def init_db() -> None:
    cfg = Config(str(BASE_DIR / "alembic.ini"))
    cfg.attributes["configure_logger"] = False  # アプリのログ設定を壊さない
    command.upgrade(cfg, "head")

configure_logger = False は地味だが重要で、これを渡さないとAlembicが alembic.ini のログ設定でアプリのロガーを上書きしてしまう。

これでサーバーを立ち上げ直すだけでスキーマが追いつく。ただしこの形が成立するのは単一インスタンスのうちだけだ。Fargateでタスクを複数立てる日には、起動が競合して同じマイグレーションを取り合う可能性がある。その段階ではデプロイ時に一度だけ流す形へ移す必要がある。今は1タスクなので、手軽さを取っている。

何が変わったか

投機的にスキーマを膨らませる必要がなくなった。 これが一番大きい。

冒頭の「テナントIDを今のうちに」という迷いは、結局「入れない」で決着した。理由は簡単で、必要になった日にマイグレーション1本で足せるからだ。多テナントの本当の難所は列を足すことではなく、全クエリが必ずテナントで絞られる保証を作ることで、列だけ先に置いても安全性は1ミリも増えない。NULLのまま何ヶ月も残るだけだ。

マイグレーションがない環境では、人は必ず防衛的に列を先回りさせる。YAGNIを実行可能にする装置という側面が、便利さ以上に効く。

もう一つは、スキーマ変更がコードレビューの対象になったこと。マイグレーションは、それを必要とするコードと同じコミットに乗る。列の追加とその列を使うコードが噛み合っているかを同時に判断できるし、コミットを戻すときどのマイグレーションを戻せばいいかも自明になる。

危険な変更を実行前に止められる、というのも実利だ。autogenerateは列名の変更を「削除+追加」と誤認する。生成物に drop_column が出ていればレビューで気づけるが、psqlのプロンプトで手打ちしていたらデータが消えてから気づくことになる。Alembic自身も生成物に please adjust! とコメントを残す。自動生成を信じすぎるな、ということだ。

導入直後にやらかした話

最後に、同じ日にやらかした失敗を書いておく。

別件のバグ調査で、過去の呼を再処理すると終了時刻が実行時刻で上書きされることが判明した。40秒の通話が3631秒に、ひどいものは5日間になっていた。修正を入れたうえで、既存データを録音の長さから復元した——その場限りのスクリプトで。

結果どうなったか。DBは直った。だが移行元のJSONファイルは壊れた値のまま残っていて、そこからDBを作り直すと壊れた値が復活する。

これは本来マイグレーションとして書くべき変更だった。そうすればgitに残り、どの環境でもちょうど一度だけ適用され、他の人が同じDBを立てても自動的に修復済みになる。データの修正もスキーマ変更と同じ「変更履歴」の一部なのに、道具を入れた当日に自分でその原則を破った。

Alembicを入れる意味は「列を安全に足せる」ことだと思われがちだが、実際にはもう少し広い。一文で言うならこうだ。

データを保ったままDBを変更し、その変更をすべての環境で同じ順序でちょうど一度ずつ再現するための仕組み

「ちょうど一度ずつ」の部分を、私は当日に取りこぼした。

まとめ

• create_all は既存テーブルに列を足せない。エラーも出ずに無視される
• ALTER文をテキストで積むのは手順書であって実施記録ではない。適用済みかどうかに答えられない時点で管理になっていない
• 後付け導入の手順は init → env.pyを設定と一本化 → baselineをautogenerate → alembic stamp で既存DBを基準化 → 以降は通常運転
• 得られる一番の効能は「後から足せるので、今足さなくていい」という身軽さ
• autogenerateは列名変更を検出できない。生成物は必ず読む
• データの修正もマイグレーションで書く(当日にやらかした)

次回はいよいよ本題、相手がキレているかどうかをAIに判定させるフェーズに入る。スキーマが変えられるようになったので、判定結果の持ち方を試しながら決められる。