SQLAlchemyはPythonでよくつかわれるORマッパー。DjangoのORマッパーは使ってたけどSQLAlchemyは最近よく使うようになったのでDBからのデータ取得やCRUDなどをまとめます。
以前の関連記事。Flaskで Flask SQLAlchemyを使ったDBデータ取得、作成、更新、削除を行ってます。
Djangoで同じような記事書いてたので一応リンク載せます。
今回のSQLAlchemyとはORMと言うだけであまり関係ないですが、Django ORMでもSQLAlchemyでもどちらかできればそれほど違和感なく使えるようになるかと思います。
ついでに、SQLAlchemyの方がSQLの書き方に近い感じはします。
環境
この記事の最後に補足に今回使うSQLAlchemyの環境構築書いてますのでそちらを参照。
db_session
と言う名前に意味はないので記事や書き方によって名前が違うと思いますので注意。
- Windows10
- Python 3.8
- SQLite
- SQLAlchemy 2.0
あと、Flask SQLAlchemyなどで使っていると少し違うところが気になりますが、使えるメソッドなどは基本は同じなので適時読み替えてください。
# 今回のSQLAlchemyの書き方の例
employees = db_session.query(Employee).all()
employee = db_session.query(Employee).get(1)
employees = db_session.query(Employee).filter(Employee.id == 2)
# Flask SQLAlchemyの書き方の例
employees = Employee.query.all()
employee = Employee.query.get(1)
employee = Employee.query.filter(Employee.id == 2)
あとFlask SQLAlchemyを使ったDBのCRUDについては以前記事書いたので確認していただけると幸いです。
1. get: データ1つ取得
filterと似てますが、データを「1つだけ」取得します。
- 単体のオブジェクト取得
- プライマリーキーを指定
- データがなかったら
None
を返す
雑ですが、例えると、
filterで取ったものはリストみたいにforなどで取り出してやらないと個々のデータ(レコードみたいなイメージ)は使えない。getは取り出さなくても個々のデータ(レコード)は使える。
と思っておけばとりあえず良いと思います。
一般的にはidをプライマリーキーにしていることが多いと思います。詳細ページの情報を取得するときなどに使えますね。
データがなくてもエラーが出ないです。DjangoのORMにもgetありましたがここは違いますね。
例:プライマリーキー(この場合はid)が1
のデータ取得
employee = db_session.query(Employee).get(1)
print(f'{employee.name}さん 勤続{employee.year}年')
実行結果
Tanakaさん 勤続2年
Query.get()はversion 1.4から非推奨だそうです。1.4以降はSession.get()が使えますのでそちら使うようにとのこと。
ただ、普通にしばらくは使えると思います。 2021/12
version 1.4以上の場合
version 1.4以上ならこちら使った方が無難らしい。
書き方が少し変わるくらいです。変わったところはquery
ではないところ。
db_session.query(Employee).get(1)
↓
db_session.get(Employee, 1)
この二つは同じデータ取れます。
employee = db_session.get(Employee, 1)
2.1 all:データすべて取得する
employees = db_session.query(Employee).all()
for employee in employees:
print(f'{employee.name}さん 勤続{employee.year}年')
Tanakaさん 勤続2年
Takahashiさん 勤続3年
Satoさん 勤続10年
Saitoさん 勤続5年
SELECT * FROM employee
を実行することになります。大量のデータが取得されるので実際の開発ではあまり使わないかも。
2.2 filterで条件付きデータ取得まとめ
SQLだったらwhereで条件付けるってことです。
filter() が一般的ですが、version 1.4から where() でも大丈夫になりました。filterをwhereに置き換えても動きます。完全に好みの問題な気がします。
完全一致
セッション.query(モデル).filter(条件)
例:idが1のデータ取得
employees = db_session.query(Employee).filter(Employee.id == 1)
2.2.1 AND条件
複数の条件付ける。方法は二つある。
パターン1
単純な方。カンマ「,
」で区切ると複数の条件を付けられる。
例:nameがSato、yearが10のデータ取得
employees = db_session.query(Employee).filter(Employee.name == 'Sato', Employee.year == 10)
パターン2
and_
というものをimportする必要あり。カンマで複数の条件を指定
セッション.query(モデル).filter(and_(条件1, 条件2))
例:nameがSato、yearが10のデータ取得
from sqlalchemy import and_
employees = db_session.query(Employee).filter(and_(Employee.name == 'Sato', Employee.year == 10))
importする必要があり、記述が少し長くなる。次に説明するOR条件と書き方を一緒にするならこちらを使ってもいいかもしれないとは思うが、分かりやすい方でよいと思う。
2.2.2 OR条件
or_
というものをimportする必要あり。and_
と使い方同じ。カンマで複数の条件を指定。
例:nameがSatoもしくはyearが2のデータ取得
employees = db_session.query(Employee).filter(or_(Employee.name == 'Sato', Employee.year == 2))
※ ついでにand_
とor_
くっつけて複合条件もちろん作れます。
2.2.3 IN句 (where in)
filterに条件をつけまくる方法もあるけど、条件をリストで渡して条件に一致するデータを取得できてとてもスマートに書ける。in_([...])
を使う。
セッション.query(モデル).filter(モデル.フィールド.in_([リスデータ]))
例:nameがSato, Takahashi, Tanakaの、どれかだったら取得
employees = db_session.query(Employee).filter(Employee.name.in_(['Sato', 'Takahashi', 'Tanaka']))
NOT IN句 (where not in)
where in の逆。リストであたえたデータが含まないデータを取得する。
例:nameがSato, Takahashi, Tanakaでなかったら取得
employees = db_session.query(Employee).filter(~Employee.name.in_(['Sato', 'Takahashi', 'Tanaka']))
すごくわかりにくいかもしれないけどモデルの前に~
がついてます。最初公式ドキュメント見たときinと同じじゃん。
誤植か?と思いました。
https://docs.sqlalchemy.org/en/14/orm/tutorial.html#querying
以下は同じ意味なので下記のように書く方が分かりやすいと思います。
employees = db_session.query(Employee).filter(Employee.name.notin_(['Sato', 'Takahashi', 'Tanaka']))
2.2.4 不等号使った条件(「>」「<」「>=」「<=」)
普通に不等号使えば比較が条件を作ることができます。
より大きい・超える「>」
例:「社歴が2年より大きい」(year > 2
)なら
employees = db_session.query(Employee).filter(Employee.year > 2)
以上「>=」
例:「社歴が2年以上」(year >= 2
)なら
employees = db_session.query(Employee).filter(Employee.year >= 2)
未満「<」
例:「社歴が2未満」(year < 2
)なら
employees = db_session.query(Employee).filter(Employee.year < 2)
以下「<=」
例:「社歴が2以下」(year <= 2
)なら
employees = db_session.query(Employee).filter(Employee.year <= 2)
値の範囲指定
単純にカンマで区切って複数つければOK
例:「社歴が2年より大きく5年以下」(2 < year <= 5
)
employees = db_session.query(Employee).filter(Employee.year > 2, Employee.year <= 5)
2.2.5 BETWEEN句
値の範囲指定でも同じようなことができますがBETWEENでも可能です。
数値の範囲や日時期間をstartとendを指定。
from sqlalchemy import between
をimportする必要があります。
セッション.query(モデル).filter(between(比較するもの, start, end))
例:SQLにて、age BETWEEN 0 AND 20
つまり、2 <= year AND year <= 5
from sqlalchemy import between
employees = db_session.query(Employee).filter(between(Employee.year, 2, 5))
少し違いますがこんな感じのSQLになります。
SELECT * FROM employee
WHERE employee.year BETWEEN 2 AND 5;
これはSQLの話ですがBETWEENは指定したは「含む」ので注意。
2.2.6 LIKE句
部分一致について記述します。一応ですが、完全一致はlike使わずにfilter使えばOKです。
今回記述していく方法でも可能ですが、BRSFさんのコメントの方がスマートとのことですのでぜひ参考に!!
文字列のどこかに指定した文字列を含んでいる場合
書き方二つある。SQLっぽく素直に書くならlike
ですがcontains
の方がPythonで書くならよい気がします。好みですけど。
例:「ha」を含む名前のデータ取得
employees = db_session.query(Employee).filter(Employee.name.like('%ha%'))
employees = db_session.query(Employee).filter(Employee.name.contains('ha'))
実行結果
Takahashiさん 勤続3年
名前の中に小文字で「ha」が入っている「Takaha
shi」さんのデータが取得できる
例:「Sa」から始まる名前のデータ取得
前方一致
書き方二つある。こちらも好みにて。
例:Tから始まる名前取得
employees = db_session.query(Employee).filter(Employee.name.startswith('T'))
employees = db_session.query(Employee).filter(Employee.name.ilike('T%'))
後方一致
書き方二つある。こちらも好みにて。
例:aで終わる名前取得
employees = db_session.query(Employee).filter(Employee.name.endswith('a'))
employees = db_session.query(Employee).filter(Employee.name.like('a%'))
大文字と小文字
使用するlike()
は大文字小文字を見るのですが、今回使用したSQLiteでは大文字と小文字見ないらしいので大文字でも小文字でも関係なく取得できていました。
ilike()
を使うと、指定した文字列を小文字に変換してからSQLで取得する処理をしてくれます。
employees = db_session.query(Employee).filter(Employee.name.ilike('TA%'))
を実行したら
↓生成したSQLでこんなのを入れてくれる。
WHERE lower(employee.name) LIKE lower(?)
like()
だとlower()
が入りません。
2.2.7 NULLを検索(IS NULL)、除外(IS NOT NULL)
単純な== None
でIS NULL
になります。ただ、Pythonはis None使うしなんか気になるという人はis_(None)
を使うことも可能です。
セッション.query(モデル).filter(モデル.フィールド == None)
もしくは
セッション.query(モデル).filter(モデル.フィールド.is_(None))
これでNULLを取得。
例:nameがNULLのデータ取得
employees = db_session.query(Employee).filter(Employee.name == None)
employees = db_session.query(Employee).filter(Employee.name.is_(None))
NULLを除外は
セッション.query(モデル).filter(モデル.フィールド != None)
もしくは
セッション.query(モデル).filter(モデル.フィールド.is_not(None))
例:nameがNULLでないデータ取得
employees = db_session.query(Employee).filter(Employee.name != None)
employees = db_session.query(Employee).filter(Employee.name.is_not(None))
2.2.8 order_by: データ並べ替え
order_by
というのを付けるとデータを並べ替えることが可能。SQLのorder_byと同じような感じで使えるのでわかりやすいと思う。
昇順
order_by使うとデフォルトは昇順になる。
例:is_remote Falseで勤続年数にて昇順で並び替え
employees = db_session.query(Employee).filter(Employee.is_remote == False).order_by(Employee.year)
Tanakaさん 勤続2年
Takahashiさん 勤続3年
Satoさん 勤続10年
昇順の場合つける意味もないですが明示的に書きたい場合は以下のようにも書けます。
employees = db_session.query(Employee).filter(Employee.is_remote == False).order_by(Employee.year.asc())
from sqlalchemy import asc
employees = db_session.query(Employee).filter(Employee.is_remote == False).order_by(asc(Employee.year))
asc
という関数使う場合はimportして使うこともできます。
複数で並べ替え
yearで並べ替えたあとにidで並べ替えるとういう場合には単純に複数指定すればOK。
employees = db_session.query(Employee).filter(Employee.is_remote == False).order_by(Employee.year, Employee.id)
降順
降順の場合はdesc()
を使います。
例:is_remote Falseで勤続年数にて降順で並び替え
employees = db_session.query(Employee).filter(Employee.is_remote == False).order_by(Employee.year.desc())
↓実行結果
Satoさん 勤続10年
Takahashiさん 勤続3年
Tanakaさん 勤続2年
asc()
と同じようにfrom sqlalchemy import desc
にて並べ替えることも可能です。
employees = db_session.query(Employee).filter(Employee.is_remote == False).order_by(desc(Employee.year))
複数で並べ替える場合は昇順と同じように複数指定すればOKなのでコードは飛ばします。
他の並べ替え
これ以上はきりがないので参照だけ。他にもnullsfirst
とかnullslast
とかもあります。並べ替えたときにNULLを最初にするか最後にするかですかね。(使ったことないです)
他の物については公式を参照してください。
https://docs.sqlalchemy.org/en/13/core/sqlelement.html#sqlalchemy.sql.expression.nullsfirst
https://docs.sqlalchemy.org/en/13/core/sqlelement.html#sqlalchemy.sql.expression.nullslast
2.2.9 count:データ数取得
データの個数を取得できる。intで取得できます。
例:全従業員数を数える
ctn = db_session.query(Employee).count()
2.2.10 exists:存在するか?(True or Falseでとれる)
例:リモートワークの人いたら条件の中を処理
query = db_session.query(Employee).filter(Employee.is_remote == True)
if db_session.query(query.exists()).scalar():
print('リモートワークの人いますよ~')
else:
print('リモートワークの人いないよ')
db_session.query(query.exists())
だと<class 'sqlalchemy.orm.query.Query'>
になります。
SQL文は作ったけど、実行はしていない状態だと思えばいいと思います。
Pythonでこのexists()使うときはデータが存在するかBoolで取得したい場合がほとんどだと思います。なのでscalar()
というのを使うとboolで取得できるわけです。
蛇足ですがDjangoにもexists()あるから同じようにメソッドとして最後に付ければOKだろうとか思ったら、、、少し面倒でしたね。
存在しない場合True
~
をつけてやるだけです。~
つけると条件が反転することになっています。
query = db_session.query(Employee).filter(Employee.is_remote == True)
db_session.query(~query.exists()).scalar()
この例だとEmployee.is_remote == False
にすればいいだけなんですけど、「○○でない場合」で書いた方が分かりやすい場合もあるので使用する場合もあるかと思います。
2.2.11 limit, offset:取得データ個数指定
limit
使って最新のデータ3個だけ取得するとか書く場合に使います。
offset
は指定は必須でないです。ない場合.offset(0)
使ったのと同じSQLが発行されます。
例:最新のデータ3件を取得
employees = db_session.query(Employee).order_by(Employee.created_at.desc()).limit(3)
print_all(employees)
実行結果
Saitoさん 勤続5年
Satoさん 勤続10年
Takahashiさん 勤続3年
offset
offset(1)
みたいにつければいいだけです。
employees = db_session.query(Employee).order_by(Employee.created_at.desc()).limit(3).offset(1)
print_all(employees)
Satoさん 勤続10年
Takahashiさん 勤続3年
Tanakaさん 勤続2年
2.3 データ1つ取得したい
一番最初にget
は使いましたが、filterしてからデータを1つに絞りたい、データ1つ使いたいって時があります。とういうときに使う方法を紹介します。
レコード(1行分のデータ)をPythonで使える感じです。
2.3.1 one:データを1つだけ確実に取得したい場合
似たような機能でget()
がありましたが、こちらはfilter
書いた後にデータ1つだけなのを保証したいなーというときに使います。
- データが一つも取得できなかったらエラー
- データが2つ以上取得したらエラー
例:idが1のデータを取得
employee = db_session.query(Employee).filter(Employee.id == 1).one()
print(f'{employee.name}さん 勤続{employee.year}年')
結果
Tanakaさん 勤続2年
データが存在しない場合のエラー
これを存在していないidで取得すると
例えばemployee = db_session.query(Employee).filter(Employee.id == 100).one()
とか。
$ python db/main.py
.
.
.
sqlalchemy.exc.NoResultFound: No row was found when one was required
データがない場合にはNoResultFound
というエラーが出すことができます。
データが複数の場合エラー
今回登録したデータにはis_remote == False
は複数人います。なので以下を実行すると
employee = db_session.query(Employee).filter(Employee.is_remote == False).one()
print(f'{employee.name}さん 勤続{employee.year}年')
$ python db/main.py
.
.
.
sqlalchemy.exc.MultipleResultsFound: Multiple rows were found when exactly one was required
データが複数とれたのでMultipleResultsFound
というエラーが出すことができます。
エラー起こしたときの対応
データが取得できない場合、複数取れてしまった場合には処理を終了させたりログ取ったりできます。
いちいち、データがあるかとか個数をカウントするとかしなくてもわかりやすく処理もシンプルです。例えばこんなふうに使えます。
from sqlalchemy.exc import NoResultFound
try:
employee = db_session.query(Employee).filter(Employee.id == 100).one()
print(f'{employee.name}さん 勤続{employee.year}年')
except NoResultFound:
print('データないじゃん')
except MultipleResultsFound:
print('データ2個以上あるじゃん')
2.3.2 one_or_none()
ほぼone()
と同じです。違いは
- データが一つも取得できなかったらNone
- データが2つ以上取得したらエラー
ということで、データが取れなかったらエラーのかわりにNone
を返します。
注意は複数取れたら「エラー」ということ。複数取れたときもNoneだろうと思って処理書くとバグにつながります。
あとget()
と処理が近いです。
employee = db_session.query(Employee).filter(Employee.id == 100).one_or_none()
データが複数取れるとone()
と同じくMultipleResultsFound
が出ます。
2.3.3 first()
取得できたデータの中で最初のデータを1つ返す。
employee = db_session.query(Employee).filter(Employee.year > 3).first()
print(f'{employee.name}さん 勤続{employee.year}年')
勤続年数が3年以上で最初のデータを1つデータとして取得できます。
もし最後のデータ欲しい場合はorder_by()
とdesc()
使えばいいと思います。
一応気になったのでSQL見てみました。少し変えたけど以下みたいな感じ。
SELECT * FROM employee
WHERE employee.year > 3
LIMIT 1 OFFSET 0
つまりLIMIT
とOFFSET
使っても同じことできるけどシンプルに書けるということです。
補足:SQLAlchemy導入
導入方法を一応補足としてつけておきます。
DBはSQLiteを使おうと思います。MySQLやPostgreSQLなどで最初の設定が少し違いますがそれほど難しくはないはず。
あと、DBからデータ取得などの記法はDBの違いによって違いは基本的にないです。
$ pip install sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Boolean, DateTime
from datetime import datetime
engine = create_engine('sqlite:///sample.db', echo=True)
db_session = scoped_session(
sessionmaker(
autocommit=False,
autoflush=False,
bind=engine
)
)
Base = declarative_base()
Base.query = db_session.query_property()
class Employee(Base):
__tablename__ = 'employee'
id = Column(Integer, primary_key=True) # システムで使う番号
name = Column(String(255)) # 社員名
mail = Column(String(255)) # メール
is_remote = Column(Boolean) # リモート勤務しているか
department = Column(String(255)) # 部署
year = Column(Integer, default=0) # 社歴
created_at = Column(DateTime, nullable=False, default=datetime.now) # 作成日時
updated_at = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now) # 更新日時
if __name__ == "__main__":
Base.metadata.create_all(bind=engine)
実行したディレクトリでDBのファイルが作られるはず。class Employee(Base):
がテーブルの定義engine
というのがDBの設定データ入ってる。
Base.metadata.create_all(bind=engine)
の箇所でDB作成とテーブル作成を行っている。
これより前に定義や他ファイルから読み込みしてないとテーブル作成されないので注意。
実行してみます。
$ python sqlalchemy/alc_base.py
ディレクトリは以下のように作成してます。
.
└── db
├── main.py
├── Employee.py
└── setting.py # DB接続設定など
設定関係のファイル
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base
engine = create_engine('sqlite:///sample.db', echo=True)
db_session = scoped_session(
sessionmaker(
autocommit=False,
autoflush=False,
bind=engine
)
)
Base = declarative_base()
テーブルの定義などを書くモデルです。
from sqlalchemy import Column, Integer, String, Boolean, DateTime
from datetime import datetime
from setting import engine, Base
class Employee(Base):
__tablename__ = 'employee'
id = Column(Integer, primary_key=True) # システムで使う番号
name = Column(String(255)) # 社員名
mail = Column(String(255)) # メール
is_remote = Column(Boolean) # リモート勤務しているか
department = Column(String(255)) # 部署
year = Column(Integer, default=0) # 社歴
created_at = Column(DateTime, nullable=False, default=datetime.now) # 作成日時
updated_at = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now) # 更新日時
if __name__ == "__main__":
Base.metadata.create_all(bind=engine)
このファイルを実行するとBase.metadata.create_all(bind=engine)
が実行されるのでDBとテーブルが作成されます。
$ python db/Employee.py
.
.
.
CREATE TABLE employee (
id INTEGER NOT NULL,
name VARCHAR(255),
mail VARCHAR(255),
is_remote BOOLEAN,
department VARCHAR(255),
year INTEGER,
created_at DATETIME NOT NULL,
updated_at DATETIME NOT NULL,
PRIMARY KEY (id)
)
SQLの構文が出ています。モデルに書いた通りに作られているはず。
どのように作成されるかは、Employee.py
を変更したりして試してみても良いと思います。
SQLiteの場合、ファイルを削除すればDBの内容を消せますので、改めて作るときはsample.db
を削除して再度実行してみてください。
DBにデータを入れてみる
from Employee import *
from setting import db_session
employee = Employee(
name='Tanaka',
mail="aaa@aa.com",
is_remote=False,
department="develop",
year=2
)
db_session.add(employee)
db_session.commit()
固定値を入れてみました。
main.py
を実行してみると、SQL文が表示されて追加されているのがわかると思います。
$ python db/main.py
2021-12-21 01:34:55,783 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2021-12-21 01:34:55,785 INFO sqlalchemy.engine.Engine INSERT INTO employee (name, mail, is_remote, department, year, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)
2021-12-21 01:34:55,786 INFO sqlalchemy.engine.Engine [generated in 0.00054s] ('Tanaka', 'aaa@aa.com', 0, 'develop', 2, '2021-12-21 01:34:55.785500', '2021-12-21 01:34:55.785500')
2021-12-21 01:34:55,790 INFO sqlalchemy.engine.Engine COMMIT
Insertされていますね。
ついで、複数データをDBにいれる簡単なサンプル
def sample_add2(arr):
for data in arr:
employee = Employee(
name=data['name'],
mail=data['mail'],
is_remote=data['is_remote'],
department=data['department'],
year=data['year'],
)
db_session.add(employee)
db_session.commit()
if __name__ == "__main__":
data1 = {
'name': 'Takahashi',
'mail': 'bbb@aa.com',
'is_remote': False,
'department': 'sales',
'year': 3,
}
data2 = {
'name': 'Sato',
'mail': 'ccc@aa.com',
'is_remote': False,
'department': 'president',
'year': 10,
}
data3 = {
'name': 'Saito',
'mail': 'ddd@aa.com',
'is_remote': True,
'department': 'develop',
'year': 5,
}
sample_add2([data1, data2, data3])