【sqlalchemy】PythonでDB操作：SQLAlchemyの基本的な使用方法

概要

sqlalchemyモジュールを使用してDBを操作する方法についてまとめた。
基本的なCRUD機能の記述方法について扱う。

前提

SQLiteのDB環境を構築していること。

CREATE TABLE item (
        id INTEGER PRIMARY KEY,
        name TEXT,
        memo TEXT,
        price INTEGER,
        category_id INTEGER,
        FOREIGN KEY(category_id) REFERENCES item_category(id)
    );
CREATE TABLE item_category (
        id INTEGER PRIMARY KEY,
        name TEXT,
        memo TEXT
    );

テーブルの中身

予め、それぞれのテーブルに以下のデータを用意しておく。

itemテーブル

item_categoryテーブル

sqlalchemyモジュールの使い方

特徴

sqlalchemyには以下の特徴がある。

ポイントとしては、sqlite3モジュールと比較するとSQL文を記述する必要がないこと。

DB操作の流れ

全体像

# ①モジュールの読み込み
from sqlalchemy.orm import declarative_base
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy import Column, Integer, String

# ②ベースモデルを作成
Base = declarative_base()
class Item(Base):
    """ itemクラス """
    # テーブル名称
    __tablename__ = 'item'
    # カラム
    id = Column(Integer, primary_key=True)
    name = Column(String)
    price = Column(Integer)
    memo = Column(String)
    category_id = Column(Integer)

    @property
    def item_info(self):
        """ itemクラスの属性を整形して返却 """
        return f"id: {self.id}, 商品名: {self.name}, 値段: {self.price}円, メモ: {self.memo}, カテゴリid: {self.category_id}"

def get_session():
    """ セッション情報を返却 """
    # ③エンジンの作成
    engine = create_engine('sqlite:///db/test.db')
    # ④セッションオブジェクトの作成
    # DBと通信するセッションオブジェクトの作成
    # sqlite3におけるconnectionオブジェクトに近い
    Session = sessionmaker(bind=engine)
    session = Session()

    return session

def main():
    """ メイン実行関数 """
    session = get_session()
    # ⑤テーブル操作
    items = session.query(Item).all()
    for item in items:
        print(item.item_info)

if __name__ == "__main__":
    main()
# --------------------------------------------------------
# 出力
#id: 1, 商品名: アクエリアス, 値段: 120円, メモ: スポーツドリンク, カテゴリid: 1
#id: 2, 商品名: ハンバーガー, 値段: 80円, メモ: マクドナルド, カテゴリid: 2
#id: 3, 商品名: 爪切り, 値段: 1100円, メモ: , カテゴリid: 3

①モジュールの読み込み

モジュールのインストール

sqlalchemyモジュールはimportするだけだと使用できない。
モジュール自体をインストールする必要があるため、以下のようにインストールする。
※jupyter labの場合はセルに記述してShift+Enter

pip install sqlalchemy

モジュールのインポート

以下のようにモジュールのインポート文を記述する。

from sqlalchemy.orm import declarative_base
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy import Column, Integer, String

尚、declarative_baseについては「from sqlalchemy.ext.declarative import declarative_base」とインポート文に記述した場合、以下のようなエラーが出た。

これはdeclarative_baseのインポートを「from sqlalchemy.orm import declarative_base」に変更すれば解決した。

②ベースモデルの作成

テーブルのレコードに紐づくクラスを作成する。
属性はテーブルのカラムに合わせること。
尚、テーブルに紐づくクラスであることを示すBaseクラスを継承する必要がある。

# ベースモデルを作成
Base = declarative_base()
class Item(Base):
    """ itemクラス """
    # テーブル名称
    __tablename__ = 'item'
    # カラム
    id = Column(Integer, primary_key=True)
    name = Column(String)
    price = Column(Integer)
    memo = Column(String)
    category_id = Column(Integer)

    @property
    def item_info(self):
        """ itemクラスの属性を整形して返却 """
        return f"id: {self.id}, 商品名: {self.name}, 値段: {self.price}円, メモ: {self.memo}, カテゴリid: {self.category_id}"

itemテーブルに紐づくitemクラスを作成している。
また、属性を参照するためのプロパティメソッドを作成している。

③エンジンの作成

SQLiteで作成したDBを以下の形式で指定して読み込む。

engine = create_engine('sqlite:///db/test.db')

④セッションオブジェクトの作成

テーブル操作を行うためのセッションオブジェクトを作成する。

