🔖
SQLAlchemyのQuery API
Document
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'
Discussion