UTF-16 LE の判定とBOM

時折、他システム Windows 環境から貰うことになったCSVが、
UTF-16 LE (リトルエンディアン)だったりする。

UTF-8 の BOM は、efbbbf
UTF-16 LE の BOM は、fffe
UTF-16 BE の BOM は、feff

である。

Python の標準ライブラリではない chardet による文字コード判定で、
  https://pypi.org/project/chardet/
このUTF-16の BOM 有/無 の判定結果は、BOMが付くとどちらのエンディアンかわからなくなる。

BOM無しで、
UTF-16 LE を chardet で判定すると

{'encoding': 'utf-16-le', 'confidence': 0.95, 'language': 'ja', 'mime_type': 'text/plain'}

UTF-16 BE を chardet で判定すると

{'encoding': 'utf-16-be', 'confidence': 0.95, 'language': 'ja', 'mime_type': 'text/plain'}

BOM有りで、
chardet で判定すると、UTF-16 BE 、UTF-16 LE 両方とも

{'encoding': 'UTF-16', 'confidence': 1.0, 'language': 'ja', 'mime_type': 'text/plain'}

と、どちらのエンディアンかを示すことができない。

with open(filepath, 'rb') as f:
  c = f.read()
  result = chardet.detect(c)
  print(result)

PostgreSQLでのUUID生成 ⇒ ハイフン無しの文字列へ

UUID生成してハイフン無しの文字列を取得するのは、Java や Python それぞれの言語では
Java は、.toString().replace("-", "") で、Python は、.hex を実行するであろう。

PostgreSQL UUID生成 ⇒ ハイフン無しの文字列

UUID version 4

SELECT REPLACE(gen_random_uuid()::text, '-', '')

UUID version 7

SELECT REPLACE(uuid_generate_v7()::text, '-', '')


ついでに Python の hex 属性を使う場合
UUID version 4

import uuid

_uuidstr = uuid.uuid4().hex

UUID バージョン7は、Python3.14以降なら同様に uuid7() で良いのだが、
Python3.13 以前では未対応なので uuid7() が使えない。
どうする?と思ってたら、、、
以下、丁寧に書いてくださっているのがあって非常に助かりました。感謝です。
https://zenn.dev/timoneko/articles/dee5374bc3cc61

このソースを uuidv7.py として以下のように使用します。

import uuidv7

_uuidstr = uuidv7._uuid7().hex

シーケンス生成直後の last_value

GENERATED ALWAYS AS IDENTITY で採番する列を持つテーブルを
  TRUNCATE TABLE テーブル名
を実行したとき採番はまだ続いてしまうので、
初期値から振るようにする場合は、
  TRUNCATE TABLE   テーブル名 restart identity
で行う。

GENERATED ALWAYS AS IDENTITY は、
自動でシーケンスが作成されているので、
テーブル生成時と TRUNCATE TABLE テーブル名  restart identity 実行後は、
  SELECT last_value FROM 自動生成されたシーケンス
は、1が返ってくる。
シーケンスの生成して nextval を実行していない状態も
  SELECT last_value FROM シーケンス
は、1が返ってくる。

ストアドファンクションでシーケンス呼び出し(PostgreSQL)

ストアドファンクションでシーケンスを呼び出して加工して返す。
任意の文字2個+シーケンス0埋め4桁を作る目的です。

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
;

SQLAlchemy 共通の列定義

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 compile

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 upsert

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 両方の実行合計数が返ってくる。