商品の在庫を正しく管理するには、数量の記録だけでなく「いつ・どれだけ変わったか」の履歴も欠かせません。この記事では、SQLAlchemyでテーブルを定義し、Alembicでマイグレーションを段階的に適用し、SQLiteに保存する在庫管理CLIを実際に作りました。
argparseのmigrate・add・list・update・historyという5つのサブコマンドで、商品登録から在庫数の更新、更新履歴の確認までを一通り動かします。初心者でも追体験できるよう、モデル定義やマイグレーションの要点を1つずつ噛み砕いて解説します。
SQLAlchemyの基本概念、要件定義、実装、動作確認までを順番に学べる構成です。動画は目次から確認したい場面へ移動でき、本文だけでも手順と考え方が完結します。
動画の内容をテキストで確認する
オープニング。SQLAlchemyとAlembicを使って在庫管理CLIを作るカリキュラムを始めます。概要紹介。
SQLAlchemyとAlembicの役割と使い方を学ぶ 在庫管理CLIの要件を整理する 完成コードと実行結果を確認する 最後に実コマンドとファイル状態で完成挙動を確かめる 具体的にやること。
未適用のリビジョンだけを順に反映するmigrateコマンド 商品コードと商品名と在庫数を登録するaddコマンド 登録済みの在庫を商品コード順に表示するlistコマンド 商品コードを指定して在庫数を更新するupdateコマンド 商品コードを指定して更新履歴を表示するhistoryコマンド 実装環境・必須アプリ。
OS：Windows 11 Pro Python：3.13.3シェル：PowerShell 5.1必須アプリ：コードエディター、ターミナル、エクスプローラー パッケージ：pip、SQLAlchemy、Alembic SQLAlchemyとAlembicとは。
SQLAlchemy：SQLAlchemyは、PythonのクラスとデータベースのテーブルをひもづけてSQLを直接書かずにデータを操作できるORMライブラリAlembic：Alembicは、データベースのテーブル構成の変更をリビジョンという単位で段階的に適用するマイグレーションツール SQLAlchemyで作る在庫管理CLIの要点。
更新前の在庫数をbefore_quantityとして控える 新しい在庫数をafter_quantityとして記録する 更新理由をreasonへ保存して後から追える形にする 在庫管理CLIの要件定義。
migrateで0001と0002の2件のリビジョンが適用されることaddで登録した在庫がlistに商品コード順で表示されることupdateで在庫数を変えると変更前後の数量が履歴に残ることhistoryで指定した商品コードの更新履歴が確認できること 同じ商品コードを再登録すると登録済みと通知されること 在庫数に負の値を渡すと例外で処理が止まること INTRO: Monaco Editorで在庫管理CLIを実装。
コードを1行ずつ入力し、補完と自動インデントを使いながら実行結果を確認します。LINE 001: モジュール概要の説明開始。このファイル全体がSQLAlchemy・Alembic・SQLiteを使った在庫管理CLIであることを示すドキュメント文字列の始まりです。
ファイルの目的を最初に読み手へ伝える役割があります。LINE 003: 機能一覧の見出し。このあとに続く機能一覧を紹介するための見出し文です。
CLIが提供する主要な処理を箇条書きで示す準備をしています。LINE 004: マイグレーション機能の説明。AlembicによるDBマイグレーション適用機能があることを説明しています。
データベースの構造を最新の状態に保つための機能です。LINE 005: 登録・一覧表示機能の説明。在庫データの登録と一覧表示ができる機能があることを説明しています。
CLIの基本的な操作の一つを示しています。LINE 006: 在庫数更新機能の説明。商品コードを指定して在庫数を更新できる機能があることを説明しています。
在庫の変動を反映するための機能です。LINE 007: 更新履歴確認機能の説明。在庫がどのように更新されたかの履歴を確認できる機能があることを説明しています。
過去の変更を追跡するための機能です。LINE 008: ドキュメント文字列の終了。モジュール全体の説明文字列を閉じています。
ここまでの内容がファイル冒頭のコメントとして扱われます。LINE 010: argparseモジュールの読み込み。コマンドライン引数を扱うための標準ライブラリargparseを読み込んでいます。
CLIの引数解析に使われます。LINE 012: MigrationContextの読み込み。Alembicのマイグレーション処理を実行するためのMigrationContextクラスを読み込んでいます。
DB接続とマイグレーションを結びつけるために使います。LINE 013: Operationsクラスの読み込み。テーブル作成などのマイグレーション操作を行うOperationsクラスを読み込んでいます。
実際のスキーマ変更処理で利用します。LINE 014: SQLAlchemyからの複数要素の読み込み開始。SQLAlchemyパッケージから必要な複数の要素をまとめて読み込む処理の始まりです。
以降の行で個々の要素を指定しています。LINE 015: Column要素の読み込み。テーブルの列を定義するためのColumnクラスを読み込んでいます。
マイグレーション処理でテーブル構造を指定する際に使います。LINE 016: ForeignKey要素の読み込み。外部キー制約を定義するためのForeignKeyクラスを読み込んでいます。
テーブル間の関連付けに使用します。LINE 017: Integer型の読み込み。整数型を表すIntegerクラスを読み込んでいます。
数値を扱う列の型指定に使います。LINE 018: String型の読み込み。文字列型を表すStringクラスを読み込んでいます。
文字列を扱う列の型指定に使います。LINE 019: create_engine関数の読み込み。データベース接続を作成するためのcreate_engine関数を読み込んでいます。
SQLiteへの接続を作る際に使います。LINE 020: select関数の読み込み。データ取得のクエリを組み立てるためのselect関数を読み込んでいます。
データベースから在庫データを検索する際に使用します。LINE 021: インポート文の終了。SQLAlchemyから複数要素をまとめて読み込む処理を閉じる括弧です。
ここまでの要素が使えるようになります。LINE 022: ORM関連要素の読み込み。SQLAlchemyのORM機能であるDeclarativeBase、Mapped、Session、mapped_columnをまとめて読み込んでいます。
モデル定義とデータベース操作の両方に使う要素です。LINE 025: 基底クラスの定義開始。SQLAlchemyの宣言的マッピングで使う基底クラスBaseを定義しています。
以降のモデルクラスはこのクラスを継承します。LINE 026: 基底クラスの説明文。Baseクラスがどのような役割を持つかを説明するドキュメント文字列です。
宣言的マッピングの基盤であることを示しています。LINE 029: 在庫商品モデルの定義開始。在庫商品を表すInventoryItemクラスをBaseクラスを継承して定義しています。
このクラスがinventory_itemsテーブルに対応します。LINE 030: 在庫商品モデルの説明文。InventoryItemクラスが商品コード、商品名、現在庫数を保持することを説明するドキュメント文字列です。
LINE 032: テーブル名の指定。InventoryItemモデルが対応するテーブル名をinventory_itemsとして指定しています。データベース上の実テーブル名を決める設定です。
LINE 034: 主キー列の定義。idという整数型の主キー列を定義しています。各在庫データを一意に識別するための列です。
LINE 035: 商品コード列の定義。skuという商品コードを表す文字列型の列を定義しています。重複不可かつ必須の制約が付いており、商品を一意に識別する役割を持ちます。
LINE 036: 商品名列の定義。nameという商品名を表す文字列型の列を定義しています。必須項目として設定されており、商品の名称を保持します。
LINE 037: 在庫数列の定義。quantityという在庫数を表す整数型の列を定義しています。必須項目として設定されており、現在の在庫数量を保持します。
RUN 1/8: 在庫商品モデルの列を確認する。InventoryItemモデルの定義が完成した状態です。__tablename__とMapped型で宣言した列をprintで確認します。
CHECK 1/8: 途中実行に成功。inventory_items ['id', 'sku', 'name', 'quantity'] RETURN 01: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。
LINE 040: 在庫変更履歴モデルの定義開始。在庫の変更履歴を表すStockChangeクラスをBaseクラスを継承して定義しています。このクラスがstock_changesテーブルに対応します。
LINE 041: 在庫変更履歴モデルの説明文。StockChangeクラスが在庫更新前後の数量と理由を履歴として保持することを説明するドキュメント文字列です。LINE 043: 履歴テーブル名の指定。
StockChangeモデルが対応するテーブル名をstock_changesとして指定しています。在庫更新の記録を保存する実テーブル名です。LINE 045: 履歴の主キー列の定義。
idという整数型の主キー列を定義しています。各更新履歴を一意に識別するための列です。LINE 046: 関連商品ID列の定義開始。
item_idという整数型の列を定義し始めています。どの在庫商品に対する変更かを結びつけるための列です。LINE 047: 外部キー制約の指定。
item_id列がinventory_itemsテーブルのidを参照する外部キーであることを指定しています。必須項目としても設定されています。LINE 048: item_id列定義の終了。
item_id列の定義を閉じる括弧です。ここまでで在庫商品との関連付けが完成します。LINE 049: 更新前数量列の定義。
before_quantityという更新前の在庫数を表す整数型の列を定義しています。変更履歴として必須項目です。LINE 050: 更新後数量列の定義。
after_quantityという更新後の在庫数を表す整数型の列を定義しています。変更履歴として必須項目です。LINE 051: 更新理由列の定義。
reasonという更新理由を表す文字列型の列を定義しています。なぜ在庫数を変更したのかを記録する必須項目です。RUN 2/8: 在庫更新履歴モデルの列を確認する。
在庫更新履歴を表すStockChangeモデルまで定義できた状態です。テーブル名と列の一覧を出力して構造を確かめます。CHECK 2/8: 途中実行に成功。
stock_changes ['id', 'item_id', 'before_quantity', 'after_quantity', 'reason'] RETURN 02: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。LINE 054: 初回マイグレーション関数の定義開始。
初回リビジョンの処理を行う_upgrade_0001関数を定義しています。Operationsを受け取り、テーブル作成処理を実行します。LINE 055: 初回マイグレーションの説明文。
この関数が初回リビジョンで在庫商品テーブルを作ることを説明するドキュメント文字列です。LINE 056: テーブル作成処理の開始。op.create_tableメソッドを呼び出してテーブルを作成する処理を開始しています。
以降の行で列を指定していきます。LINE 057: 作成対象テーブル名の指定。作成するテーブルの名前としてinventory_itemsを指定しています。
在庫商品を保存するテーブルであることを表します。LINE 058: id列の定義。主キーとなるid列を整数型で定義しています。
各在庫レコードを一意に識別するための列です。LINE 059: sku列の定義。商品コードを表すsku列を文字列型で定義しています。
重複不可かつ必須の制約が付いています。LINE 060: name列の定義。商品名を表すname列を文字列型で定義しています。
必須項目として設定されています。LINE 061: quantity列の定義。在庫数を表すquantity列を整数型で定義しています。
必須項目として設定されています。LINE 062: テーブル作成処理の終了。op.create_tableの呼び出しを閉じる括弧です。
ここまでの列定義でinventory_itemsテーブルの作成処理が完成します。LINE 065: 2番目のマイグレーション関数の定義開始。2番目のリビジョンの処理を行う_upgrade_0002関数を定義しています。
Operationsを受け取り、履歴テーブルの作成処理を実行します。LINE 066: 2番目のマイグレーションの説明文。この関数が2番目のリビジョンで在庫更新履歴テーブルを追加することを説明するドキュメント文字列です。
LINE 067: 履歴テーブル作成処理の開始。op.create_tableメソッドを呼び出して履歴テーブルを作成する処理を開始しています。以降の行で列を指定していきます。
LINE 068: 作成対象テーブル名の指定。作成するテーブルの名前としてstock_changesを指定しています。在庫更新の履歴を保存するテーブルであることを表します。
LINE 069: 履歴テーブルのid列の定義。主キーとなるid列を整数型で定義しています。各履歴レコードを一意に識別するための列です。
LINE 070: item_id列定義の開始。どの在庫商品の変更かを結びつけるitem_id列の定義を開始しています。以降の行で型や制約を指定します。
LINE 071: item_id列名の指定。列の名前としてitem_idを指定しています。在庫商品との関連付けに使う列名です。
LINE 072: item_id列の型指定。item_id列の型として整数型Integerを指定しています。IDを数値として扱うための設定です。
LINE 073: item_idの外部キー指定。item_id列がinventory_itemsテーブルのidを参照する外部キーであることを指定しています。在庫商品との紐付けを行います。
LINE 074: item_id列の必須制約指定。item_id列がnullを許可しない必須項目であることを指定しています。必ずどの商品かを紐付ける必要があることを表します。
LINE 075: item_id列定義の終了。item_id列の定義を閉じる括弧です。ここまでで在庫商品との関連付け列が完成します。
LINE 076: before_quantity列の定義。更新前の在庫数を表すbefore_quantity列を整数型で定義しています。必須項目として設定されています。
LINE 077: after_quantity列の定義。更新後の在庫数を表すafter_quantity列を整数型で定義しています。必須項目として設定されています。
LINE 078: reason列の定義。更新理由を表すreason列を文字列型で定義しています。必須項目として設定されています。
LINE 079: 履歴テーブル作成処理の終了。op.create_tableの呼び出しを閉じる括弧です。ここまでの列定義でstock_changesテーブルの作成処理が完成します。
LINE 082: マイグレーション一覧の定義開始。適用すべきマイグレーションをまとめたMIGRATIONSリストの定義を開始しています。リビジョン名と対応する関数の組み合わせを保持します。
LINE 083: 初回マイグレーションの登録。リビジョン名0001_create_inventory_itemsと関数_upgrade_0001を組にしてリストへ登録しています。マイグレーション適用時にこの順番で処理されます。
LINE 084: 2件目のマイグレーション登録。MIGRATIONSリストに2番目の要素として、リビジョン名"0002_add_stock_changes"と対応する更新関数_upgrade_0002のペアを追加しています。この定義により在庫更新履歴テーブルを作る処理が管理対象になります。
LINE 085: マイグレーション一覧の定義終了。MIGRATIONSリストの定義を閉じる括弧です。ここまでで適用すべきマイグレーションの一覧が確定します。
RUN 3/8: 登録したリビジョンを確認する。2つのリビジョンをMIGRATIONSにまとめたところです。登録したリビジョン名を並べて件数と順序を確認します。
CHECK 3/8: 途中実行に成功。2 ['0001_create_inventory_items', '0002_add_stock_changes'] RETURN 03: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。
LINE 088: マイグレーション実行関数の定義。データベースのエンジンを受け取り、未適用のマイグレーションを適用するrun_migrations関数を定義しています。戻り値は適用したリビジョン名のリストです。
LINE 089: 関数の説明文。この関数がAlembicによるDBマイグレーションの適用を担当することを説明するdocstringです。処理内容の理解を助けます。
LINE 090: 適用済みリビジョンの初期化。今回の実行で新たに適用したリビジョン名を記録するための空リストを用意しています。処理の最後にこのリストが戻り値として返されます。
LINE 091: トランザクションの開始。engine.begin()を使ってデータベース接続を開き、この中の処理をひとつのトランザクションとして扱います。エラーが起きた場合は自動的にロールバックされます。
LINE 092: バージョン管理テーブル作成の開始。schema_versionテーブルが存在しない場合に作成するSQL文の実行を開始しています。適用済みリビジョンを記録するための土台になるテーブルです。
LINE 093: CREATE TABLE文の前半。テーブル作成SQL文の前半部分の文字列で、存在しない場合のみschema_versionテーブルを作成することを指定しています。LINE 094: CREATE TABLE文の後半。
revision列をテキスト型の主キーとして定義するSQL文の後半部分です。前の行と結合して1つのCREATE TABLE文になります。LINE 095: テーブル作成呼び出しの終了。
exec_driver_sqlの呼び出しを閉じる括弧です。ここでschema_versionテーブルの作成SQLが実行されます。LINE 096: 適用済みリビジョン集合の開始。
すでに適用されたリビジョン名を重複なく保持するための集合内包表記の記述を開始しています。LINE 097: リビジョン名の取り出し。クエリ結果の各行から先頭の列、つまりリビジョン名だけを取り出す処理です。
集合の要素として使われます。LINE 098: 問い合わせ結果の反復。schema_versionテーブルへの問い合わせ結果を1行ずつ取り出すfor句です。
ここでの各rowが直前の行で使われています。LINE 099: リビジョン取得SQL。schema_versionテーブルから登録済みのリビジョン名をすべて取得するSELECT文です。
LINE 100: 問い合わせ呼び出しの終了。exec_driver_sqlの呼び出しを閉じる括弧で、SELECT文の実行結果を反復可能な形で受け取ります。LINE 101: 適用済みリビジョン集合の完成。
集合内包表記を閉じる括弧です。ここでcompletedという、既に適用済みのリビジョン名を集めた集合が完成します。LINE 102: マイグレーション文脈の構築。
AlembicのMigrationContextを現在のデータベース接続に紐づけて作成しています。この文脈情報がマイグレーション操作の実行に使われます。LINE 103: 操作オブジェクトの作成。
作成した文脈を使ってOperationsオブジェクトを生成しています。このオブジェクトを通じてテーブル作成などのマイグレーション操作を実行します。LINE 104: マイグレーション一覧のループ。
MIGRATIONSに登録された全リビジョンを順番に取り出すループです。リビジョン名と対応する更新関数がそれぞれrevision、upgradeに入ります。LINE 105: 適用済みかどうかの判定。
取り出したリビジョンがすでに適用済み集合completedに含まれているかを確認しています。既に適用済みなら処理をスキップするための条件です。LINE 106: 適用済みリビジョンのスキップ。
すでに適用されているリビジョンの場合はこの回のループ処理を中断し、次のリビジョンへ進みます。LINE 107: マイグレーション処理の実行。未適用のリビジョンに対応する更新関数を呼び出し、テーブル作成などの実際のスキーマ変更を実行しています。
LINE 108: 適用記録の挿入開始。適用したリビジョンをschema_versionテーブルへ記録するためのINSERT文の実行を開始しています。LINE 109: リビジョン挿入SQL。
schema_versionテーブルにリビジョン名を1件挿入するSQL文です。プレースホルダを使い安全に値を渡します。LINE 110: 挿入するリビジョン名の指定。
直前のSQL文のプレースホルダに渡す実際の値として、今回適用したリビジョン名をタプルで指定しています。LINE 111: 挿入処理呼び出しの終了。exec_driver_sqlの呼び出しを閉じる括弧で、リビジョン記録用のINSERT文が実行されます。
LINE 112: 適用済みリストへの追加。今回新たに適用したリビジョン名を、戻り値として返すためのappliedリストに追加しています。LINE 113: 適用結果の返却。
今回のマイグレーションで新たに適用されたリビジョン名の一覧を呼び出し元に返します。RUN 4/8: マイグレーションを実際に適用する。run_migrationsが書けたので、一時的なSQLiteに対して実際にマイグレーションを流し、適用されたリビジョンを表示します。
CHECK 4/8: 途中実行に成功。['0001_create_inventory_items', '0002_add_stock_changes'] RETURN 04: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。
LINE 116: マイグレーションコマンド関数の定義。CLIのmigrateサブコマンドに対応する処理を行うcmd_migrate関数を定義しています。エンジンと引数を受け取り、結果を文字列で返します。
LINE 117: 関数の説明文。未適用のリビジョンだけを順番に反映するという、この関数の役割を説明するdocstringです。LINE 118: マイグレーションの実行呼び出し。
run_migrations関数を呼び出してデータベースに未適用のマイグレーションを適用し、その結果をapplied変数に受け取っています。LINE 119: 適用対象がなかった場合の判定。appliedリストが空、つまり新たに適用されたマイグレーションがなかったかどうかを確認しています。
LINE 120: 最新状態のメッセージ返却。新規適用がなかった場合に、データベースが既に最新の状態であることを示すメッセージを返します。LINE 121: 適用結果のメッセージ返却。
新たに適用されたリビジョン名をカンマ区切りで連結し、実行結果として画面に表示するメッセージを組み立てて返します。RUN 5/8: migrateが冪等に動くか確認する。cmd_migrateが未適用分だけを反映する挙動を確認します。
2回続けて呼び、2回目は最新である旨が返ることを見ます。CHECK 5/8: 途中実行に成功。適用したマイグレーション: 0001_create_inventory_items, 0002_add_stock_changes マイグレーションは最新です RETURN 05: エディターへ戻りました。
入力済みのコードを維持したまま、次の実装へ進みます。LINE 124: 在庫登録コマンド関数の定義。CLIのaddサブコマンドに対応するcmd_add関数を定義しています。
エンジンと引数を受け取り、在庫データの登録処理を行います。LINE 125: 関数の説明文。在庫データの登録機能を担当することを説明するdocstringで、この関数の役割を明確にしています。
LINE 126: 在庫数の下限チェック開始。入力された在庫数が0未満かどうかを判定しています。不正な値を早い段階で見つけるための条件です。
LINE 127: 不正な在庫数のエラー送出。在庫数が負の値だった場合に、エラーメッセージ付きの例外を発生させて処理を中断します。LINE 128: データベースセッションの開始。
Sessionオブジェクトを使ってデータベースとのやり取りを開始しています。withブロックを抜けると自動的にセッションが閉じられます。LINE 129: 既存データ検索の開始。
指定された商品コードが既に登録されているかを確認するための問い合わせ処理を開始しています。LINE 130: 商品コード一致条件の指定。InventoryItemテーブルの中から、引数で渡された商品コードと一致する行を検索する条件を組み立てています。
LINE 131: 検索結果の取得。条件に一致する行が1件だけあればそれを取得し、なければNoneを返すscalar_one_or_noneで結果を受け取っています。LINE 132: 登録済みかどうかの判定。
既存の在庫データが見つかったかどうかを確認し、既に登録されている商品コードかを判断しています。LINE 133: 重複登録時のメッセージ返却。商品コードが既に登録済みだった場合に、その旨を伝えるメッセージを返して処理を終了します。
LINE 134: 新しい在庫データの作成開始。データベースに登録する新しいInventoryItemオブジェクトの生成を開始しています。LINE 135: 商品コードの設定。
引数から受け取った商品コードを、新規作成する在庫データのsku属性に設定しています。LINE 136: 商品名の設定。引数から受け取った商品名を、新規作成する在庫データのname属性に設定しています。
LINE 137: 在庫数の設定。引数から受け取った在庫数を、新規作成する在庫データのquantity属性に設定しています。LINE 138: 在庫データ生成の完了。
InventoryItemオブジェクトの生成処理を閉じる括弧で、ここまでの設定を反映した新規データが確定します。LINE 139: セッションへのデータ追加。作成した新しい在庫データをセッションに登録し、データベースへ反映する準備をしています。
LINE 140: 変更内容のコミット。セッションに登録した変更を確定し、実際にデータベースへ書き込みます。LINE 141: 登録完了メッセージの返却。
登録した商品コード、商品名、在庫数を含む完了メッセージを組み立てて呼び出し元に返します。LINE 144: 在庫一覧コマンド関数の定義。CLIのlistサブコマンドに対応するcmd_list関数を定義しています。
エンジンを受け取り、在庫の一覧表示処理を行います。LINE 145: 関数の説明文。在庫データの一覧表示機能を担当することを説明するdocstringで、この関数の役割を明確にしています。
LINE 146: データベースセッションの開始。Sessionオブジェクトを使って在庫データを取得するためのやり取りを開始しています。LINE 147: 在庫データ取得の開始。
登録されている在庫データを商品コード順に取得するための問い合わせ処理を開始しています。LINE 148: 並び順の指定。InventoryItemテーブルの全件を対象に、商品コードの昇順で並び替える条件を組み立てています。
LINE 149: 結果一覧の取得。問い合わせ結果を全件取得し、在庫データのリストとしてitems変数に格納しています。LINE 150: データ有無の判定。
取得した在庫データが1件も存在しないかどうかを確認しています。LINE 151: データなし時のメッセージ返却。登録された在庫データが1件もなかった場合に、その旨を伝えるメッセージを返します。
LINE 152: 一覧見出しの作成。取得した在庫データの件数を含む見出し行を作り、表示用のlinesリストの最初の要素にしています。LINE 153: 一覧明細行の追加開始。
各在庫データの内容を1行ずつ整形してlinesリストに追加する処理を開始しています。LINE 154: 明細の書式指定。商品コード、商品名、在庫数を区切り文字でつなげた1件分の表示文字列を組み立てています。
LINE 155: 全件分の繰り返し。取得した在庫データitemsを1件ずつ順に取り出し、直前の書式で明細行を生成する繰り返し処理です。LINE 156: リスト生成の閉じ括弧。
extendメソッドに渡すジェネレーター式の閉じ括弧です。ここまでの処理で在庫一覧の各行を表す文字列がlinesに追加されます。LINE 157: 一覧結果を改行結合して返す。
linesに格納された各行の文字列を改行文字で連結し、1つの文字列として呼び出し元に返します。これによりコマンドラインに整形された在庫一覧が表示されます。RUN 6/8: 登録と一覧の流れを確認する。
登録と一覧のコマンドがそろった状態です。空の一覧、登録、登録後の一覧を順に出力して流れを確かめます。CHECK 6/8: 途中実行に成功。
在庫データはありません 在庫を登録しました: A-01ペン 在庫数=10 在庫一覧: 1件 A-01 | ペン | 在庫数=10 RETURN 06: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。LINE 160: 在庫更新コマンドの定義。
商品コードを指定して在庫数を更新するcmd_update関数を定義します。engineとargsを受け取り、更新結果を文字列で返します。LINE 161: 関数の説明を記述するdocstring。
この関数が商品コードを指定して在庫数を更新する処理であることを説明する文書コメントです。関数の役割を明確にする目的で記述しています。LINE 162: 在庫数が負の値でないか確認。
引数で指定された在庫数が0未満かどうかを判定します。負の値であれば不正な入力として後続の処理でエラーにします。LINE 163: 不正な在庫数の場合に例外を発生。
在庫数が0未満だった場合に、その理由を示すメッセージ付きでValueErrorを発生させます。これにより不正なデータの登録を防ぎます。LINE 164: データベース操作用セッションを開始。
engineを使ってSessionを作成し、withブロックでデータベース操作を行います。処理が終わるとセッションは自動的に閉じられます。LINE 165: 対象商品の検索を開始。
更新対象となる商品データをデータベースから取得するためのselect文の実行を開始します。結果はitem変数に格納されます。LINE 166: 商品コードで対象商品を絞り込む。
InventoryItemテーブルの中から、指定された商品コード（sku）と一致するレコードを検索条件として指定します。LINE 167: 検索結果を1件または該当なしとして取得。検索結果が1件であればそのオブジェクトを、該当がなければNoneを返すscalar_one_or_noneメソッドで結果を取得します。
LINE 168: 商品が見つからない場合の判定。検索結果のitemがNoneであるかどうかを確認します。Noneであれば指定された商品コードが登録されていないことを意味します。
LINE 169: 商品未検出時のエラーメッセージを返す。商品コードに該当するデータが見つからなかった場合に、その商品コードを含むエラーメッセージを返して処理を終了します。LINE 170: 更新前の在庫数を保存。
更新処理を行う前の在庫数をbefore変数に保存します。これは更新履歴に記録するために必要な値です。LINE 171: 在庫数を新しい値に更新。
取得したitemオブジェクトのquantity属性に、引数で指定された新しい在庫数を設定します。この時点ではまだデータベースには反映されていません。LINE 172: 更新履歴データの追加を開始。
更新内容を記録するためのStockChangeオブジェクトをセッションに追加する処理を開始します。LINE 173: 更新履歴オブジェクトの生成。在庫更新の履歴として記録するStockChangeオブジェクトを生成します。
各属性に更新前後の数量などの情報を設定していきます。LINE 174: 対象商品のIDを履歴に記録。更新対象となった商品のIDをitem_idとしてStockChangeオブジェクトに設定します。
これにより履歴と商品が結び付けられます。LINE 175: 更新前の在庫数を履歴に記録。更新前に保存しておいたbeforeの値を、履歴データのbefore_quantityとして設定します。
LINE 176: 更新後の在庫数を履歴に記録。引数で指定された新しい在庫数を、履歴データのafter_quantityとして設定します。LINE 177: 更新理由を履歴に記録。
コマンド実行時に指定された更新理由を、履歴データのreasonとして設定します。LINE 178: StockChangeオブジェクトの生成完了。StockChangeオブジェクトへの各属性の設定が完了したことを示す閉じ括弧です。
この後、履歴オブジェクトがセッションに追加されます。LINE 179: session.add呼び出しの終了。session.addメソッドの呼び出しを閉じる括弧です。
これにより生成した履歴オブジェクトがセッションの管理対象になります。LINE 180: 変更内容をデータベースに確定。セッションに保存した在庫数の更新と履歴の追加をまとめてデータベースに確定（コミット）します。
これにより変更内容が永続化されます。LINE 181: 更新結果メッセージの生成を開始。更新処理が成功したことをユーザーに伝えるメッセージ文字列の作成を開始します。
LINE 182: 商品コードを含むメッセージの前半。更新が完了したことと更新対象の商品コードを含む文字列の前半部分を作成します。LINE 183: 更新前後の在庫数と理由を含むメッセージの後半。
更新前の在庫数、更新後の在庫数、更新理由を含む文字列の後半部分を作成し、前半と合わせてメッセージ全体を完成させます。LINE 184: 更新結果メッセージを返す。作成したメッセージ文字列の括弧を閉じ、呼び出し元に更新結果を返します。
LINE 187: 履歴確認コマンドの定義。更新履歴を確認するcmd_history関数を定義します。engineとargsを受け取り、履歴情報を文字列で返します。
LINE 188: 関数の説明を記述するdocstring。この関数が更新履歴を確認する処理であることを説明する文書コメントです。関数の役割を明確にする目的で記述しています。
LINE 189: データベース操作用セッションを開始。engineを使ってSessionを作成し、withブロックでデータベース操作を行います。処理が終わるとセッションは自動的に閉じられます。
LINE 190: 対象商品の検索を開始。履歴を確認する対象となる商品データをデータベースから取得するためのselect文の実行を開始します。LINE 191: 商品コードで対象商品を絞り込む。
InventoryItemテーブルの中から、指定された商品コード（sku）と一致するレコードを検索条件として指定します。LINE 192: 検索結果を1件または該当なしとして取得。検索結果が1件であればそのオブジェクトを、該当がなければNoneを返すscalar_one_or_noneメソッドで結果を取得します。
LINE 193: 商品が見つからない場合の判定。検索結果のitemがNoneであるかどうかを確認します。Noneであれば指定された商品コードが登録されていないことを意味します。
LINE 194: 商品未検出時のエラーメッセージを返す。商品コードに該当するデータが見つからなかった場合に、その商品コードを含むエラーメッセージを返して処理を終了します。LINE 195: 更新履歴データの検索を開始。
対象商品に関する更新履歴をStockChangeテーブルから取得するためのselect文の実行を開始します。LINE 196: 検索対象テーブルを指定。StockChangeテーブルを検索対象として指定し、続く条件で絞り込みを行います。
LINE 197: 対象商品の履歴だけに絞り込む。取得した商品のIDと一致するitem_idを持つ履歴データだけを検索条件として指定します。LINE 198: 履歴を作成順に並べ替える。
検索結果をStockChangeのidの昇順で並び替えることで、履歴が発生した順番に表示できるようにします。LINE 199: 検索結果を一覧として取得。検索結果をオブジェクトのリストとして取得し、changes変数に格納します。
LINE 200: 履歴が存在するかどうかを判定。取得したchangesが空であるかどうかを確認します。空であれば対象商品に更新履歴が存在しないことを意味します。
LINE 201: 履歴なしの場合のメッセージを返す。対象商品に更新履歴がない場合に、その商品コードを含むメッセージを返して処理を終了します。LINE 202: 履歴一覧の見出し行を作成。
商品コードと履歴件数を含む見出し行を作成し、linesリストの最初の要素として設定します。LINE 203: 各履歴行の追加を開始。取得した各履歴データを1行の文字列に変換してlinesリストに追加する処理を開始します。
LINE 204: 履歴番号と更新前数量を含む行の前半。各履歴のIDと更新前の在庫数を含む文字列の前半部分を作成します。LINE 205: 更新後数量と理由を含む行の後半。
各履歴の更新後の在庫数と更新理由を含む文字列の後半部分を作成し、前半と合わせて1行分の文字列を完成させます。LINE 206: changesの各要素を順に処理。取得した履歴データchangesの各要素を1件ずつ順に取り出し、対応する文字列を生成するためのループ条件です。
LINE 207: extendメソッドに渡すジェネレーター式の完了。extendメソッドに渡すジェネレーター式を閉じます。これによりlinesリストにすべての履歴行が追加されます。
LINE 208: 履歴結果を改行結合して返す。linesに格納された見出し行と各履歴行を改行文字で連結し、1つの文字列として呼び出し元に返します。LINE 211: 引数パーサー作成関数の定義。
在庫管理CLIのコマンドライン引数とサブコマンドを定義するbuild_parser関数を定義します。作成したパーサーオブジェクトを返します。LINE 212: 関数の説明を記述するdocstring。
この関数がCLIの引数とサブコマンドを定義する処理であることを説明する文書コメントです。LINE 213: 引数パーサーの生成を開始。argparseモジュールのArgumentParserを生成する処理を開始します。
CLIの全体的な説明などをこの後の行で設定します。LINE 214: CLI全体の説明文を設定。コマンドライン上で表示されるこのツール全体の説明文を設定します。
ヘルプ表示時にこの内容が表示されます。LINE 215: ArgumentParser生成処理の完了。ArgumentParserの生成に必要な設定を閉じる括弧です。
この結果がparser変数に格納されます。LINE 216: データベースパス指定オプションの追加を開始。CLI実行時にデータベースファイルのパスを指定できる--dbオプションの追加を開始します。
LINE 217: デフォルト値とヘルプ文を設定。--dbオプションが指定されなかった場合のデフォルト値と、ヘルプ表示時に表示する説明文を設定します。LINE 218: add_argument呼び出しの完了。
--dbオプションの追加処理を閉じる括弧です。これによりCLI実行時にデータベースパスを指定できるようになります。LINE 219: サブコマンド用のパーサーを準備。
migrate・add・list・update・historyといったサブコマンドを追加するためのcommandsオブジェクトを作成します。commandは必須項目として設定されます。LINE 220: マイグレーションコマンドの追加。
DBマイグレーションを適用するためのmigrateサブコマンドを追加し、ヘルプ表示用の説明文を設定します。LINE 222: 在庫登録コマンドの追加。在庫データを登録するためのaddサブコマンドを追加し、ヘルプ表示用の説明文を設定します。
以降でこのコマンドに必要な引数を追加していきます。LINE 223: 商品コード引数の追加。addコマンドに商品コードを指定する--sku引数を追加します。
必須項目として設定されており、指定しない場合はエラーになります。LINE 224: 商品名引数の追加。addコマンドに商品名を指定する--name引数を追加します。
必須項目として設定されており、指定しない場合はエラーになります。LINE 225: 在庫数引数の追加。addコマンドに在庫数を指定する--quantity引数を追加します。
整数型として必須項目に設定されており、指定しない場合はエラーになります。LINE 227: 在庫一覧コマンドの追加。在庫一覧を表示するためのlistサブコマンドを追加し、ヘルプ表示用の説明文を設定します。
このコマンドには追加の引数はありません。LINE 229: updateサブコマンドの追加。commandsに「update」というサブコマンドを追加しています。
これにより、コマンド実行時に「update」を指定すると在庫数を更新する処理を呼び出せるようになります。LINE 230: 商品コード引数の定義。updateサブコマンドに「sku」という必須の位置引数を追加しています。
この値で、どの商品の在庫を更新するかを指定します。LINE 231: 更新後の在庫数引数の定義。「--quantity」というオプション引数を追加し、更新後の在庫数を整数として受け取れるようにしています。
required=Trueなので、この引数を指定しないとエラーになります。LINE 232: 更新理由引数の定義。「--reason」というオプション引数を追加し、在庫を更新する理由を文字列で受け取れるようにしています。
この値は後で履歴として保存されます。LINE 234: historyサブコマンドの追加。commandsに「history」というサブコマンドを追加しています。
これにより、コマンド実行時に「history」を指定すると更新履歴を表示する処理を呼び出せるようになります。LINE 235: 履歴用の商品コード引数の定義。historyサブコマンドに「sku」という必須の位置引数を追加しています。
この値で、どの商品の更新履歴を表示するかを指定します。LINE 236: パーサーオブジェクトの返却。ここまで設定してきたパーサーをbuild_parser関数の呼び出し元に返しています。
これにより、コマンドライン引数の解析処理を他の場所から利用できるようになります。RUN 7/8: サブコマンドの引数解析を確認する。build_parserでサブコマンドと引数を定義したところです。
addとupdateの引数を解析し、取り込まれた値を表示します。CHECK 7/8: 途中実行に成功。add A-01ペン10 update A-01 3出荷 RETURN 07: エディターへ戻りました。
入力済みのコードを維持したまま、次の実装へ進みます。LINE 239: main関数の定義。CLI全体の実行の起点となるmain関数を定義しています。
argvという引数を受け取れるようにしており、テスト時などにコマンドライン引数を差し替えられるようになっています。LINE 240: 関数の説明を記すドキュメント。この関数が「選択されたサブコマンドを実行して結果を返す」という役割を持つことを説明しています。
コード自体の動作には影響しませんが、読む人の理解を助けます。LINE 241: コマンドライン引数の解析。build_parser関数でパーサーを作成し、parse_argsで実際に渡されたargvを解析しています。
解析結果はargsという変数に格納され、以降の処理で利用されます。LINE 242: データベースエンジンの作成。args.dbで指定されたパスを使って、SQLiteデータベースに接続するためのエンジンを作成しています。
このエンジンは、以降のデータベース操作すべてで使われます。LINE 243: ハンドラー辞書の定義開始。サブコマンド名と、それに対応する処理関数を紐付ける辞書handlersを作り始めています。
この辞書を使うことで、コマンド名に応じた処理を簡潔に呼び分けられます。LINE 244: migrateコマンドの紐付け。「migrate」というコマンド名にcmd_migrate関数を対応付けています。
これにより、コマンドで「migrate」が指定されたときにマイグレーション処理が呼び出されます。LINE 245: addコマンドの紐付け。「add」というコマンド名にcmd_add関数を対応付けています。
これにより、コマンドで「add」が指定されたときに在庫登録処理が呼び出されます。LINE 246: listコマンドの紐付け。「list」というコマンド名にcmd_list関数を対応付けています。
これにより、コマンドで「list」が指定されたときに在庫一覧の表示処理が呼び出されます。LINE 247: updateコマンドの紐付け。「update」というコマンド名にcmd_update関数を対応付けています。
これにより、コマンドで「update」が指定されたときに在庫数の更新処理が呼び出されます。LINE 248: historyコマンドの紐付け。「history」というコマンド名にcmd_history関数を対応付けています。
これにより、コマンドで「history」が指定されたときに更新履歴の表示処理が呼び出されます。LINE 249: ハンドラー辞書の定義終了。サブコマンド名と処理関数を対応付ける辞書handlersの定義を閉じています。
この辞書は次の行で実際の処理呼び出しに使われます。LINE 250: 対応する処理の実行と結果の返却。args.commandに入っているコマンド名をキーにしてhandlers辞書から対応する関数を取り出し、engineとargsを渡して実行しています。
その戻り値がmain関数の結果としてそのまま返されます。RUN 8/8: mainから一連の動作を確認する。mainからサブコマンドを呼び出せる状態です。
migrate、add、listを順に実行して一連の動作を確認します。CHECK 8/8: 途中実行に成功。
適用したマイグレーション: 0001_create_inventory_items, 0002_add_stock_changes 在庫を登録しました: A-01ペン 在庫数=10 在庫一覧: 1件 A-01 | ペン | 在庫数=10 RETURN 08: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。
LINE 253: スクリプト直接実行時の判定。このファイルがモジュールとしてインポートされたのではなく、直接スクリプトとして実行された場合にだけ、以下の処理を実行するようにしています。LINE 254: main関数の呼び出しと結果表示。
main関数を呼び出して処理を実行し、その戻り値である結果メッセージを画面に表示しています。これがCLI実行時の最終的な出力になります。実行1/5: 在庫テーブルを作成する。
SQLiteのデータベースパスを指定してmigrateを実行し、在庫テーブルと履歴テーブルを作成します。確認1/5: 在庫テーブルを作成する。SQLiteのデータベースパスを指定してmigrateを実行し、在庫テーブルと履歴テーブルを作成します。
RETURN 09: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。実行2/5: 商品を登録して一覧を表示する。
商品を1件登録してから一覧を表示し、登録内容が在庫一覧に反映されることを確認します。確認2/5: 商品を登録して一覧を表示する。商品を1件登録してから一覧を表示し、登録内容が在庫一覧に反映されることを確認します。
RETURN 10: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。実行3/5: 在庫数を更新する。
登録済みの商品コードを指定して在庫数を更新し、更新前後の数量が結果に表示されることを確認します。確認3/5: 在庫数を更新する。登録済みの商品コードを指定して在庫数を更新し、更新前後の数量が結果に表示されることを確認します。
RETURN 11: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。実行4/5: 更新履歴を確認する。
在庫数を更新したあとhistoryを実行し、変更前後の数量と理由が履歴として残っていることを確認します。確認4/5: 更新履歴を確認する。在庫数を更新したあとhistoryを実行し、変更前後の数量と理由が履歴として残っていることを確認します。
RETURN 12: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。実行5/5: 重複した商品コードを弾く。
同じ商品コードをもう一度登録しようとして、登録済みと知らされる動作を確認します。確認5/5: 重複した商品コードを弾く。同じ商品コードをもう一度登録しようとして、登録済みと知らされる動作を確認します。
RETURN 13: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。学習内容のまとめ。
DeclarativeBaseを継承したBaseクラスを共通の基底に使う テーブル変更をリビジョン単位に分けて管理する 更新前の在庫数をbefore_quantityとして控える 在庫数が負の値なら例外を送出して登録と更新を防ぐバリデーション 小さく実行確認しながら完成状態まで段階的に組み立てる エンディング。
Python研修はCodeCampでご確認ください。
SQLAlchemyとAlembicとは
今回使用する主要なライブラリについて、役割と使い分けを順番に確認します。
SQLAlchemyのORMでテーブルとクラスを対応させる
SQLAlchemyは、PythonのクラスとデータベースのテーブルをひもづけてSQLを直接書かずにデータを操作できるORMライブラリです。今回はDeclarativeBaseを継承したモデルに列を宣言し、Sessionからinsertやselectといった処理を呼び出します。
SQLの文字列を組み立てる手間が減り、在庫データの読み書きをPythonの記述だけで扱えるようになります。
在庫データを扱うモデル定義とセッション操作で押さえる要素を挙げます。
- DeclarativeBaseを継承したBaseクラスを共通の基底に使う
- mapped_columnでprimary_keyやnullableなどの制約を指定する
- skuにunique制約を付けて商品コードの重複を防ぐ
- Session(engine)をwith文で開いて処理後にcommitする
- selectとwhereで商品コードから該当データを取得する
Alembicのマイグレーションでテーブルを段階的に作る
Alembicは、データベースのテーブル構成の変更をリビジョンという単位で段階的に適用するマイグレーションツールです。この教材ではMigrationContextとOperationsを使い、create_tableでテーブルを作る処理をリビジョンごとに用意します。適用済みのリビジョンを記録しておけば、同じ変更を二重に流す心配なくスキーマを育てられます。
Alembicで在庫テーブルを段階的に育てる際に押さえる要素を挙げます。
- テーブル変更をリビジョン単位に分けて管理する
- Operationsのcreate_tableでテーブル作成を表現する
- 適用済みリビジョンをschema_versionに記録する
- 未適用のリビジョンだけをループで順に反映する
- 2番目のリビジョンで履歴テーブルを後から追加する
Python・SQLAlchemyで開発する場合の環境構築
この記事のセットアップ手順と掲載コードは、Windows 11 Pro、PowerShell 5.1、Python 3.13.3で動作確認しています。仮想環境を有効化せず、その中のPythonを直接指定するため、以下のコマンドはPowerShellとコマンドプロンプト（cmd）の両方で使えます。
python -m venv .venv
.\.venv\Scripts\python.exe -m pip install --upgrade pip
.\.venv\Scripts\python.exe -m pip install SQLAlchemy Alembic
.\.venv\Scripts\python.exe -c "import sqlite3; print('SQLite version:', sqlite3.sqlite_version)"macOS・Linuxでは仮想環境内のPythonパスが異なります。今回の動作確認環境とは異なるため、以下は環境差分を補う参考手順です。
python3 -m venv .venv
./.venv/bin/python -m pip install --upgrade pip
./.venv/bin/python -m pip install SQLAlchemy Alembic
./.venv/bin/python -c "import sqlite3; print('SQLite version:', sqlite3.sqlite_version)"- SQLiteはPython標準のsqlite3モジュールとして利用し、追加インストールは不要です。
- 最初にmigrateを実行し、商品テーブルと更新履歴テーブルを作成します。
- --dbで指定した書き込み可能な場所へSQLiteファイルが作られます。
在庫管理CLIの要件定義
目的は、SQLAlchemyでモデルを定義しAlembicでマイグレーションを適用してSQLiteに在庫数と更新履歴を保存し、商品登録から在庫更新までをCLIで確認できるようにすることです。
対象者として、Pythonの基本文法を理解しSQLAlchemyやAlembicを使ったデータベース操作とマイグレーションの実装を学びたい人を想定しています。
完成物は、SQLAlchemyとAlembicとSQLiteで在庫と更新履歴を管理し、migrate・add・list・update・historyの5コマンドで操作できる在庫管理CLIです。
実装へ入る前に、機能・品質・受け入れ条件を分けて確認します。
機能要件
- 未適用のリビジョンだけを順に反映するmigrateコマンド
- 商品コードと商品名と在庫数を登録するaddコマンド
- 登録済みの在庫を商品コード順に表示するlistコマンド
- 商品コードを指定して在庫数を更新するupdateコマンド
- 商品コードを指定して更新履歴を表示するhistoryコマンド
- 在庫更新のたびに変更前後の数量と理由を残す履歴記録
- 既に登録済みの商品コードを検知して知らせる重複チェック
非機能要件
- 在庫数が負の値なら例外を送出して登録と更新を防ぐバリデーション
- 商品コードのskuにunique制約を付けて重複登録を防ぐテーブル定義
- 適用済みリビジョンをschema_versionで管理する冪等なマイグレーション
- SQLiteのデータベースパスを--dbオプションで指定できる構成
- DeclarativeBaseとMapped型でモデルを定義する宣言的マッピング
- Sessionをwith文で開いて処理後にcommitする接続管理
実装方針
今回はSQLAlchemyとAlembicの基本動作を追いやすくするため、在庫管理CLI本体を1つのPythonファイルへまとめます。
入力、判定、結果表示の役割を分け、実行結果を確認しながら機能を積み上げます。
在庫管理CLIを安全に組み立てるための実装方針は次のとおりです。
- 在庫数が負の値なら例外を送出して登録と更新を防ぐバリデーション
- 商品コードのskuにunique制約を付けて重複登録を防ぐテーブル定義
- 適用済みリビジョンをschema_versionで管理する冪等なマイグレーション
- SQLiteのデータベースパスを--dbオプションで指定できる構成
- DeclarativeBaseとMapped型でモデルを定義する宣言的マッピング
- Sessionをwith文で開いて処理後にcommitする接続管理
完成と判断する条件
- migrateで0001と0002の2件のリビジョンが適用されること
- addで登録した在庫がlistに商品コード順で表示されること
- updateで在庫数を変えると変更前後の数量が履歴に残ること
- historyで指定した商品コードの更新履歴が確認できること
- 同じ商品コードを再登録すると登録済みと通知されること
- 在庫数に負の値を渡すと例外で処理が止まること
SQLAlchemyとAlembicで在庫管理CLIを作る際の重要ポイント
この在庫管理CLIの中心は、在庫数を書き換えるだけでなく、変更前後の数量と理由を履歴として残すところにあります。updateコマンドでは、更新前の在庫数を控えてから新しい値を設定し、その差分をStockChangeとして同じトランザクションに保存する流れです。historyコマンドでその記録をたどれば、いつどれだけ在庫が動いたかを後から確認できます。
在庫更新と履歴記録でコマンドが行う処理を挙げます。
- 更新前の在庫数をbefore_quantityとして控える
- 新しい在庫数をafter_quantityとして記録する
- 更新理由をreasonへ保存して後から追える形にする
- 更新と履歴追加を同じcommitでまとめて確定する
- 存在しない商品コードには見つからない旨を返す
updateコマンドが在庫数と履歴を書き換える手順
updateコマンドは、まず商品コードで対象のInventoryItemを探します。見つかった場合は現在の在庫数をbeforeへ退避し、item.quantityに新しい値を代入する流れです。続けてbeforeと新しい値、更新理由を持つStockChangeを追加し、商品の更新と履歴の追加を一度のcommitで確定します。
これにより、在庫の現在値と変更の記録が食い違わないようにそろえています。
updateコマンドが順に行う処理を示します。
- 商品コードでInventoryItemを1件検索する
- 現在の在庫数をbeforeへ控える
- item.quantityへ更新後の数量を代入する
- StockChangeを追加して差分と理由を残す
在庫データと更新履歴を一覧で確認する方法
listコマンドは在庫を商品コード順に並べ、件数と各商品の在庫数を一覧にします。historyコマンドは指定した商品コードの更新履歴をidの昇順で並べ、変更前後の数量と理由を1行ずつ示す形です。どちらもデータが無い場合には専用のメッセージを返すため、空の状態でも結果を読み取れます。
一覧系コマンドが返す出力の違いを示します。
- listは在庫一覧の件数と商品ごとの在庫数を表示
- historyは更新履歴の件数と変更前後の数量を表示
- 在庫が無いときは在庫データが無い旨のメッセージを返す
- 履歴が無いときは更新履歴が無い旨のメッセージを返す
Pythonで在庫管理CLIの完成コード
コードは大きく3つのパートに分かれます。モデル定義、マイグレーション定義、そしてargparseによるサブコマンドの処理です。
モデルはDeclarativeBaseを継承したBaseを土台に、InventoryItemとStockChangeを宣言します。列の型と制約はmapped_columnで指定する形です。
マイグレーションはOperationsのcreate_tableでテーブルを組み立てる関数を用意し、MIGRATIONSという一覧に順番で並べておきます。
この一覧はrun_migrationsが上から適用するため、リビジョンの追加はリストへの1行追加だけで完了する仕組みです。
このセクションの用語
- DeclarativeBase
- SQLAlchemyのモデルが継承する基底クラスです。これを継承したクラスがテーブルに対応します。
- mapped_column
- 列の型や制約を指定してクラス属性に結び付ける関数です。主キーや一意制約もここで書きます。
- ForeignKey
- 別テーブルの列を参照する外部キーを表す指定です。履歴と商品を結び付けます。
- Operations
- Alembicがテーブル作成や列追加などの操作を行うためのオブジェクトです。create_tableなどを呼び出せます。
- リビジョン
- マイグレーションの1段階を表す単位です。0001や0002のように順序を付けて管理します。
"""SQLAlchemy・Alembic・SQLiteで作る在庫管理CLI。
実装する機能:
- AlembicのDBマイグレーションを適用する
- 在庫データを登録・一覧表示する
- 商品コードを指定して在庫数を更新する
- 更新履歴を確認する
"""
import argparse
from alembic.migration import MigrationContext
from alembic.operations import Operations
from sqlalchemy import (
Column,
ForeignKey,
Integer,
String,
create_engine,
select,
)
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column
class Base(DeclarativeBase):
"""SQLAlchemyの宣言的マッピングで使う基底クラス。"""
class InventoryItem(Base):
"""商品コード、商品名、現在庫数を保持する。"""
__tablename__ = "inventory_items"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
sku: Mapped[str] = mapped_column(String(40), unique=True, nullable=False)
name: Mapped[str] = mapped_column(String(120), nullable=False)
quantity: Mapped[int] = mapped_column(Integer, nullable=False)
class StockChange(Base):
"""在庫更新前後の数量と理由を履歴として保持する。"""
__tablename__ = "stock_changes"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
item_id: Mapped[int] = mapped_column(
Integer, ForeignKey("inventory_items.id"), nullable=False
)
before_quantity: Mapped[int] = mapped_column(Integer, nullable=False)
after_quantity: Mapped[int] = mapped_column(Integer, nullable=False)
reason: Mapped[str] = mapped_column(String(200), nullable=False)
def _upgrade_0001(op: Operations) -> None:
"""初回リビジョンで在庫商品テーブルを作る。"""
op.create_table(
"inventory_items",
Column("id", Integer, primary_key=True),
Column("sku", String(40), unique=True, nullable=False),
Column("name", String(120), nullable=False),
Column("quantity", Integer, nullable=False),
)
def _upgrade_0002(op: Operations) -> None:
"""2番目のリビジョンで在庫更新履歴テーブルを追加する。"""
op.create_table(
"stock_changes",
Column("id", Integer, primary_key=True),
Column(
"item_id",
Integer,
ForeignKey("inventory_items.id"),
nullable=False,
),
Column("before_quantity", Integer, nullable=False),
Column("after_quantity", Integer, nullable=False),
Column("reason", String(200), nullable=False),
)
MIGRATIONS = [
("0001_create_inventory_items", _upgrade_0001),
("0002_add_stock_changes", _upgrade_0002),
]
def run_migrations(engine) -> list[str]:
"""AlembicのDBマイグレーションを適用する。"""
applied: list[str] = []
with engine.begin() as connection:
connection.exec_driver_sql(
"CREATE TABLE IF NOT EXISTS schema_version "
"(revision TEXT PRIMARY KEY)"
)
completed = {
row[0]
for row in connection.exec_driver_sql(
"SELECT revision FROM schema_version"
)
}
context = MigrationContext.configure(connection)
operations = Operations(context)
for revision, upgrade in MIGRATIONS:
if revision in completed:
continue
upgrade(operations)
connection.exec_driver_sql(
"INSERT INTO schema_version (revision) VALUES (?)",
(revision,),
)
applied.append(revision)
return applied
def cmd_migrate(engine, _args) -> str:
"""未適用リビジョンだけを順番に反映する。"""
applied = run_migrations(engine)
if not applied:
return "マイグレーションは最新です"
return "適用したマイグレーション: " + ", ".join(applied)
def cmd_add(engine, args) -> str:
"""在庫データを登録・一覧表示する機能の登録側を担当する。"""
if args.quantity < 0:
raise ValueError("在庫数は0以上で指定してください")
with Session(engine) as session:
existing = session.execute(
select(InventoryItem).where(InventoryItem.sku == args.sku)
).scalar_one_or_none()
if existing is not None:
return f"商品コードは登録済みです: {args.sku}"
item = InventoryItem(
sku=args.sku,
name=args.name,
quantity=args.quantity,
)
session.add(item)
session.commit()
return f"在庫を登録しました: {item.sku} {item.name} 在庫数={item.quantity}"
def cmd_list(engine, _args) -> str:
"""在庫データを登録・一覧表示する機能の一覧側を担当する。"""
with Session(engine) as session:
items = session.execute(
select(InventoryItem).order_by(InventoryItem.sku)
).scalars().all()
if not items:
return "在庫データはありません"
lines = [f"在庫一覧: {len(items)}件"]
lines.extend(
f"{item.sku} | {item.name} | 在庫数={item.quantity}"
for item in items
)
return "\n".join(lines)
def cmd_update(engine, args) -> str:
"""商品コードを指定して在庫数を更新する。"""
if args.quantity < 0:
raise ValueError("在庫数は0以上で指定してください")
with Session(engine) as session:
item = session.execute(
select(InventoryItem).where(InventoryItem.sku == args.sku)
).scalar_one_or_none()
if item is None:
return f"商品コードが見つかりません: {args.sku}"
before = item.quantity
item.quantity = args.quantity
session.add(
StockChange(
item_id=item.id,
before_quantity=before,
after_quantity=args.quantity,
reason=args.reason,
)
)
session.commit()
return (
f"在庫数を更新しました: {item.sku} "
f"{before} -> {item.quantity} 理由={args.reason}"
)
def cmd_history(engine, args) -> str:
"""更新履歴を確認する。"""
with Session(engine) as session:
item = session.execute(
select(InventoryItem).where(InventoryItem.sku == args.sku)
).scalar_one_or_none()
if item is None:
return f"商品コードが見つかりません: {args.sku}"
changes = session.execute(
select(StockChange)
.where(StockChange.item_id == item.id)
.order_by(StockChange.id)
).scalars().all()
if not changes:
return f"更新履歴はありません: {item.sku}"
lines = [f"更新履歴: {item.sku} {len(changes)}件"]
lines.extend(
f"#{change.id} {change.before_quantity} -> "
f"{change.after_quantity} 理由={change.reason}"
for change in changes
)
return "\n".join(lines)
def build_parser() -> argparse.ArgumentParser:
"""在庫管理CLIの引数とサブコマンドを定義する。"""
parser = argparse.ArgumentParser(
description="SQLAlchemy・Alembic・SQLiteで作る在庫管理CLI"
)
parser.add_argument(
"--db", default="inventory.db", help="SQLiteデータベースのパス"
)
commands = parser.add_subparsers(dest="command", required=True)
commands.add_parser("migrate", help="DBマイグレーションを適用する")
add_parser = commands.add_parser("add", help="在庫データを登録する")
add_parser.add_argument("--sku", required=True, help="商品コード")
add_parser.add_argument("--name", required=True, help="商品名")
add_parser.add_argument("--quantity", required=True, type=int, help="在庫数")
commands.add_parser("list", help="在庫一覧を表示する")
update_parser = commands.add_parser("update", help="在庫数を更新する")
update_parser.add_argument("sku", help="商品コード")
update_parser.add_argument("--quantity", required=True, type=int, help="更新後の在庫数")
update_parser.add_argument("--reason", required=True, help="更新理由")
history_parser = commands.add_parser("history", help="更新履歴を表示する")
history_parser.add_argument("sku", help="商品コード")
return parser
def main(argv=None) -> str:
"""選択されたサブコマンドを実行して結果を返す。"""
args = build_parser().parse_args(argv)
engine = create_engine(f"sqlite:///{args.db}")
handlers = {
"migrate": cmd_migrate,
"add": cmd_add,
"list": cmd_list,
"update": cmd_update,
"history": cmd_history,
}
return handlers[args.command](engine, args)
if __name__ == "__main__":
print(main())
コード全文は上の折り畳みに入れてあるので、全部を上から読む必要はありません。ここでは特に重要な部分だけを抜き出して、何をしているのか順番に見ていきます。
DeclarativeBaseで基底クラスを定義
class Base(DeclarativeBase):
"""SQLAlchemyの宣言的マッピングで使う基底クラス。"""DeclarativeBaseを継承したBaseが、すべてのモデルの共通の土台になります。以降のクラスはこれを継承するだけで、テーブルとして扱えるようになります。
InventoryItemで商品テーブルをマッピング
class InventoryItem(Base):
__tablename__ = "inventory_items"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
sku: Mapped[str] = mapped_column(String(40), unique=True, nullable=False)
name: Mapped[str] = mapped_column(String(120), nullable=False)
quantity: Mapped[int] = mapped_column(Integer, nullable=False)__tablename__でテーブル名を決め、各列をmapped_columnで定義します。skuにはunique=Trueを付け、商品コードが重複しないようにしています。
StockChangeで変更前後と理由を記録
item_id: Mapped[int] = mapped_column(
Integer, ForeignKey("inventory_items.id"), nullable=False
)
before_quantity: Mapped[int] = mapped_column(Integer, nullable=False)
after_quantity: Mapped[int] = mapped_column(Integer, nullable=False)
reason: Mapped[str] = mapped_column(String(200), nullable=False)履歴テーブルはitem_idにForeignKeyを付け、どの商品の変更かを結び付けます。変更前後の数量とreasonを持たせ、経緯をあとから追えるようにしています。
_upgrade_0001でinventory_itemsを作成
def _upgrade_0001(op: Operations) -> None:
"""初回リビジョンで在庫商品テーブルを作る。"""
op.create_table(
"inventory_items",
Column("id", Integer, primary_key=True),
Column("sku", String(40), unique=True, nullable=False),
Column("name", String(120), nullable=False),
Column("quantity", Integer, nullable=False),
)_upgrade_0001はOperationsのcreate_tableで商品テーブルを組み立てます。列の型や制約はモデル定義とそろえ、最初のリビジョンとして実行します。
MIGRATIONSでリビジョンの順序を管理
MIGRATIONS = [
("0001_create_inventory_items", _upgrade_0001),
("0002_add_stock_changes", _upgrade_0002),
]MIGRATIONSはリビジョン名と適用関数の組を順番に並べたリストです。新しい変更はこの末尾に足すだけで、適用の順序を保てます。
run_migrationsでマイグレーションを適用
def run_migrations(engine) -> list[str]:
"""AlembicのDBマイグレーションを適用する。"""
applied: list[str] = []
with engine.begin() as connection:run_migrationsはengine.begin()でトランザクションを開き、リストの関数を順に適用します。適用したリビジョン名をリストにまとめて返す作りです。
参考：
©SQLAlchemy 2.0 DocumentationThe Declarative Mapping is the typical way that mappings are constructed in modern SQLAlchemy.
在庫管理CLIの動作確認
作ったCLIを/tmp/inventory_demo.dbに対して動かしました。migrateからhistoryまで5回のコマンドを順に実行し、すべて終了コード0で完了しています。
最初のmigrateではテーブルが作られ、続くaddでSKU-001の有機コーヒーを在庫25で登録しました。
listで登録内容を確認し、update SKU-001 --quantity 18 --reason 出荷で在庫を18へ更新しています。
最後のhistory SKU-001では、在庫が25から18へ変わった記録と、理由『出荷』が表示されました。
このセクションの用語
- 終了コード
- コマンドが正常終了か異常終了かを表す数値です。0は正常終了を意味します。
- SKU
- 商品を一意に識別するための管理コードです。今回はsku列に保存します。
実行した5つのコマンドと、確認できた結果です。
-
migrate: テーブル作成が終了コード0で完了 -
add --sku SKU-001 --name 有機コーヒー --quantity 25: 商品を登録 -
list: 登録した在庫を一覧表示 -
update SKU-001 --quantity 18 --reason 出荷: --reasonを省略すると必須引数エラーになるため、更新時は理由の入力が必須 -
history SKU-001: 25から18への変更履歴を確認
PythonのSQLAlchemyとAlembicのエラー対処
在庫管理CLIを動かすとき、初心者がつまずきやすいのはテーブル未作成と制約違反です。多くはエラーメッセージに原因が表れます。
特にmigrateを忘れると『テーブルが無い』という例外が出ます。落ち着いてメッセージを読み、原因の列や制約を確認しましょう。
このセクションの用語
- UNIQUE制約
- 列の値が重複しないことを保証するデータベースの制約です。sku列に付けています。
- IntegrityError
- NOT NULLや一意制約に違反したときにSQLAlchemyが出す例外です。整合性が崩れる操作を防ぎます。
- OperationalError
- テーブルが無い、DBファイルを開けないなど操作自体が失敗したときの例外です。
| エラー例 | 原因 | 対処 |
|---|---|---|
| no such table: inventory_items | migrateを実行していない | 先にmigrateサブコマンドでテーブルを作成する |
| UNIQUE constraint failed: inventory_items.sku | 同じskuを重複してaddした | 既存商品はupdateで更新し、addは新規のみに使う |
| ModuleNotFoundError: No module named 'alembic' | alembicが未インストール | pip installでalembicを導入する |
| IntegrityError: NOT NULL constraint failed | quantityなど必須列が未指定 | 必須の引数をすべて渡して実行する |
| OperationalError: unable to open database file | --dbのパス先ディレクトリが存在しない | 存在するディレクトリのパスを指定する |
在庫管理CLIで注意したい点
在庫管理CLIで最も大切なのは、更新と履歴の一貫性です。在庫数を変えた事実と、その前後の数量・理由を必ずセットで残します。
そのためには、在庫の更新とStockChangeへの追加を同じトランザクションで確定させる運用です。片方だけ保存される事故を防げます。
マイグレーションも順序が命です。0001を飛ばして0002を当てると参照先の商品テーブルが無く失敗します。ポイントとしては、マイグレーションの順序と履歴の一貫性という2つの押さえどころをまとめました。
リビジョン順序：0001→0002の順で適用
在庫と履歴：同一トランザクションで確定
SKUの一意性：重複登録を事前に確認
在庫管理CLIが役立つ場面とAlembic活用のアイデア
この在庫管理CLIは、そのまま小さな在庫台帳として使えます。履歴が残るので、後から増減の経緯を振り返れるのが強みです。
学習用途にも向いています。Alembicのリビジョンを足していくと、スキーマ変更の管理をわかりやすく体験できます。
このセクションの用語
- 監査ログ
- 誰がいつ何を変更したかを後から追える記録です。StockChangeの履歴がこれに当たります。
- スキーマ
- テーブルや列の構造を指す言葉です。Alembicはこのスキーマの変更を管理します。
| 使える場面 | 具体的な使い方 |
|---|---|
| 小規模店舗の在庫台帳 | 商品をaddで登録し、入出荷のたびにupdateで在庫数と理由を記録する |
| Alembicの学習 | リビジョンを0003以降に増やし、列追加や制約変更を段階的に試す |
| 監査ログ設計の練習 | StockChangeの前後数量と理由を使い、変更を追跡できる設計を学ぶ |
| 個人の持ち物管理 | skuを自分のラベルにして、手持ちの在庫数を更新しながら管理する |
PythonとSQLiteで作った在庫管理CLIの振り返り
SQLAlchemyでモデルを宣言し、Alembicでマイグレーションを段階適用し、SQLiteに保存する在庫管理CLIを作りました。
実際に5回のコマンドを実行し、登録・一覧・更新・履歴確認までがすべて終了コード0で動きました。
更新と履歴を同じトランザクションで残す設計は、実務の在庫管理でも応用できる考え方です。まずは手元のDBファイルで、サブコマンドを1つずつ試してみてください。
さらに詳しく学べるPython研修の詳細はこちら参考にした一次情報
- ^ SQLAlchemy ORM Quick Start. https://docs.sqlalchemy.org/en/20/orm/quickstart.html, (参照26-08-05).
- ^ Alembic Operation Reference. https://alembic.sqlalchemy.org/en/latest/ops.html, (参照26-08-05).
- ^ Alembic Runtime Objects. https://alembic.sqlalchemy.org/en/latest/api/runtime.html, (参照26-08-05).
- ^ Python argparse. https://docs.python.org/ja/3/library/argparse.html, (参照26-08-05).
※内容は執筆時点のものです。ライブラリやサイトの仕様は変わる可能性があるため、公式ドキュメントもあわせてご確認ください。
ITやプログラミングに関するコラム
【Python】FastAPIで料金プラン見積もりシミュレーターを作ってみた
【Python】pandasとmatplotlibで在庫データのABC分析と構成比を可視化してみた
【Python】Flaskで社内FAQをカテゴリ検索できるWebアプリを作ってみた
【Python】argparseでJSON整形・構文検証・キー検索CLIを試してみた
【Python】NumPyとmatplotlibでモンテカルロ法による円周率推定と収束過程の可視化を試してみた
【Python】Playwrightでスクレイピングを試してみた
【CSS】notで複数の件を除外する方法
【Git】remote設定を変更する方法
【VBA】コメントアウトを設定する方法
x86とx64の違いを分かりやすく解説
ITやプログラミングに関するニュース
VercelがAI GatewayにSeedream 5.0 Proを追加、AI SDKのモデル指定で画像生成と編集が可能に
AWSがAmazon LocationのPlaces APIを強化、住所表記の指定と移動手段別の検索が可能に
VercelがトレースにTree・Waterfallビューを追加、ログ画面で処理の階層と所要時間を確認可能に
Googleがエージェント評価の再考を提唱、難易度を情報量で測るDiscovery Benchを解説
Google CloudがCloud Runサンドボックスを公開プレビューで提供、サービスヘルスは一般提供に
Google Cloud EMEAが英国金融の重要第三者に指定、イングランド銀行・PRA・FCAの直接監督下に
AWS DMS Schema ConversionがSQL Serverのオフライン変換に対応、ソースDBへ接続せずスキーマを変換可能に
EC2 G7インスタンスが米国東部（バージニア北部）で利用可能に、G6比でAI推論性能が最大4.6倍
SageMaker HyperPodが継続プロビジョニングでのAMIベース構成に対応、S3のスクリプト管理なしでSlurmクラスターを作成可能に
AWSがEMR on EKSでSparkトラブルシューティングエージェントに対応、失敗ジョブの原因分析を自然言語で依頼可能に