def get_session():
    """ セッション情報を返却 """
    # エンジンを作成
    engine = create_engine('sqlite:///db/test.db')
    # DBと通信するセッションオブジェクトの作成
    # sqlite3におけるconnectionオブジェクトに近い
    Session = sessionmaker(bind=engine)
    session = Session()

    return session

⑤テーブル操作

セッションオブジェクトを取得して、テーブル操作を行う。

def main():
    """ メイン実行関数 """
    session = get_session()
    # itemテーブルの中身をすべて取得
    items = session.query(Item).all()
    for item in items:
        print(item.item_info)

if __name__ == "__main__":
    main()

SQL文を作成せず、sessonオブジェクトのメソッドを使用してテーブル操作を行う。
all()メソッドはテーブルの中身をすべてリストで返却する。

尚、all()メソッドをつけない場合Queryオブジェクトが返却されるが、同じようにfor文でループ処理が可能。

様々なテーブル操作

参照（READ）

プライマリキー検索

getを使用する。

# プライマリキー検索
item = session.get(Item,2)
print(item.item_info)

# -------------------------------------- 
# 出力 
#id: 2, 商品名: ハンバーガー, 値段: 80円, メモ: マクドナルド, カテゴリid: 2

「#item = session.query(Item).get(2)」と記述しても検索はできるが、以下の警告が表示された。

# WHERE（id検索）
item = session.query(Item).filter_by(id=2).first()
print(item.item_info)
# --------------------------------------
# 出力
#id: 2, 商品名: ハンバーガー, 値段: 80円, メモ: マクドナルド, カテゴリid: 2

# AND（id、name）
item = session.query(Item).filter_by(id=2, name="ハンバーガー").first()
print(item.item_info)
# --------------------------------------
# 出力
#id: 2, 商品名: ハンバーガー, 値段: 80円, メモ: マクドナルド, カテゴリid: 2

# 比較（price>100）
items = session.query(Item).filter(Item.price>100)
for item in items:
    print(item.item_info)
------------------------------------
# 出力
#id: 1, 商品名: アクエリアス, 値段: 120円, メモ: スポーツドリンク, カテゴリid: 1
#id: 3, 商品名: 爪切り, 値段: 1100円, メモ: , カテゴリid: 3

# IN
items = session.query(Item).filter(Item.price.in_([80, 120, 150]))
for item in items:
    print(item.item_info)
------------------------------------
# 出力
#id: 1, 商品名: アクエリアス, 値段: 120円, メモ: スポーツドリンク, カテゴリid: 1
#id: 2, 商品名: ハンバーガー, 値段: 80円, メモ: マクドナルド, カテゴリid: 2

# LIKE
item = session.query(Item).filter(Item.name.like("%エリ%")).first()
print(item.item_info)
------------------------------------
# 出力
#id: 1, 商品名: アクエリアス, 値段: 120円, メモ: スポーツドリンク, カテゴリid: 1

# or_をインポート
from sqlalchemy import Column, Integer, String, or_

# OR
items = session.query(Item).filter(or_(Item.id==2, Item.id==3))
for item in items:
    print(item.item_info)
------------------------------------
# 出力
#id: 2, 商品名: ハンバーガー, 値段: 80円, メモ: マクドナルド, カテゴリid: 2
#id: 3, 商品名: 爪切り, 値段: 1100円, メモ: , カテゴリid: 3

# BETWEEN
items = session.query(Item).filter(Item.price.between(50, 150))
for item in items:
    print(item.item_info)
------------------------------------
# 出力
#id: 1, 商品名: アクエリアス, 値段: 120円, メモ: スポーツドリンク, カテゴリid: 1
#id: 2, 商品名: ハンバーガー, 値段: 80円, メモ: マクドナルド, カテゴリid: 2

# ORDER_BY（昇順）
items = session.query(Item).order_by(Item.price.asc())
for item in items:
    print(item.item_info)
--------------------------------
# 出力
#id: 2, 商品名: ハンバーガー, 値段: 80円, メモ: マクドナルド, カテゴリid: 2
#id: 1, 商品名: アクエリアス, 値段: 120円, メモ: スポーツドリンク, カテゴリid: 1
#id: 3, 商品名: 爪切り, 値段: 1100円, メモ: , カテゴリid: 3

# ORDER_BY（降順）
items = session.query(Item).order_by(Item.price.desc())
for item in items:
    print(item.item_info)

--------------------------------
# 出力
#id: 3, 商品名: 爪切り, 値段: 1100円, メモ: , カテゴリid: 3
#id: 1, 商品名: アクエリアス, 値段: 120円, メモ: スポーツドリンク, カテゴリid: 1
#id: 2, 商品名: ハンバーガー, 値段: 80円, メモ: マクドナルド, カテゴリid: 2

