LoginSignup
5

More than 1 year has passed since last update.

SQLAlchemy 2.0でよく使う文法まとめ

Posted at

はじめに

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())

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
5