Kumanote LLC.

2020/09/29

[Python]SQLAlchemyのよくやる使い方まとめ

pythonpostgresql

最近では、PythonのWebアプリのバックエンドで使うことが多く、
そのRDBのDriverとして、SQLAlchemyを採用することが多いです。
（使っているDBはpostgreqlです。）

今回は、SQLAlchemyの実際によく使う使い方をまとめてみました。

どうやって実装したっけ？とよく昔のプロジェクトのソースコードを参照することが多かったので、自分向けのメモとしても残しておきます。

Model編

Base Class

以下のような基底クラスを定義しておき、それを継承させると便利です。

from sqlalchemy.ext.declarative import as_declarative, declared_attr


@as_declarative()
class Base:
    @declared_attr
    def __tablename__(cls):
        return cls.__name__.lower()

Modelのフィールド定義

ID

  • auto incrementするprimary key
from sqlalchemy import Column, Integer

class User(Base):  # Baseクラスを継承
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)

Unique Key

from sqlalchemy import Column, String

class User(Base):
    __tablename__ = "users"
    email = Column(String, unique=True, index=True)

Unique Constraint

  • 複数カラムでの一意制約
from sqlalchemy import Column, Integer, UniqueConstraint

class Sample(Base):
    __tablename__ = "samples"
    user_id = Column(Integer, index=True, nullable=False)
    item_id = Column(Integer, index=True, nullable=False)
    __table_args__ = (UniqueConstraint(
        'user_id', 'item_id', name='unique_user_id_item_id'),)  # user_idとitem_idの組み合わせで一意になる

Index for Multiple Columns

  • 複数カラムでのIndex
from sqlalchemy import Column, Integer, Index

class Sample(Base):
    __tablename__ = "samples"
    user_id = Column(Integer, index=True, nullable=False)
    item_id = Column(Integer, index=True, nullable=False)
    __table_args__ = (Index(
        'idx_user_id_item_id', 'user_id', 'item_id'),)  # user_idとitem_idの組み合わせでINDEXを作成

Foreign Key

  • 親が削除されたら、子も削除
from sqlalchemy import Column, ForeignKey, Integer

class Item(Base):
    __tablename__ = "items"
    owner_id = Column(Integer, ForeignKey("users.id", ondelete="CASCADE"))
  • 親が削除されたら、NULLになる
from sqlalchemy import Column, ForeignKey, Integer

class Item(Base):
    __tablename__ = "items"
    owner_id = Column(Integer, ForeignKey("users.id", ondelete="SET NULL"), nullable=True)

Populates

外部参照制約が効いているカラムを使って、ちょっと便利に参照するデータをプロパティ経由で取得可能になる
（あまり多用しないようにしています）

from sqlalchemy.orm import relationship

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    items = relationship("Item", back_populates="owner")  # itemsというプロパティで、このユーザーが所持するitemのリストを取得可能になる


class Item(Base):
    __tablename__ = "items"
    owner_id = Column(Integer, ForeignKey("users.id", ondelete="CASCADE"))
    owner = relationship("User", back_populates="items")  # ownerというプロパティで、このitemを所持するuserを取得可能になる

    # もしusersテーブルに対して複数のカラムでリレーションがある場合は、relationshipにどのカラムのリレーションを使うか明示する必要があるので注意
    # その場合は以下のような感じになる
    # owner = relationship("User", back_populates="items", foreign_keys=[owner_id])

Nullable & Has max length

from sqlalchemy import Column, String

class User(Base):
    __tablename__ = "users"
    bio = Column(String(250), nullable=True)

Json

from sqlalchemy import Column, JSON

class User(Base):
    __tablename__ = "users"
    json_data = Column(JSON, nullable=True)

created_at

from sqlalchemy import DateTime
from sqlalchemy.sql.functions import current_timestamp

class User(Base):
    __tablename__ = "users"
    created_at = Column(DateTime, nullable=True, server_default=current_timestamp())

その他

カラムのデータ型などの詳しい対応はこちらをよく見ます。

SQL編

sessionの作り方

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URI = "postgresql://postgres-username:postgres-password@postgres-host/postgres-dbname"
engine = create_engine(DATABASE_URI, pool_pre_ping=True)
Session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
db_session = Session()  # webアプリ場合、middlewareなどで作成します

Select by ID

  • 以下 User は modelクラス
  • db_session は sqlalchemy.orm.Sessionインスタンス
from typing import Optional

# db_userはUserモデルのインスタンス
db_user: Optional[User] = db_session.query(User).filter(User.id == id).first()

Select multi / Like

from typing import List

db_users: List[User] = db_session.query(User).filter(User.email.like("%@gmail.com")).all()

