SQLAlchemyで`ON CONFLICT DO UPDATE`を使用する際のサンプルコード
SQLAlchemyでon_conflict_do_updateを使う際のエラーと解決策
sqlalchemyでON CONFLICT DO UPDATEを使用する際に、ProgrammingErrorが発生するケースがあります。このエラーは、主に以下の2つの原因で発生します。
ON CONFLICT句の記述に誤りがあるUPDATEされるカラムにデフォルト値や生成関数が設定されている
原因と解決策
ON CONFLICT句は、競合が発生した際に実行する処理を指定します。記述に誤りがあると、ProgrammingErrorが発生します。
例：
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class MyTable(Base):
__tablename__ = "my_table"
id = Column(Integer, primary_key=True)
name = Column(String, unique=True)
engine = create_engine("sqlite:///:memory:", echo=True)
Base.metadata.create_all(engine)
# 誤った記述
stmt = insert(MyTable).values(name="foo")
stmt = stmt.on_conflict_do_update(constraint="name", set_=dict(name="bar"))
# エラーが発生
with Session(engine) as session:
session.execute(stmt)
上記の例では、constraint引数にnameという名前の制約を指定しています。しかし、my_tableテーブルにはnameという名前の制約は存在しません。そのため、ProgrammingErrorが発生します。
解決策：
ON CONFLICT句を正しく記述する必要があります。具体的には、以下の点を確認してください。
constraint引数には、存在する制約の名前を指定する必要があります。index_elements引数には、競合が発生するカラムを指定する必要があります。
# 正しい記述
stmt = insert(MyTable).values(name="foo")
stmt = stmt.on_conflict_do_update(constraint="unique_name", set_=dict(name="bar"))
# エラーは発生しない
with Session(engine) as session:
session.execute(stmt)
上記の例では、constraint引数にunique_nameという名前の制約を指定しています。my_tableテーブルにはunique_nameという名前の制約が存在するため、エラーが発生しません。
UPDATEされるカラムにデフォルト値や生成関数が設定されている場合、ON CONFLICT DO UPDATEを使用しても、それらの値は反映されません。
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql import func
Base = declarative_base()
class MyTable(Base):
__tablename__ = "my_table"
id = Column(Integer, primary_key=True)
name = Column(String, unique=True, default=func.lower(func.uuid4()))
engine = create_engine("sqlite:///:memory:", echo=True)
Base.metadata.create_all(engine)
# デフォルト値が設定されているカラムをUPDATE
stmt = insert(MyTable).values(name="foo")
stmt = stmt.on_conflict_do_update(constraint="unique_name", set_=dict(name="bar"))
# デフォルト値は反映されない
with Session(engine) as session:
session.execute(stmt)
row = session.query(MyTable).first()
print(row.name) # "foo"
上記の例では、nameカラムにデフォルト値としてfunc.lower(func.uuid4())という生成関数が設定されています。しかし、ON CONFLICT DO UPDATEを使用しても、nameカラムの値はfooのままです。
UPDATEされるカラムのデフォルト値や生成関数を、on_conflict_do_updateのset_引数で明示的に指定する必要があります。
# デフォルト値を明示的に指定
stmt = insert(MyTable).values(name="foo")
stmt = stmt
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class MyTable(Base):
__tablename__ = "my_table"
id = Column(Integer, primary_key=True)
name = Column(String, unique=True)
engine = create_engine("sqlite:///:memory:", echo=True)
Base.metadata.create_all(engine)
# INSERT
stmt = insert(MyTable).values(name="foo")
stmt = stmt.on_conflict_do_update(constraint="unique_name", set_=dict(name="bar"))
# エラーは発生しない
with Session(engine) as session:
session.execute(stmt)
# UPDATE
stmt = update(MyTable).where(MyTable.name == "foo")
stmt = stmt.values(name="baz")
stmt = stmt.on_conflict_do_update(constraint="unique_name", set_=dict(name="qux"))
# エラーは発生しない
with Session(engine) as session:
session.execute(stmt)
# SELECT
with Session(engine) as session:
row = session.query(MyTable).first()
print(row.name) # "qux"
解説
上記のコードでは、以下の処理を行っています。
MyTableテーブルというテーブルを作成します。nameカラムにUNIQUE制約を設定します。nameカラムが競合した場合、nameカラムの値をbarに更新するようにON CONFLICT DO UPDATEを使用します。nameカラムの値をfooからbazに変更します。nameカラムの値がquxになっていることを確認します。
ON CONFLICT DO UPDATEの代替方法
ON CONFLICT DO UPDATEの代わりに、以下の方法を使用することができます。
方法1：INSERT ... ON DUPLICATE KEY UPDATEを使用する
MySQLやMariaDBなどのデータベースでは、INSERT ... ON DUPLICATE KEY UPDATEという構文を使用することができます。この構文は、レコードの挿入時に競合が発生した場合、指定されたカラムの値を更新します。
INSERT INTO my_table (name, age) VALUES ("foo", 10)
ON DUPLICATE KEY UPDATE age = age + 1;
上記の例では、my_tableテーブルにnameカラムとageカラムを持つレコードを挿入します。nameカラムが競合した場合、ageカラムの値を1増やします。
方法2：UPSERTを使用する
PostgreSQLなどのデータベースでは、UPSERTという構文を使用することができます。この構文は、レコードの挿入または更新を行います。
INSERT INTO my_table (name, age) VALUES ("foo", 10)
ON CONFLICT (name) DO UPDATE SET age = age + 1;
方法3：アプリケーション側で処理を行う
上記の2つの方法が使用できない場合、アプリケーション側で処理を行うことができます。
def upsert(name, age):
try:
# レコードを挿入
session.add(MyTable(name=name, age=age))
session.flush()
except sqlalchemy.exc.IntegrityError:
# 競合が発生した場合、レコードを更新
session.query(MyTable).filter(MyTable.name == name).update(dict(age=age + 1))
session.flush()
# 例
upsert("foo", 10)
sqlalchemy
