はじめに

PythonのRDBのORMapperの代表的なものとしては、SQLAlchemyがメジャーかと思います。
今SQLAlchemyはメジャーバージョン2へのアップデートへの移行に伴い、バージョン1.4では新しい文法を使えるようになっています。

新しい文法での、一般的なクエリの書き方を備忘的に残します。

コネクションの作成

from sqlalchemy import create_engine
from sqlalchemy.engine.url import URL
engine = create_engine(URL(...接続設定))

print(next(engine.execute('select 1')))

with engine.connect() as conn:
    print(next(conn.execute('select 1')))

エンジンオブジェクト自体もクエリを実行できるし、コネクションオブジェクトもクエリを実行できる

セッションの作成

from sqlalchemy import create_engine
from sqlalchemy.orm import Session
engine = create_engine("sqlite+pysqlite:///:memory:")

with Session(engine) as session:
    print(next(session.execute('select 1')))

セッションも従来と同じように作成できる

モデルの作成とShowTables

from sqlalchemy import create_engine, Column
from sqlalchemy.dialects.mysql import VARCHAR
from sqlalchemy.engine.reflection import Inspector
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

class User(Base):
    __tablename__ = "user"

    user_id = Column(VARCHAR(10), primary_key=True)
    user_name = Column(VARCHAR(10))

engine = create_engine("sqlite+pysqlite:///:memory:")

print(Inspector(engine).get_table_names())
Base.metadata.create_all(engine)
print(Inspector(engine).get_table_names())

Insert

エンジンとモデルは上記を再利用しています

from sqlalchemy.orm import Session
from sqlalchemy import select

with Session(engine) as session:
    session.add(User(user_id="test_id1", user_name="test_name1"))
    session.commit()

print(engine.execute(select(User)).all())

モデルを使う書き方は、上記のような以前と変わらない書き方や、下記のような書き方もできるようになっています

from sqlalchemy import insert

engine.execute(
    insert(User)
    .values(user_id='test_id1', user_name='test_name1')
)

print(engine.execute(select(User)).all())

Select

where

from sqlalchemy.orm import Session
from sqlalchemy import select

with Session(engine) as session:
    session.add(User(user_id="test_id1", user_name="test_name1"))
    session.add(User(user_id="test_id2", user_name="test_name2"))
    session.commit()

stmt = select(User).where(User.user_name == 'test_name2')
print(engine.execute(stmt).all())

group_by

from sqlalchemy.orm import Session
from sqlalchemy import select, func

with Session(engine) as session:
    session.add(User(user_id="test_id1", user_name="test_name1"))
    session.add(User(user_id="test_id2", user_name="test_name2"))
    session.add(User(user_id="test_id3", user_name="test_name2"))
    session.commit()

stmt = (
    select(User.user_name, func.count())
    .group_by(User.user_name)
)
print(engine.execute(stmt).all())