SQLAlchemyのQuery API

Document

https://docs.sqlalchemy.org/en/14/orm/query.html

Version

• SQLAlchemy: 1.4.40

モデル定義

class User(Base):
  __tablename__ = "users"

  id = Column(Integer, primary_key=True, autoincrement=True)


class Post(Base):
  __tablename__ = "posts"

  id = Column(Integer, primary_key=True, autoincrement=True)
  user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
  title = Column(String(50), nullable=False)

sqlalchemyが発行するSQLを確認する方法

session.query(Post).statement.compile().string
# => 'SELECT posts.* \nFROM posts'

all()

全てのレコードを返す

session.query(Post).all()
# => [<Post 1>, <Post 2>]

count()

レコードの件数を返す

session.query(Post).count()
# => 2

group by や distinct と組み合わせて count したい場合

from sqlalchemy import func

session.query(func.count(Post.id)).statement.compile().string
# => 'SELECT count(posts.id) AS count_1 \nFROM posts'

session.query(func.count(Post.id)).group_by(Post.user_id).statement.compile().string
# => 'SELECT count(posts.id) AS count_1 \nFROM posts GROUP BY posts.user_id'

session.query(func.count(distinct(Post.user_id))).statement.compile().string
# => 'SELECT count(DISTINCT posts.user_id) AS count_1 \nFROM posts'

exists()

EXISTS のサブクエリを生成する

query = session.query(Post).filter(Post.title == 'title1')

session.query(query.exists()).statement.compile().string
# => 'SELECT EXISTS (SELECT 1 \nFROM posts \nWHERE posts.tirle = :title1) AS pole_1'

session.query(query.exists()).scalar()
# => True

filter()

WHERE で条件を絞る

session.query(Post).filter(Post.title == 'title1').statement.compile().string
# => 'SELECT posts.* \nFROM posts \nWHERE posts.title = :title1'

filter_by()

SQLは filter() と同じ
こっちの方が短く書ける
join() と合わせて絞り込みたい場合は filter() を使った方が良さそう

session.query(Post).filter_by(title = 'title1').statement.compile().string
# => 'SELECT posts.* \nFROM posts \nWHERE posts.title = :title1'

first()

レコードを1件だけ取得する

session.query(Post).first()
# => <Post 1>

レコードが存在しない場合Noneを返す

session.query(Post).filter_by(id=100).first()
# => None

get()

primary id を指定して1つのレコードを返す
もし存在しない場合はNoneを返す

session.query(Post).get(1)
# => <Post 1>

join()

JOIN（INNER JOIN）する

session.query(Post).join(User).statement.compile().string
# => 'SELECT posts.* \nFROM posts JOIN users ON users.id = posts.user_id'

session.query(Post).join(User).filter(User.id == 1).statement.compile().string
# => 'SELECT posts.* \nFROM posts JOIN users ON users.id = posts.user_id \nWHERE users.id = :id_1'

もしモデル定義で relationship を付けていれば、Post.user みたいに書ける

session.query(Post).join(Post.user).statement.compile().string
# => 'SELECT posts.* \nFROM posts JOIN users ON users.id = posts.user_id'

limit()

LIMIT を指定する

session.query(Post).limit(1).all()
# => [<Post 1>]

one()

1件だけレコードを取得する

session.query(Post).filter_by(title = 'title1').one()
# => <Post 1>

2件以上ある場合エラーになる

session.query(Post).one()
# => *** sqlalchemy.exc.MultipleResultsFound: Multiple rows were found when exactly one was required

1件もない場合エラーになる

session.query(Post).filter_by(title = 'title100').one()
# => *** sqlalchemy.exc.NoResultFound: No row was found when one was required

one_or_none()

session.query(Post).filter_by(title = 'title1').one_or_none()
# => <Post 1>

2件以上ある場合エラーになる

session.query(Post).one_or_none()
# => *** sqlalchemy.exc.MultipleResultsFound: Multiple rows were found when one or none was required

1件もない場合Noneが返る

session.query(Post).filter_by(title = 'title100').one_or_none()
# => None

order_by()

ORDER BY を指定する

session.query(Post).order_by(Post.id).statement.compile().string
# => 'SELECT posts.* \nFROM posts ORDER BY posts.id'

from sqlalchemy import desc
session.query(Post).order_by(desc(Post.id)).statement.compile().string
# => 'SELECT posts.* \nFROM posts ORDER BY posts.id DESC'

outerjoin()

OUTER JOIN する
使い方は join() と同じ

session.query(Post).outerjoin(User).statement.compile().string
# => 'SELECT posts.* \nFROM posts LEFT OUTER JOIN users ON users.id = posts.user_id'

where()

filter() と同じ

with_for_update()

クエリに FOR UPDATE を付与する

session.query(Post).with_for_update().statement.compile().string
# => 'SELECT posts.* \nFROM posts FOR UPDATE'