SQLAlchemyの使い方

2022/11/26

PythonのORMとして「SQLAlchemy」が代表的です。ここでは、SQLAlchemyを利用して、基本的なORM操作（INSERT, SELECT, UPDATE, DELETE）を確認します。

パッケージインストール

「sqlalchemy」と「DB接続するためのドライバ」をインストールします。
ここでは、DBにmysqlを利用するので、ドライバとして mysqlclient をインストールします。

pipenv install sqlalchemy
pipenv install mysqlclient

ファイル構成

以下ファイル構成で実装していきます。

.
└── sql
    ├── main.py          # ここで、ORM操作をします。
    ├── models
    │   └── user.py      # userテーブルの定義をします。
    └── setting
        └── setting.py   # Engineインスタンス, session生成など行います。

DB接続に必要な設定

DB接続に必要な設定を行います。他のファイルから利用します。

以下、 setting.py に記述します。

from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base

dialect = "mysql"
driver = "mysqldb"
username = "root"
password = "xxxxxxxx"
host = "192.168.30.30"
port = "3306"
database = "sample_db"
charset_type = "utf8"
db_url = f"{dialect}+{driver}://{username}:{password}@{host}:{port}/{database}?charset={charset_type}"

# DB接続するためのEngineインスタンス
ENGINE = create_engine(db_url, echo=True)

# DBに対してORM操作するときに利用
# Sessionを通じて操作を行う
session = scoped_session(
    sessionmaker(autocommit=False, autoflush=False, bind=ENGINE)
)

# 各modelで利用
# classとDBをMapping
Base = declarative_base()

補足

create_engine
- echo=True とすることで、SQLを出力します。
- https://docs.sqlalchemy.org/en/latest/core/engines.html#sqlalchemy.create_engine
sessionmaker
- https://docs.sqlalchemy.org/en/latest/orm/session_api.html?highlight=sessionmaker#sqlalchemy.orm.session.sessionmaker
- https://docs.sqlalchemy.org/en/latest/orm/tutorial.html#creating-a-session
declarative_base
- https://docs.sqlalchemy.org/en/latest/orm/extensions/declarative/api.html?highlight=declarative_base#sqlalchemy.ext.declarative.declarative_base

テーブル作成

user.py にてusersテーブルを定義します。
さきほど作成した setting.py を利用します。

from sqlalchemy import Column, Integer, String, DateTime, Sequence
from sql.setting.setting import ENGINE, Base
from datetime import datetime
import sys


class User(Base):
    """
    UserModel
    """
    __tablename__ = 'users'
    id = Column(Integer, Sequence('user_id_seq'), primary_key=True)
    name = Column(String(50))
    email = Column(String(255))
    age = Column(Integer)
    created_at = Column('created', DateTime, default=datetime.now, nullable=False)
    updated_at = Column('modified', DateTime, default=datetime.now, nullable=False)


def main(args):
    Base.metadata.create_all(bind=ENGINE)


if __name__ == "__main__":
    main(sys.argv)

実行すると、下記クエリが実行されました。

CREATE TABLE users (
	id INTEGER NOT NULL AUTO_INCREMENT, 
	name VARCHAR(50), 
	email VARCHAR(255), 
	age INTEGER, 
	created DATETIME NOT NULL, 
	modified DATETIME NOT NULL, 
	PRIMARY KEY (id)
)

ORM操作

main.py にて、userモデルと設定情報を読み込み、ORM操作を記述していきます。

from sql.models.user import *
from sql.setting.setting import session

INSERT

user = User()
user.name = 'yamada'
user.email = 'xxx@xxx.com'
user.age = 32
session.add(user)
session.commit()

INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
INFO sqlalchemy.engine.base.Engine INSERT INTO users (name, email, age, created, modified) VALUES (%s, %s, %s, %s, %s)
INFO sqlalchemy.engine.base.Engine ('yamada', 'xxx@xxx.com', 32, datetime.datetime(2018, 8, 21, 7, 21, 8, 941790), datetime.datetime(2018, 8, 21, 7, 21, 8, 941814))
INFO sqlalchemy.engine.base.Engine COMMIT

SELECT

users = session.query(User).all()
for user in users:
    print(f'{user.name} {user.email} {user.age}')

INFO sqlalchemy.engine.base.Engine SELECT users.created AS users_created, users.modified AS users_modified, users.id AS users_id, users.name AS users_name, users.email AS users_email, users.age AS users_age 
FROM users
yamada xxx@xxx.com 32

WHERE

users = session.query(User). \
    filter(User.age == 32). \
    all()
for user in users:
    print(f'{user.name} {user.email} {user.age}')

INFO sqlalchemy.engine.base.Engine SELECT users.created AS users_created, users.modified AS users_modified, users.id AS users_id, users.name AS users_name, users.email AS users_email, users.age AS users_age 
FROM users 
WHERE users.age = %s
yamada xxx@xxx.com 32

UPDATE

user = session.query(User). \
    filter(User.name == 'yamada'). \
    first()
user.age = 36
session.commit()

INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
INFO sqlalchemy.engine.base.Engine SELECT users.created AS users_created, users.modified AS users_modified, users.id AS users_id, users.name AS users_name, users.email AS users_email, users.age AS users_age 
FROM users 
WHERE users.name = %s 
 LIMIT %s
INFO sqlalchemy.engine.base.Engine ('yamada', 1)
INFO sqlalchemy.engine.base.Engine UPDATE users SET age=%s WHERE users.id = %s
INFO sqlalchemy.engine.base.Engine (36, 1)
INFO sqlalchemy.engine.base.Engine COMMIT

DELETE

session.query(User). \
    filter(User.age > 10).delete()
session.commit()

INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
INFO sqlalchemy.engine.base.Engine DELETE FROM users WHERE users.age > %s
INFO sqlalchemy.engine.base.Engine (10,)
INFO sqlalchemy.engine.base.Engine COMMIT