# 新規追加
item = Item()
item.name = "コカ・コーラ"
item.price = 160
item.category_id = 1
session.add(item)
session.commit()

# 確認
items = session.query(Item).all()
for item in items:
    print(item.item_info)
--------------------------------
# 出力
#id: 1, 商品名: アクエリアス, 値段: 120円, メモ: スポーツドリンク, カテゴリid: 1
#id: 2, 商品名: ハンバーガー, 値段: 80円, メモ: マクドナルド, カテゴリid: 2
#id: 3, 商品名: 爪切り, 値段: 1100円, メモ: , カテゴリid: 3
#id: 4, 商品名: コカ・コーラ, 値段: 160円, メモ: None, カテゴリid: 1

# 全件削除
session.query(Item).delete()
# 複数挿入
data = [
Item(id=1, name="サイダー", memo="", price=140, category_id=1),
Item(id=2, name="お茶", memo="", price=90, category_id=1),
Item(id=3, name="ポテトチップス", memo="", price=190, category_id=2)
]
session.add_all(data)
session.commit()

# 確認
items = session.query(Item)
for item in items:
print(item.item_info)
--------------------------------
# 出力
#id: 1, 商品名: サイダー, 値段: 140円, メモ: , カテゴリid: 1
#id: 2, 商品名: お茶, 値段: 90円, メモ: , カテゴリid: 1
#id: 3, 商品名: ポテトチップス, 値段: 190円, メモ: , カテゴリid: 2

# 更新
item = session.query(Item).filter_by(id=3).first()
item.price = 250
session.add(item)
session.commit()

# 確認
items = session.query(Item)
for item in items:
    print(item.item_info)
--------------------------------
# 出力
#id: 1, 商品名: サイダー, 値段: 140円, メモ: , カテゴリid: 1
#id: 2, 商品名: お茶, 値段: 90円, メモ: , カテゴリid: 1
#id: 3, 商品名: ポテトチップス, 値段: 250円, メモ: , カテゴリid: 2

# 削除
session.query(Item).filter_by(id=2).delete()
session.commit()

# 確認
items = session.query(Item)
for item in items:
    print(item.item_info)
--------------------------------
# 出力
#id: 1, 商品名: サイダー, 値段: 140円, メモ: , カテゴリid: 1
#id: 3, 商品名: ポテトチップス, 値段: 250円, メモ: , カテゴリid: 2

# 削除
item = session.query(Item).filter_by(id=1).first() 
session.delete(item) 
session.commit()

# 確認
items = session.query(Item)
for item in items:
    print(item.item_info)
--------------------------------
# 出力
#id: 3, 商品名: ポテトチップス, 値段: 250円, メモ: , カテゴリid: 2

# 全件削除削除
session.query(Item).delete()
session.commit()

# テーブル結合
res = session.query(Item.id, ItemCategory.name.label("category"), Item.name).join(ItemCategory, Item.category_id == ItemCategory.id).limit(2)
for row in res:
    print(f"商品ID：{row.id}, カテゴリ：{row.category}, 商品名：{row.name}")
# --------------------------------------------------------
# 出力
#商品ID：1, カテゴリ：飲料水, 商品名：アクエリアス
#商品ID：2, カテゴリ：食品, 商品名：ハンバーガー

relationship

relationshipはあるモデルクラスと別のモデルクラスとの間に関係（リレーションシップ）を定義する。
これにより、関連するレコードを簡単に取得できる。

使い方

①関連するクラス名：

関連するクラス名を設定する。
例えばItemCategoryクラスからすると、Itemクラスが関連するクラスになる。

②back_populates または backref：

back_populatesまたはbackrefは設定しない場合、一方向から参照できる。

back_populatesまたはbackrefの目的は同じでリレーションシップが双方向になる設定。
記載方法が異なるだけ。

一方向の参照

from sqlalchemy.orm import declarative_base
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship

Base = declarative_base()
class Item(Base):
    """ itemクラス """
    # テーブル名称
    __tablename__ = 'item'
    # カラム
    id = Column(Integer, primary_key=True)
    name = Column(String)
    price = Column(Integer)
    memo = Column(String)
    category_id = Column(Integer, ForeignKey('item_category.id')) 
    

class ItemCategory(Base):
    """ item_categoryクラス """
    # テーブル名称
    __tablename__ = "item_category"
    # カラム
    id = Column(Integer, primary_key=True)
    name = Column(String)
    memo = Column(String)
    items = relationship("Item") 


