SQLAlchemy コネクション、セッション

以前、あまりにも内容が薄い安易なものをここに書いたので、もう少しまともなメモを
書こうと思った。

create_engine

from sqlalchemy import create_engine, text

host = 'localhost'
port = 5432
username = 'myuser'
password = 'p1234'
dbname = 'sampledb'
engine = create_engine(f'postgresql://{username}:{password}@{host}:{port}/{dbname}')

with engine.begin() as connection:
    sql = 'SELECT * FROM samples'
    rows = connection.execute(text(sql))

rows は、sqlalchemy.engine.cursor.CursorResult
for文で展開すると

    for row in rows:
        print(row)

この row は、class sqlalchemy.engine.row.Row の Object で、row[0] は１列目の値
rows = connection.execute(text(sql)) の結果を dict の配列にするなら
.mappings().all()を使い、

    for row in rows.mappings().all():
        for columname, value in row.items():
            print(f'{columname} -> {value}')

クエリ結果行数の算出

with engine.begin() as connection:
    sql = 'SELECT COUNT(*) FROM samples'
    rows = connection.execute(text(sql))
    count = list(rows)[0][0]
    print('count = %d' % count)

list(rows) は、レコード数=6 の時は、[(6,)] である
だから list(rows)[0][0] でクエリ結果行数を取得する。

Session で実行する

from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session

Session() で作る

host = 'localhost'
port = 5432
username = 'myuser'
password = 'p1234'
dbname = 'sampledb'
engine = create_engine(f'postgresql://{username}:{password}@{host}:{port}/{dbname}')

with Session(autocommit=False, autoflush=True, bind=engine) as session:
    with session.begin():
        sql = 'SELECT * FROM samples'
        rows = session.execute(text(sql))

sessionmaker で作る

from sqlalchemy.orm import sessionmaker

sfactory = sessionmaker(autocommit=False, autoflush=True, bind=engine)
with sfactory() as session:
    sql = 'SELECT * FROM samples'
    rows = session.execute(text(sql))

scoped_session を使う

from sqlalchemy.orm import sessionmaker, 

sfactory = sessionmaker(autocommit=False, autoflush=True, bind=engine)
Session = scoped_session(sfactory)
session1 = Session()
session2 = Session()

threading.local() となるので、session1 と session2 は、同一セッションになる。

Session の本来の使い方は、、

with Session(engine) as session:
    session.begin()
    try:
        session.add(some_object)
        session.add(some_other_object)
    except:
        session.rollback()
        raise
    else:
        session.commit()