GENERATED ALWAYS AS IDENTITY で採番する列を持つテーブルを
TRUNCATE TABLE テーブル名
を実行したとき採番はまだ続いてしまうので、
初期値から振るようにする場合は、
TRUNCATE TABLE テーブル名 restart identity
で行う。
GENERATED ALWAYS AS IDENTITY は、
自動でシーケンスが作成されているので、
テーブル生成時と TRUNCATE TABLE テーブル名 restart identity 実行後は、
SELECT last_value FROM 自動生成されたシーケンス
は、１が返ってくる。
シーケンスの生成して nextval を実行していない状態も
SELECT last_value FROM シーケンス
は、１が返ってくる。
ストアドファンクションでシーケンスを呼び出して加工して返す。
任意の文字２個＋シーケンス０埋め４桁を作る目的です。
CREATE OR REPLACE FUNCTION makeuid(area CHAR, gp CHAR) RETURNS VARCHAR AS $$ DECLARE id BIGINT; BEGIN SELECT nextval('user_id_seq') INTO id; RETURN area || gp || LPAD(CAST(id AS VARCHAR), 4, '0'); END; $$ LANGUAGE plpgsql ;
作成で以前はこんなことなかったと思ったが、いつからだろうか
LANGUAGE plpgsql を付けないと作れなくなっている。
ついでに、現在のシーケンスを返す同様の関数を、
nextval ⇒ currval に変えたものを作ってみます。
でも、これは注意が必要で、nextval を動かさないで実行すると、
SQLエラー [55000]: ERROR: 本セッションでシーケンス～のcurrvalは
まだ定義されていませんになります。
これは PostgreSQLの仕様で、セッションで nextval を実行していないと
55000 のエラーを起こすようです。
www.postgresql.org
nextval を実行していないセッションでは、55000エラーを catch して
nextval を実行させて、setval で戻してから currvalを実行する手もありますが、
それは、ナンセンスです。↓↓↓↓↓↓↓
CREATE OR REPLACE FUNCTION curuid(area CHAR, gp CHAR) RETURNS VARCHAR AS $$ DECLARE id BIGINT; BEGIN SELECT currval('user_id_seq') INTO id; RETURN area || gp || LPAD(CAST(id AS VARCHAR), 4, '0'); EXCEPTION WHEN OTHERS THEN SELECT nextval('user_id_seq') INTO id; PERFORM setval('user_id_seq', id-1, false); SELECT currval('user_id_seq') INTO id; RETURN area || gp || LPAD(CAST(id AS VARCHAR), 4, '0'); END; $$ LANGUAGE plpgsql ;
↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑
これはナンセンスなので、
シーケンスの last_value を参照しましょう。
SELECT last_value FROM シーケンス
です。
CREATE OR REPLACE FUNCTION curuid(area CHAR, gp CHAR) RETURNS VARCHAR AS $$ DECLARE id BIGINT; BEGIN SELECT last_value INTO id FROM user_id_seq ; RETURN area || gp || LPAD(CAST(id AS VARCHAR), 4, '0'); END; $$ LANGUAGE plpgsql ;
Java においては、エンティティクラスで共通の列（カラム）は、基底クラスで定義して
継承クラスで解決していた。
同じ感覚で、Python SQLAlchemy で構成しようとしても、
Java と同様の単一継承の方法ではできない。
Python クラスの多重継承の方法で共通カラムの定義を行う。
【サンプル】
前提として、エンティティクラスは、dict 辞書型を返すメソッドと
JSON を返すメソッドを保有しているものとする。
最初の前提の為の基底クラス
entitybase.py
from datetime import date, datetime from sqlalchemy import inspect from sqlalchemy.ext.hybrid import hybrid_property import json class EBase: # dict 辞書型を返す (snakeCase ⇒ camelCase 対応） def to_dict(self) -> {}: dict_ = {} for key in self.__mapper__.c.keys(): if not key.startswith('_'): dict_[key] = getattr(self, key) for key, prop in inspect(self.__class__).all_orm_descriptors.items(): if isinstance(prop, hybrid_property): dict_[key] = getattr(self, key) return dict_ # JSON を返す def to_json(self) -> str: return json.dumps(self.to_dict(), default=lambda o: o.strftime('%Y-%m-%d %H:%M:%S') if isinstance(o, datetime) else o.strftime('%Y-%m-%d') if isinstance(o, date) else o.__dict__, indent=2)
日付と時刻を共通カラムとして持つ場合の
共通カラム基底クラス
ここで、上の EBase を継承する。
commoncolumns.py
import sqlalchemy as db from datetime import date, datetime from sqlalchemy import inspect from sqlalchemy.ext.hybrid import hybrid_property import json class CommonColumn: createDate = db.Column(db.DATE, name='create_date') createTime = db.Column(db.TIMESTAMP, name='create_time', nullable=False) def __init__(self, createDate:date, createTime:datetime): self.createDate = createDate self.createTime = createTime def __repr__(self): return f'createDate={self.createDate}, createTime={self.createTime}'
この commoncolumns.py で定義した CommonColumn を
目的のエンティティは、SQLAlchemy declarative_base と一緒に多重継承する。
テーブル名 = itemwork であるエンティティ
import sqlalchemy as db from sqlalchemy.orm import declarative_base from datetime import date, datetime from commoncolumns import CommonColumn class Twork(declarative_base(), CommonColumn): __tablename__ = 'itemwork' id = db.Column(db.BigInteger, nullable=False, primary_key=True) userName = db.Column(db.String(120), name='user_name', nullable=False) vPoint = db.Column(db.INT, name='v_point', nullable=False, default=0) def __init__(self, id:int, userName:str, vPoint:int, createDate:date, createTime:datetime): CommonColumn.__init__(self, createDate, createTime) self.id = id self.userName = userName self.vPoint = vPoint def __repr__(self): return f'Twork(id={self.id}, userName={self.userName}, vPoint={self.vPoint}, {CommonColumn.__repr__(self)})'
__init__ メソッドで、CommonColumn.__init__ を実行することに注意
__repr__ メソッドで、CommonColumn.__repr__(self) の結果を反映させている。
使用例
session.query(Twork).from_statement(text(sql)).all()
SqlAlchemy statement 実行するのではなく生成するSQL文を確認したい場合がある。
statement の compile() メソッドで確認することができる。
SQLAlchemy bulk upsert - Oboe吹きプログラマの黙示録
ここで書いたアップサートを例にしてみる。
compileメソッドの引数 dialect で PostgreSQL の dialectを指定して
PostgreSQL 対応SQLの結果になるようにする。
from sqlalchemy.dialects.mssql import dialect from stask import stasks from sqlalchemy.dialects import postgresql slist = [ stasks(1, 'Aaa', 23), stasks(2, 'Bbb', 33), stasks(4, 'Ddd', 44) ] stmt = postgresql.insert(stasks).values([e.toDict() for e in slist]) do_stmt = stmt.on_conflict_do_update( index_elements=['id'], set_=dict( color=stmt.excluded.color, vlen=stmt.excluded.vlen ) ) print(do_stmt.compile(dialect=postgresql.dialect()))
結果
INSERT INTO stasks (id, color, vlen) VALUES (%(id_m0)s, %(color_m0)s, %(vlen_m0)s), (%(id_m1)s, %(color_m1)s, %(vlen_m1)s), (%(id_m2)s, %(color_m2)s, %(vlen_m2)s) ON CONFLICT (id) DO UPDATE SET color = excluded.color, vlen = excluded.vlen
バインド値のところが、%s になってしまう。ちょっと期待と違う
literal_binds = True を指定することで、
print(do_stmt.compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}))
結果
INSERT INTO stasks (id, color, vlen) VALUES (1, 'Aaa', 23), (2, 'Bbb', 33), (4, 'Ddd', 44) ON CONFLICT (id) DO UPDATE SET color = excluded.color, vlen = excluded.vlen
SQLAlchemy bulk insert - Oboe吹きプログラマの黙示録
を書いたので、
ここで書いたテーブルを例に、bulk のアップサートを書いてみる。
対象データーベースがPostgreSQL なので、
sqlalchemy.dialects の中の postgresql を使って statement を作る。
マニュアルは、⇒ https://docs.sqlalchemy.org/en/21/dialects/postgresql.html#insert-on-conflict-upsert
@_generative @_on_conflict_exclusive def on_conflict_do_update(self, constraint: str | ColumnCollectionConstraint | Index | None = None, index_elements: Iterable[Column | str | DDLConstraintColumnRole] | None = None, index_where: WhereHavingRole | None = None, set_: Mapping | ColumnCollection | None = None, where: WhereHavingRole | None = None) -> Self
on_conflict_do_update() で、以下のように引数で DO UPDATE ～ に該当するものを
引数で指定する。
index_elements に、実際のSQL上の ON CONFLICT で指定する列名を指定する。
複数列を指定する場合があるので、リストを渡すことになる。
CONSTRAINTを作ったテーブルでは実際のSQLで ON CONFLICT ON CONSTRAINT 制約名
を使う方法もあるので、その場合は index_elements を指定せずに、
constraint で制約名を指定する。
set_ で、UPDATE SET文、更新を列名がキーの辞書型で指定する。
前の
SQLAlchemy bulk insert - Oboe吹きプログラマの黙示録
のテーブルを例に、アップサートのサンプルは以下のようになる。
from stask import stasks import database from sqlalchemy.dialects import postgresql slist = [ stasks(1, 'Aaa', 23), stasks(2, 'Bbb', 33), stasks(4, 'Ddd', 44) ] stmt = postgresql.insert(stasks).values([e.toDict() for e in slist]) do_stmt = stmt.on_conflict_do_update( index_elements = ['id'], set_ = dict( color = stmt.excluded.color, vlen = stmt.excluded.vlen ), ) with database.getSession(True) as session: res = session.execute(do_stmt) print(type(res)) print(res.rowcount)
実行結果で返ってくるクラスは、sqlalchemy.engine.cursor.CursorResult
rowcount には、insert 、update 両方の実行合計数が返ってくる。
Java mybatis で bulk insert するときに使った SQLMap XML に書くforeach のような
任意に記述するSQL生成での差し込んだ方法は、
SQLAlchemy の execute(text(SQL文) で同様の自由なSQL文での自動SQL生成は
無理みたいで残念です。
以下、DBの接続(database.py) と対象テーブルのクラス(stask.py)
が、別ソースにあるとして bulk insert のサンプルです。
sqlalchemy の insert() に、対象テーブル定義のクラスを指定して
execute の実行で、VALUES 複数に該当する dict のリストを渡します。
import database from sqlalchemy import insert from stask import stasks with database.getSession(True) as session: res = session.execute(insert(stasks).returning(stasks),[ {"id":1, "color":"A", "vlen":21}, {"id":2, "color":"B", "vlen":32}, {"id":3, "color":"C", "vlen":46} ]) print(type(res)) # <class 'sqlalchemy.engine.result.ChunkedIteratorResult'> res = res.fetchall() print("insert 行数 = %d" % res.__len__())
結果受取りで注意したいのは、
PostgreSQL BULK INSERT の RETURNING を Pythonで受け取る - Oboe吹きプログラマの黙示録
この時と違って、sqlalchemy.engine.cursor.CursorResult が返ってくるのではなく、
sqlalchemy.engine.result.ChunkedIteratorResult が返ってきます。
だから、rowcount 属性持っていないので、fetchall() 実行して RETURNING の結果を受け取ります。
returning() で、insert 対象テーブルクラスを指定することで、挿入データ値 Object を
返してもらいます。
クラスの属性だけを指定するれば、それだけ返ってきます。
上の例では、SQL で
RETURNING id
と SQLAlchemy 実行の
session.execute(insert(stasks).returning(stasks.id),[
は、同じ id 列だけが返ってきます。
stask.py は、以下のように用意します。
import sqlalchemy as db from sqlalchemy.orm import declarative_base from sqlalchemy import inspect from sqlalchemy.ext.hybrid import hybrid_property class stasks(declarative_base()): __tablename__ = 'stasks' id = db.Column(db.BigInteger, nullable=False, primary_key=True) color = db.Column(db.String(32), nullable=False) vlen = db.Column(db.INT, nullable=False, default=0) def __init__(self, id:int, color:str, vlen:int): self.id = id self.color = color self.vlen = vlen def __repr__(self): return f'stask(id={self.id}, color={self.color}, vlen={self.vlen})' def toDict(self) -> {}: dict_ = {} for key in self.__mapper__.c.keys(): if not key.startswith('_'): dict_[key] = getattr(self, key) for key, prop in inspect(self.__class__).all_orm_descriptors.items(): if isinstance(prop, hybrid_property): dict_[key] = getattr(self, key) return dict_
class stasks が toDict() でdictを返すものを用意しておくことで、
以下のように書けます。
slist = [ stasks(1, 'A', 21), stasks(2, 'B', 32), stasks(3, 'C', 46) ] with database.getSession(True) as session: res = session.execute(insert(stasks).returning(stasks), [e.toDict() for e in slist])
PostgreSQL のアップサートで数年前にSQL内のテーブル指定の衝突回避で
PostgresSQL アップサートのインクリメント - Oboe吹きプログラマの黙示録
を書いたことがあります。
EXCLUDED 指定を使用すれば、BULKアップサートとして実行する
ON CONFLICT～DO UPDATE 文も書けるのですが、
苦労して WITH 句を使う方法もあります。
WITH 句を使う方法だけで解決する方法と
ON CONFLICT～DO UPDATE でのBULK UPSEERT の方法を
まとめておきたいと思います。
サンプルとして以下のテーブル構造です。
CREATE TABLE stasks ( id INTEGER NOT NULL, color varchar(32) NOT NULL, vlen INTEGER DEFAULT 0 NOT NULL, CONSTRAINT stasks_pkey PRIMARY KEY (id) );
WITH句による、ON CONFLICT～DO UPDATE を使用しない
方法が以下です。
WITH wk(id, color, vlen) AS ( VALUES (1, 'A', 55), (2, 'B', 66), (4, 'D', 80) ), b AS ( UPDATE stasks a SET color = wk.color, vlen = wk.vlen FROM wk WHERE a.id = wk.id RETURNING a.id ) INSERT INTO stasks (id, color, vlen) SELECT wk.id, wk.color, wk.vlen FROM wk WHERE wk.id NOT IN (SELECT id FROM b)
WITH句で作ったテーブルデータのPrimaryKeyで、UPDATE の実行 b の
RETURNING 句で返すPrimaryKey の NOT INで INSERTする方法です。
WITH句による、ON CONFLICT～DO UPDATE
を使用する方法が以下です。
PostgreSQL マニュアル書かれている
The SET and WHERE clauses in ON CONFLICT DO UPDATE have access to the existing row using the table's name (or an alias), and to the row proposed for insertion using the special excluded
ON CONFLICT DO UPDATE における SET 句と WHERE 句では、
テーブル名（またはエイリアス）を使用して既存の行にアクセスでき、
また、特殊な excluded を使用して挿入対象の行にアクセスできます。
これだけでは、対処方法が良くわからないのですが、]
次のように、SET文の右辺の方に、EXCLUDED で、更新としてセットする側の
テーブルの指定とすることができます。
WITH wk(id, color, vlen) AS ( VALUES (1, 'A', 50), (2, 'B', 60), (4, 'D', 80) ) INSERT INTO stasks (id, color, vlen) SELECT id, color, vlen FROM wk ON CONFLICT (id) DO UPDATE SET color = EXCLUDED.color, vlen = EXCLUDED.vlen
https://oboe2uran.hatenablog.com/entry/2023/10/01/003000
の例では、INSERT INTO テーブルの
テーブル名を指定して更新するテーブルカラムの曖昧さを解決していたのですが、
右辺にEXCLUDEDを指定するのです。