def get_session():
    """ セッション情報を返却 """
    engine = create_engine('sqlite:///db/test.db')
    # DBと通信するセッションオブジェクトの作成
    # sqlite3におけるconnectionオブジェクトに近い
    Session = sessionmaker(bind=engine)
    session = Session()

    return session

def main():
    """ メイン実行関数 """
    session = get_session()
    # テーブル操作
    categories = session.query(ItemCategory).all()
    for category in categories:
        for item in category.items:
            print(f"商品ID：{item.id}, カテゴリ：{category.name}, 商品名：{item.name}")

if __name__ == "__main__":
    main()
# --------------------------------------------------------
# 出力
#商品ID：1, カテゴリ：飲料水, 商品名：アクエリアス
#商品ID：2, カテゴリ：食品, 商品名：ハンバーガー
#商品ID：3, カテゴリ：生活用品, 商品名：爪切り

joinを記述しなくとも、シンプルに親子関係のデータを取得できる。
ItemCategoryを取得したときのみ、Itemを参照できる一方向のケース。

双方向の参照

◆back_populatesを使用する場合

from sqlalchemy.orm import declarative_base
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship

Base = declarative_base()
class Item(Base):
    """ itemクラス """
    # テーブル名称
    __tablename__ = 'item'
    # カラム
    id = Column(Integer, primary_key=True)
    name = Column(String)
    price = Column(Integer)
    memo = Column(String)
    category_id = Column(Integer, ForeignKey('item_category.id')) 
    category = relationship("ItemCategory", back_populates="items") 
    

class ItemCategory(Base):
    """ item_categoryクラス """
    # テーブル名称
    __tablename__ = "item_category"
    # カラム
    id = Column(Integer, primary_key=True)
    name = Column(String)
    memo = Column(String)
    items = relationship("Item", back_populates="category") 


def get_session():
    """ セッション情報を返却 """
    engine = create_engine('sqlite:///db/test.db')
    # DBと通信するセッションオブジェクトの作成
    # sqlite3におけるconnectionオブジェクトに近い
    Session = sessionmaker(bind=engine)
    session = Session()

    return session

def main():
    """ メイン実行関数 """
    session = get_session()
    # テーブル操作
    print("ItemCategoryからItemを参照")
    categories = session.query(ItemCategory).all()
    for category in categories:
        for item in category.items:
            print(f"商品ID：{item.id}, カテゴリ：{category.name}, 商品名：{item.name}")

    print("ItemからItemCategoryを参照")
    items = session.query(Item).all()
    for item in items:
        print(f"商品ID：{item.id}, カテゴリ：{item.category.name}, 商品名：{item.name}")

if __name__ == "__main__":
    main()
# --------------------------------------------------------
# 出力
#ItemCategoryからItemを参照
#商品ID：1, カテゴリ：飲料水, 商品名：アクエリアス
#商品ID：2, カテゴリ：食品, 商品名：ハンバーガー
#商品ID：3, カテゴリ：生活用品, 商品名：爪切り
#ItemからItemCategoryを参照
#商品ID：1, カテゴリ：飲料水, 商品名：アクエリアス
#商品ID：2, カテゴリ：食品, 商品名：ハンバーガー
#商品ID：3, カテゴリ：生活用品, 商品名：爪切り

back_populatesを使用する際には、お互いのクラスにそれぞれ相手のプロパティを定義する必要がある。
◆backrefを使用する場合

クラスの定義する方法が異なるが、上記と同じ結果になる。

class Item(Base):
    """ itemクラス """
    # テーブル名称
    __tablename__ = 'item'
    # カラム
    id = Column(Integer, primary_key=True)
    name = Column(String)
    price = Column(Integer)
    memo = Column(String)
    category_id = Column(Integer, ForeignKey('item_category.id'))

class ItemCategory(Base):
    """ item_categoryクラス """
    # テーブル名称
    __tablename__ = "item_category"
    # カラム
    id = Column(Integer, primary_key=True)
    name = Column(String)
    memo = Column(String)
    items = relationship("Item", backref="category")

# トランザクション制御
try:
    item = session.get(Item,1)
    item.name = "ジンジャーエール"
    session.add(item)
    raise ValueError("障害発生")
    session.commit()
except ValueError as e:
    print(e.args[0], "ロールバック")
    session.rollback()
finally:
    session.close()
item = session.get(Item, 1)
print(item.item_info)

# --------------------------------------------------------
# 出力
#障害発生 ロールバック
#id: 1, 商品名: アクエリアス, 値段: 120円, メモ: スポーツドリンク, カテゴリid: 1