Count

db_users_cnt: int = db_session.query(User).filter(User.email.like("%@gmail.com")).count()

Limit / Offset

from typing import List

offset = 20
limit = 10
db_users: List[User] = db_session.query(User) \
    .filter(User.email.like("%@gmail.com")) \
    .offset(offset).limit(limit).all()

Sum

from sqlalchemy import func

total_user_balance = db_session.query(func.sum(User.balance)).first()

Case

from sqlalchemy import func, case

# is_deletedフラグが立っていないユーザーの所持金額の合計を取得
total_user_balance = db_session.query(
    func.sum(
        case(
            [(User.is_deleted.is_(False), User.balance)],
            else_=0
        )
    )
).first()

Filter by Boolean field

  • is_ とか isnot_が使える
from typing import List

# 削除済みのユーザーを取得
db_users: List[User] = db_session.query(User).filter(User.is_deleted.is_(True)).all()

Case Insensitive

  • 大文字・小文字を無視して検索する
from sqlalchemy import func

search_text = "[email protected]"
db_users: List[User] = db_session.query(User).filter(func.lower(User.email).like(f"%{search_text.lower()}%")).all()

Subquery

from typing import List

sub_query = db_session.query(User.id).filter(User.email.like("%@gmail.com"))
db_items = db_session.query(Item).filter(Item.owner_id.in_(sub_query)).all()

Not Exists

from typing import List
from sqlalchemy import and_, exists

users = db_session.query(User).filter(~exists().where(and_(Item.owner_id == User.id, Item.deleted.is_(True)))).all()

Select Null

from sqlalchemy import null

db_session.query(
  Item,
  null().label("force_null_field"),
).all()

Inner Join

db_session.query(Item, User).join(User, User.id == Item.owner_id).all()

Outer(Left) Join

単一条件でjoin

db_session.query(Item, User).outerjoin(User, User.id == Item.owner_id).all()

複数条件でjoin

from sqlalchemy import and_
db_session.query(Item, User) \
    .outerjoin(
        User,
        and_(
            User.id == Item.owner_id,
            User.id >= 10,
        )
    ) \
    .all()

同名テーブルをJOIN

aliasedをうまく使います。こうすることで、joinするテーブルのラベリングができます。

from sqlalchemy.orm import aliased

user1 = aliased(User)
user2 = aliased(User)

db_session.query(Item, user1, user2) \
  .outerjoin(user1, user1.id == Item.owner_id) \
  .outerjoin(user2, user2.id == Item.owner_id) \
  .all()

subqueryの結果とjoin

subquery()の結果をc.field_nameで参照して使うことができます。

sub_query = db_session.query(User.id, func.sum(User.some_point).label("total_points")).group_by(User.id).subquery()
db_session.query(Item, sub_query.c.total_points) \
  .outerjoin(sub_query, sub_query.c.id == Item.owner_id) \
  .all()

Union All

user_query = db_session.query(
    User.id.label("id"),
    User.name.label("name"),
    User.email.label("email"),
)
admin_query = db_session.query(
    Administrator.id.label("id"),
    Administrator.name.label("name"),
    Administrator.email.label("email"),
)
user_query.union_all(admin_query).all()

Raw SQL

Fetch one

  • emailをparameterとして、emailに合致するユーザーを取得
import textwrap
from sqlalchemy import text

sql = textwrap.dedent('''\
select
    id
    , name
    , email
  from
    users
  where
    email = :email
  limit 1''')
db_user = db_session.execute(text(sql), params={"email": "[email protected]"}).fetchone()

Fetch all

  • emailをparameterとして、emailに後方一致するユーザーを全て取得
import textwrap
from sqlalchemy import text

sql = textwrap.dedent('''\
select
    id
    , name
    , email
  from
    users
  where
    email like '%:email'''')
db_users = db_session.execute(text(sql), params={"email": "@example.com"}).fetchall()

以上です。
今後頻繁に使うものがあれば随時追加していこうと思います。

関連する記事

psqlコマンドを使い、SQLファイルを実行して、csv/tsvに結果を出力する

postgresqldbサーバーにログインして、SQLファイルを実行して結果をcsv/tsvに出力する方法です。

[Python]ハイフンなし電話番号からハイフン付きに復元

Pythonでハイフンなしの日本の電話番号をハイフン付きのものに変換する

[Rust][Psql]Dieselのよくやる使い方まとめ

Rust Dieselのよくやる使い方をまとめてみました

POSTGRESQLで行更新時に自動で更新日時を設定する

TRIGGERを使って、行更新時に更新日時を特定のフィールドに設定できるようにしました