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

概要

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

 

前提

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

あわせて読みたい

概要 PythonでSQLite3を使用するにあたり、SQLite3の環境構築を実施したので方法についてまとめた。SQLite3実行ファイルをダウンロードして環境変数のPATHを通すところまで行っている。   […]

あわせて読みたい

概要 SQLite3のデータベースファイルを作成し、テーブル操作を行ったのでその方法についてまとめた。   前提 前提として、SQLite3の環境構築をしていること。[sitecar[…]

 

 

テーブル定義

以下のitemテーブルに対して操作を行う。


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テーブル

id name memo price category_id
1 アクエリアス スポーツドリング 120 1
2 ハンバーガー マクドナルド 80 2
3 爪切り null 1100 3

 

item_categoryテーブル

id name memo
1 飲料水 飲み物
2 食品 食べ物
3 生活用品 生活用品

 

 

sqlalchemyモジュールの使い方

特徴

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

・自動でSQL文を作成して、ORマッピング(クラスとテーブルの紐づけ)してくれる。
・セッションを介してクエリを実行する。
・データベースごとに違うSQL文の違いなどを意識する必要がなくなる。(SQL文を記述しないため)

ポイントとしては、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」とインポート文に記述した場合、以下のようなエラーが出た。

MovedIn20Warning: The “declarative_base()“ function is now available as sqlalchemy.orm.declarative_base(). (deprecated since: 2.0) (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
Base = 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クラスを作成している。
また、属性を参照するためのプロパティメソッドを作成している。

あわせて読みたい

概要 Pythonのクラス定義方法、属性とメソッド、インスタンス操作などをまとめた。   クラスの定義 Pythonでクラスを定義するには、以下のようにclassキーワードを使用する[…]

 

③エンジンの作成

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

sqlite:///データベースファイルまでのパス

 


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)」と記述しても検索はできるが、以下の警告が表示された。

The Query.get() method is considered legacy as of the 1.x series of SQLAlchemy and becomes a legacy construct in 2.0. The method is now available as Session.get() (deprecated since: 2.0)

 

WHERE

filter_byを使用する。
first()は最初のデータを返却するメソッド。


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

 

AND

filter_byの条件をカンマ区切りにすればAND条件になる。


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

 

比較

filterを使用する。


# 比較(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

 


# 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

 


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

 

OR

OR条件を使用する場合、or_()をインポートする必要あり。


# 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

 


# 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

昇順

priceの昇順にデータリストを取得する。


# 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

 

降順

priceの降順にデータリストを取得する。


# 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

 

作成(INSERT)

以下のようにデータの挿入を行う。


# 新規追加
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クラスには__init__関数を定義していないが、コンストラクタに引数を指定して定義できた。

更新(UPDATE)

以下のようにデータの更新を行う。


# 更新
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

 

削除(DELETE)

以下のように削除する。


# 削除
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()

 

結合

join

item_categoryとitemテーブルを内部結合する。


# テーブル結合
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はあるモデルクラスと別のモデルクラスとの間に関係(リレーションシップ)を定義する。
これにより、関連するレコードを簡単に取得できる。

使い方
relationship(①関連するクラス名, ②back_populates または backref)

 

関連するクラス名:

関連するクラス名を設定する。
例えば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")

 

トランザクション制御

SQLAlchemyの初期設定では自動コミットはしない。
最初のクエリ呼び出しでトランザクションが開始される。


# トランザクション制御
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

 

スポンサーリンク