ものすごく今さらな感じがするのですが、最近Pythonを書く機会が増えてきたので、有名どころのORM、SQLAlchemyを試してみようと思います。
DBはMySQLです。
インストール
SQLAlchemyを利用するには、それ自体のインストールと利用するドライバのインストールが必要になります。
今回はMySQLなので、mysqlclient をドライバとして利用します。
SQLAlchemy | Engine Configuration - MySQL
pip install mysqlclient SQLAlchemy
エンジン
SQLAlchemyはEngineと呼ばれるオブジェクトがコネクションプールとして働き、DBとのコネクションの中心となります。
Engineはグローバルなオブジェクトとして一度だけ定義します。
※ Webの場合、リクエストごとに作成せず、ワーカープロセスごとに1つだけ作成すると理解しました。
SQLAlchemy | Establishing Connectivity - the Engine
接続対象のURLフォーマットは dialect[+driver]://user:password@host/dbname[?key=value..] です。
SQLAlchemy | Engine Creation API - sqlalchemy.create_engine
from sqlalchemy import create_engine
DB_DIALECT="mysql"
DB_DRIVER="mysqldb" # mysqlclientならmysqldb, PyMySQLならpymysql
DB_USER="admin"
DB_PASSWD="adminadmin"
DB_HOST="192.168.1.53"
DB_PORT="3306"
DB_NAME="sqlalchemy_tutorial"
engine = create_engine(f'{DB_DIALECT}+{DB_DRIVER}://{DB_USER}:{DB_PASSWD}@{DB_HOST}/{DB_NAME}?charset=utf8mb4')
作成したエンジンでコネクションを取得して簡単なクエリを実行してみます。
from sqlalchemy import text
conn = engine.connect()
result = conn.execute(text("select 'hello world'"))
print(result.all())
conn.close()
セッション
セッションの作り方は大きく3つほどあるようです。
初期化
1. Sessionクラスで初期化する
一番簡単な方法で Session クラスを初期化してセッションを作成します。
当然ですが初期化ごとに新しいSessionオブジェクトができます。
SQLAlchemy | SessionBasic - Opening and Closing a Session
from sqlalchemy.orm import Session
session = Session(autocommit = False, autoflush = True, bind = engine)
session.add(some_object)
session.commit()
session.close()
2. sessionmaker を利用する
2つ目は sessionmaker というファクトリを利用する方法です。
こちらも初期化ごとに新しいSessionオブジェクトができます。
この方法を利用すると、セッション初期化時の引数( autocommit , autoflush など)を省略することができます。
SQLAlchemy | SessionBasic - Using a sessionmaker
from sqlalchemy.orm import sessionmaker
# Sessionを生成するためのファクトリーオブジェクトを生成
session_factory = sessionmaker(autocommit = False, autoflush = True, bind = engine)
# セッション初期化時に引数の指定がいらなくなる
session = session_factory()
session.add(some_object)
session.commit()
session.close()
3. scoped_sessionを利用する
Webでの利用を想定したセッションの生成方法で、1リクエストに1セッションをリンクさせます。
この方法では、 threading.local() を利用してセッションを管理し、セッション取得時は常に同じセッションオブジェクトを参照します。
SQLAlchemy | Contextual/Thread-local Sessions
from sqlalchemy.orm import scoped_session
from sqlalchemy.orm import sessionmaker
session_factory = sessionmaker(autocommit = False, autoflush = True, bind=engine)
Session = scoped_session(session_factory)
# セッションの取得
some_session = Session()
other_session = Session()
# some_sessionとother_sessionは同一のオブジェクト
some_session is other_session # True
begin / commit / roleback
begin / commit / rolebackの使い方はこんな感じです。
SQLAlchemy | Session Basics - Framing out a begin / commit / rollback block
with Session(engine) as session:
session.begin()
try:
session.add(some_object)
session.add(some_other_object)
except:
session.rollback()
raise
else:
session.commit()
テーブル定義(ORM)
SQLAlchemy | Defining Table Metadata with the ORM
SQLAlchemy | Data Manipulation with the ORM
ベースクラス
テーブル定義は declarative_base() によって生成されるベースクラスを継承して定義します。
SQLAlchemy | Setting up the Registry
from sqlalchemy.orm import declarative_base
Base = declarative_base()
テーブル定義
declarative_base() で生成した Base クラスを継承し、 __tablename__ にテーブル名を定義します。
カラムは Column クラスを利用して、プロパティとして定義します。
__repr__() を実装しておくとprintで表示できるようになるので、デバッグがはかどります。
SQLAlchemy | Declaring Mapped Classes
from sqlalchemy import Column, Integer, String
class User(Base):
__tablename__ = "user" # テーブル名
# カラムを定義
id = Column(Integer, primary_key=True)
name = Column(String(30), nullable=False, index=True, unique=True)
# strにキャストされたときのformat定義
def __repr__(self):
return f'<User({self.id}, {self.name})>'
リレーション
SQLAlchemy | Basic Relationship Patterns
1対多
1対多のリレーションでは、子テーブルが外部キーで親テーブルのidを参照します。
ORMでは relationship()を利用して、リレーション先のテーブルを指定することでリレーションを構築します。
双方向のリレーションを構築したい場合は back_populates パラメータで明示的に2つのテーブルを関連付けます。
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base, relationship
class User(Base):
__tablename__ = "user"
id = Column(Integer, primary_key=True)
name = Column(String(30), nullable=False, index=True, unique=True)
# リレーション: Fileクラスのuserプロパティに関連付ける
files = relationship("File", back_populates="user")
class File(Base):
__tablename__ = 'file'
id = Column(Integer, primary_key=True)
name = Column(String(30), nullable=False, index=True, unique=True)
# 外部キー
user_id = Column(Integer, ForeignKey('user.id')) # ForeignKeyには "テーブル名.カラム名" を指定
# Userクラスのfilesプロパティに関連付ける
user = relationship("User", back_populates="files")
多対多
多対多のリレーションでは、中間テーブルを利用して要素を関連付けます。
relation() を利用してリレーションを構築するのは、1対多の時と変わりませんが、多対多では secondary パラメータに中間テーブルを指定します。
class UserRoles(Base):
"""userとroleの中間テーブル"""
__tablename__ = "user_roles"
user_id = Column(Integer, ForeignKey("user.id"), nullable=False)
role_id = Column(Integer, ForeignKey("role.id"), nullable=False)
class User(Base):
__tablename__ = "user"
id = Column(Integer, primary_key=True)
name = Column(String(30), nullable=False, index=True, unique=True)
# リレーション: Roleクラスのusersプロパティに、中間テーブルuser_rolesを介して関連付ける
roles = relationship("Role", secondary=UserRoles.__tablename__, back_populates="users")
class Role(Base):
__tablename__ = 'role'
id = Column(Integer, primary_key=True)
name = Column(String(30), nullable=False, index=True, unique=True)
# リレーション: Userクラスのrolesプロパティに、中間テーブルuser_rolesを介して関連付ける
users = relationship("User", secondary=UserRoles.__tablename__, back_populates="roles")
テーブル操作(ORM)
SQLAlchemy | Data Manipulation with the ORM
利用するテーブル定義はこちら
from sqlalchemy.orm import declarative_base
Base = declarative_base()
class UserRoles(Base):
__tablename__ = "user_roles"
user_id = Column(Integer, ForeignKey("user.id"), nullable=False)
role_id = Column(Integer, ForeignKey("role.id"), nullable=False)
class User(Base):
__tablename__ = "user"
id = Column(Integer, primary_key=True)
name = Column(String(30), nullable=False, index=True, unique=True)
files = relationship("File", back_populates="user")
roles = relationship("Role", secondary=UserRoles.__tablename__, back_populates="users")
class Role(Base):
__tablename__ = 'role'
id = Column(Integer, primary_key=True)
name = Column(String(30), nullable=False, index=True, unique=True)
users = relationship("User", secondary=UserRoles.__tablename__, back_populates="roles")
class File(Base):
__tablename__ = 'file'
id = Column(Integer, primary_key=True)
user_id = Column(Integer, ForeignKey('user.id'))
name = Column(String(30), nullable=False, index=True, unique=True)
user = relationship("User", back_populates="files")
CREATE TABLE
Base.metadata.create_all(engine)
DROP TABLE
Base.metadata.drop_all(engine)
INSERT, UPDATE
SQLAlchemy | Inserting Rows with the ORM
SQLAlchemy | Updating ORM Objects
多対多のリレーションを構築したuserとroleのinsert。
User に Role を紐づけて add すると、中間テーブルへのinsertまで良しなにやってくれます。
with Session() as session:
role = Role()
role.name = "AdminRole"
user = User()
user.name = "admin"
user.roles.append(role)
session.add(user)
session.commit()
1対多のリレーションを構築したfile(多)のinsert
User.files に File を追加して User を保存するやり方が使いやすそうです。
# (その1): user.filesにFile追加してUserをadd
with Session() as session:
file = File()
file.name = "test_01.txt"
stmt = select(User).filter(User.name == "admin").order_by(User.id)
user = session.execute(stmt).scalar_one()
user.files.append(file)
session.add(user)
session.commit()
# (その2): user.idをFile.user_idに指定して、Fileをadd
with Session() as session:
stmt = select(User).filter(User.name == "admin").order_by(User.id)
user = session.execute(stmt).scalar_one()
file = File()
file.name = "test_02.txt"
terminology_file.user_id = user.id
session.add(file)
session.commit()
roleの追加とuserとの紐づけ
User.roles に Role を追加して保存すると、中間テーブルの更新まで良しなにやってくれます。
with Session() as session:
# roleの追加
role = Role()
role.name = rolename
session.add(role)
session.commit()
# userとroleの紐づけ
stmt = select(User).where(User.name == user_name)
user = session.execute(stmt).scalar_one()
user.roles.append(role)
session.add(user)
session.commit()
SELECT
select文はこんな感じ
with Session() as session:
stmt = select(User).filter(User.name == "admin").order_by(User.id)
result = session.execute(stmt).all()
pprint(result)
DELETE
SQLAlchemy | Deleting ORM Objects
User を delete すると、 user_roles 中間テーブルの関連レコードは削除されます。
file テーブルの関連レコードは削除されず、 user_id が NULL になります。
with Session() as session:
stmt = select(User).filter(User.name == "admin").order_by(User.id)
user = session.execute(stmt).scalar_one()
session.delete(user)
session.commit()