SQLAlchemy1.2の基本的なクエリをまとめました！
python初心者以上向けの記事です。
改めて読み返してみると、直した方が良さそうな箇所や、この機能書かないんかいってのがあるので修正予定です。
記述内容
ORMについて、動かしてみるところまで、select, limit, orderby, distinct, join, leftjoin, in, insert, update, delete, union, unionAll, sql吐き出し, sqlを直接実行について
また、flask-sqlalchemyでは実行を試していません。
SQLAlchemyとは
pythonのORMモジュール。
session.query(User).all()
このようにSQL操作ができる。
ORM
SQLをクラスとして扱えるようにしたもの。
使用する理由は、
・SQLインジェクション対策がサポートされる。
・クラスなので、SQLをオブジェクト指向で書ける。
・引数に変数を入れるため、文字列の結合などが必要ないので短く書ける。
導入の注意点は、
・ちょっとしたSQLを実行する場合は、ORMを使わない方がコード量が短い。
・英語の文献が多いので学習コストが必要。
・慣れないうちはブラックボックスのコード。
ブラックボックスだから
「セキュリティ」や「実行速度」の面から考えると、SQLAlchemyの内部コードは読む必要あり。
どういうSQLが吐き出されてるかチェックすべき。
コメントで、どういうSQLが吐き出されてるか書いておいても良いと思う。
気づかないうちに超重いSQLを叩いてたりするから注意。
install
pip install sqlalchemy
まずsessionの作成(setting.py)
ファイルを実際に作成して説明します！
「DB接続を定義」と「session変数の作成」をします。
ファイル名 setting.py
from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base
# mysqlのDBの設定
DATABASE = 'mysql://%s:%s@%s/%s?charset=utf8' % (
"user_name",
"password",
"host_ip",
"db_name",
)
ENGINE = create_engine(
DATABASE,
encoding = "utf-8",
echo=True # Trueだと実行のたびにSQLが出力される
)
# Sessionの作成
session = scoped_session(
# ORM実行時の設定。自動コミットするか、自動反映するなど。
sessionmaker(
autocommit = False,
autoflush = False,
bind = ENGINE
)
)
# modelで使用する
Base = declarative_base()
Base.query = session.query_property()
モデルの作成(user.py)
DBにテーブルに対応するモデルを作成します！
setting.pyからimportで「BaseとENGINE」変数を呼び出している。
ファイル名user.py
import sys
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Float, DateTime
from setting import Base
from setting import ENGINE
class User(Base):
"""
ユーザモデル
"""
__tablename__ = 'users'
id = Column('id', Integer, primary_key = True)
name = Column('name', String(200))
age = Column('age', Integer)
email = Column('email', String(100))
def main(args):
"""
メイン関数
"""
Base.metadata.create_all(bind=ENGINE)
if __name__ == "__main__":
main(sys.argv)
テーブルの作成
上のスクリプトを実行するとテーブルが作成されます。
「Base.metadata.create_all(bind=ENGINE)」は、importしているモデル全てに対して、テーブルが存在しなかったら作成するという関数です。
python user.py
試しに実行
上で定義したsession.pyとuser.pyをimportします。
このファイルを実行すると「Userモデルのnameカラム」に「太郎」を追加され、
その後ユーザの一覧を表示しているので、実行結果には「太郎」と表示されます！
ファイル名main.py
# セッション変数の取得
from setting import session
# Userモデルの取得
from user import *
# DBにレコードの追加
user = User()
user.name = '太郎'
session.add(user)
session.commit()
# Userテーブルのnameカラムをすべて取得
users = session.query(User).all()
for user in users:
print(user.name)
SELECT（カラムを指定）
Userテーブルの指定したカラムの全レコードを取得。
下の例では、「User.name, User.email」のカラムのみをList型で取得する。
返り値：[User, ]
users = session.query(User.name, User.email).all()
# 取り出し例
for user in users:
print(user.name)
WHERE
Userテーブルのnameカラムが"name"のユーザ全てをList型で取得する。
返り値：[User, ]
※ pythonで処理の途中で改行する時には「\」を使う
users = session.query(User).\
filter(User.name=="name").\
all()
# 取り出し例
for user in users:
print(user.name, user.age)
LIMIT
Userテーブルから10人分をList型で取得する。
all()をつけないと、query型で取得してしまうため、
呼び出すごとにqueryが発行されて重くなる。
返り値：[User, ]
user_name = session.query(User).\
limit(10).\
all()
IN
Userテーブルでnamesに含まれるユーザ名をList型で取得する。
返り値：[User, ]
names = ['taro', 'jiro', 'ichiro']
brothers = session.query(User).\
filter(User.name.in_(names)).\
all()
INは遅くなるので下の記事参考
http://jetglass.hatenablog.jp/entry/2015/05/20/185336
ORDER BY
Userテーブルのユーザ全てをcreated_atが昇順にList型で取得する。
返り値：[User, ]
from sqlalchemy import desc
users = session.query(User).\
order_by(desc(User.created_at)).\
all()
Distinct
パターン1
from sqlalchemy import distinct
user_name = session.query(User).\
distinct(User.name).\
all()
パターン2
from sqlalchemy import distinct
user_name = session.query(distinct(User.name)).\
all()
JOIN
User.idと、UserSocial.user_idで内部結合し、
ユーザ全てをList型で取得する。
user_nameはこの形式で取得される。
返り値：[(User, UserSocial)]
user_name = session.query(User, UserSocial).\
join(UserSocial, User.id==UserSocial.user_id).\
all()
公式リファレンス
LEFT JOIN
User.idと、UserSocial.user_idで外部結合し、
ユーザ全てをList型で取得する。
user_nameはこの形式で取得される。
返り値：[(User, UserSocial)]
user_name = session.query(User, UserSocial).\
outerjoin(UserSocial, User.id==UserSocial.user_id).\
all()
公式リファレンス
UNION
TagとGenreを重複なしで結合するunionをし、List型で取得する。
返り値：[Tag,]
tag_genre = session.query(Tag).\
union(session.query(Genre)).\
all()
公式リファレンス
UNION ALL
TagとGenreを重複ありで結合するunion allをし、List型で取得する。
返り値：[Tag,]
tag_genre = session.query(Tag).\
union_all(session.query(Genre)).\
all()
公式リファレンス
吐き出されるSQLの取得
user_name = session.query(User)
sql_statement = user_name.statement
print(sql_statement)
INSERT
User.nameが'太郎'のユーザを追加する。
session.commit()でクエリ実行
user_name = User()
user_name.name = '太郎'
session.add(user_name)
session.commit()
UPDATE
User.idが1のユーザの名前を更新する。
user_name = User
session.commit()でクエリ実行
user_name = session.query(User).\
filter(User.id==1).\
first()
user_name.name = '太郎'
session.commit()
DELETE
User.idが1のユーザを全て削除する。
session.commit()でクエリ実行
session.query(User).\
filter(User.id==1).\
delete()
session.commit()
SQLを直接実行
sqlインジェクションを対策しなくて良い場合、
や自前で実装している場合、
直にSQLを叩くのもありです。
user_id = 1
sql = "select name from users where id = %s" % (user_id)
res = session.execute(sql)
# 取り出し例
for v in res:
print(v.name)
注意
.all()をつけないと、リスト型で取得できません。
下のコードではループのたびにクエリが発行されます。
※ 昔実行した時は、all()つけないと速度が遅くなった気がしてたが、コメントでご指摘いただいた通り、all()つけなくても良いのかも。
なので、両方試してみて欲しいです。
２年ぐらい前に書いたやつなので、手が空いてる時に全体的に修正しようと思います。
# allが付いていない
users = session.query(User).limit(100)
# 毎回クエリ発行
for user in users:
print(user.name)