HOWTO · Flask
Flask SQLAlchemy でレコードを削除する
Flask-SQLAlchemy でレコードを削除する方法を学びます。db.session.get() でモデルインスタンスを主キーから読み込み、db.session.delete() で削除対象にし、db.session.commit() で削除を確定します。
このページの内容
Flask-SQLAlchemy でレコードを削除するには、モデルインスタンスを読み込み、それを db.session.delete() に渡し、db.session.commit() でトランザクションを確定するという3つの手順を実行します。基本的なパターンは次のとおりです。
このセクションのコードと表示されるラベルや出力は英語のままです。
def delete_customer(customer_id):
customer = db.session.get(Customer, customer_id)
if customer is not None:
db.session.delete(customer)
db.session.commit()
db.session.get() は主キーで行を取得し、一致する行がない場合は None を返します。db.session.delete() は読み込んだインスタンスをセッション内で削除対象としてマークし、db.session.commit() が実際の DELETE 文を発行します。コミットが成功するまで、行はデータベースに残ったままです。
以下のコードリストでは、各コードブロックが自己完結するように、Flask-SQLAlchemy の呼び出しを関数内にまとめています。実際のプロジェクトでは、同じステートメントをモジュールレベルやビューの中で実行し、db と Customer はファイルの前の方でインポートまたは定義します。
Flask-SQLAlchemy をインストールしてアプリを設定する
パッケージは pip install flask-sqlalchemy でインストールします(詳細は Flask-SQLAlchemy の PyPI ページ を参照してください)。その後、SQLAlchemy オブジェクトを作成する前に SQLALCHEMY_DATABASE_URI を設定して、Flask アプリが使用するデータベースを指定します。アプリケーションファクトリを使うと設定を一か所にまとめられます。プレーンなモジュールでは、代わりに2つのインポートをファイルの先頭に配置します。
このセクションのコードと表示されるラベルや出力は英語のままです。
def create_app():
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:"
db = SQLAlchemy(app)
return app, db
この設定値には任意の SQLAlchemy データベース URL を指定できます。ここで示すインメモリ SQLite URL は例を自己完結させるためのものなので、実際のアプリケーションでは使用するデータベースの URL に置き換えてください。
モデルを定義してレコードを作成する
削除を行うには対象の行が必要なので、まず小さな Customer モデルから始めます。db.Column で定義した各属性は customer テーブルのカラムになります。実際のプロジェクトでは、このクラスはモデルファイルのモジュールレベルに配置します。ここでは、コードリストを単体で成立させるためにファクトリ関数を使っています。
このセクションのコードと表示されるラベルや出力は英語のままです。
def define_models(db):
class Customer(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(50), nullable=False)
address = db.Column(db.String(200))
email = db.Column(db.String(50))
return Customer
def seed_first_customer(app, db):
Customer = define_models(db)
with app.app_context():
db.create_all()
db.session.add(
Customer(
name="Ada Lovelace",
address="12 Analytical Engine Way",
email="ada@example.com",
)
)
db.session.commit()
db.create_all() は Flask のアプリケーションコンテキスト内でテーブルを作成し、db.session.add() と db.session.commit() を組み合わせてレコードを挿入します。id は主キーなので、行の挿入時にデータベースが値を自動的に割り当て、後でレコードを検索する際に使える安定した値になります。
レコードを読み込んで削除する
db.session.get(Customer, customer_id) は、その主キーに対応するインスタンスを返し、行が存在しない場合は None を返します。None を db.session.delete() に渡さないよう、先にレコードが見つからないケースを確認します。
このセクションのコードと表示されるラベルや出力は英語のままです。
def delete_customer(customer_id):
customer = db.session.get(Customer, customer_id)
if customer is None:
return "No customer with that id."
db.session.delete(customer)
db.session.commit()
return "Customer deleted."
db.session.delete() を呼び出しても、行がすぐに削除されるわけではありません。この呼び出しはセッション内でインスタンスを削除予定としてマークし、変更は次の db.session.commit() まで保留されます。
削除をコミットして失敗に対処する
db.session.commit() は、スケジュールされた DELETE を含むすべての保留中の変更を単一トランザクション内でフラッシュし、永続化します。コミットを省略すると、削除はデータベースに届きません。db.session.commit() が実行されるまで、変更はトランザクション内に保留されたままです。次の例を実行すると、プレーンな Python で削除が永続化される様子を確認できます。
このセクションのコードと表示されるラベルや出力は英語のままです。
import sqlite3
conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE customer (id INTEGER PRIMARY KEY, name TEXT)")
conn.execute("INSERT INTO customer VALUES (1, 'Ada')")
cursor = conn.cursor()
cursor.execute("DELETE FROM customer WHERE id = 1")
conn.commit()
print(cursor.execute("SELECT COUNT(*) FROM customer").fetchone()[0])
0
コミットは、制約違反や接続切断などが原因で失敗することもあります。失敗するとトランザクションは開いたままになり、db.session.rollback() を呼び出して失敗したトランザクションを破棄するまで、セッションはそれ以降のクエリを実行できません。
def commit_changes():
try:
db.session.commit()
except Exception:
db.session.rollback()
raise
Flask-SQLAlchemy は db.session をアプリケーションコンテキストにスコープするため、失敗したセッションは次のクエリを実行する前に必ずロールバック(または削除)してください。
完全な例と結果の確認
これらの要素を組み合わせた次のコードリストは、アプリを設定し、Customer モデルを定義し、1件のレコードを挿入して削除した後、再度検索します。このコードを実行するには Flask-SQLAlchemy がインストールされている必要があります。パッケージを導入済みのプロジェクトで main() を呼び出してください。
このセクションのコードと表示されるラベルや出力は英語のままです。
def main():
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:"
db = SQLAlchemy(app)
class Customer(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(50), nullable=False)
address = db.Column(db.String(200))
email = db.Column(db.String(50))
def insert_customer(name, address, email):
customer = Customer(name=name, address=address, email=email)
db.session.add(customer)
db.session.commit()
return customer.id
def delete_customer(customer_id):
customer = db.session.get(Customer, customer_id)
if customer is None:
return "No customer with that id."
db.session.delete(customer)
db.session.commit()
return "Customer deleted."
with app.app_context():
db.create_all()
new_id = insert_customer(
"Ada Lovelace", "12 Analytical Engine Way", "ada@example.com"
)
print(delete_customer(new_id))
print(db.session.get(Customer, new_id))
main() を実行すると、スクリプトは Customer deleted. に続いて None を出力します。このメッセージは、コード内の英語文字列がそのまま表示されるものです。後続の db.session.get() 呼び出しが、削除された主キーに対応する行を見つけられないためです。自分のアプリで削除を確認するには、同じ方法でレコードを再度検索するか、フィルター付きのカウントクエリを実行します。結果が None(またはカウントが0)なら、行が削除されたことが確認できます。
