MENU
SQLAlchemy

【Python】SQLAlchemyの基本的な使い方

【Python】SQLAlchemyの基本的な使い方
naoki-hn

Python のクロスデータベースライブラリの中でよく知られている SQLAlchemy について基本的な使い方を解説します。

SQLAlchemy

Python では、SQLite やMySQL、PostrgreSQL等を使用するための各種ライブラリが用意されています。ライブラリは、それぞれのデータベース個々の使い方（方言のようなもの）を理解した上で使う必要があります。

SQLAlchemy は、DB の使い方の差異を吸収してくれるクロスデータベースライブラリとして支持を集めているライブラリです。Python 標準ライブラリではありませんが多くの人に利用されています。

SQLAlchemy は「エンジンレイヤ」「SQL表現言語」「ORM」といった複数レベルでの使用ができるようになっています。

この記事では、SQLAlchemy の各レベルでの使い方を例を使いながら紹介します。

SQLAlchemy の使い方

SQLAlchemy は、データベースに対して複数のレベルで使用することができます。

レベル概要
エンジンレイヤ最も低レベルの SQL でデータベースへのアクセスを行う。
SQL 表現言語Python の式として、SQL 操作の関数を使用することにより、データベース間の SQL 表現の違いを吸収してデータベースへアクセスを行う。エンジンレイヤと ORM のちょうど中間に位置する表現方法となる。
オブジェクト関係マッピング
(ORM: Object Relation Mapping）		Python クラスのオブジェクトを定義し、オブジェクトとデータベース構造をマッピングすることによりデータベースアクセスを行う。

以降で各レベルでの使用方法を例を用いて紹介していきます。

注意点

SQLAlchemy のバージョンについては記事執筆時点の 2.0.6 を前提として動作確認をしています。バージョンにより使用方法が変わっている可能性もありますので、ご注意ください。

各種データベースへの接続表現文字列

SQLAlchemy は、様々なデータベースの違いを吸収してくれますが、具体的には各データベースのドライバを使用します。そのため、SQLAlchemy を使用する際には、ドライバ等のデータベース接続情報を記載します。

SQLAlchemy での接続表現文字列は以下のように定義します。DB アクセスのために必要な各項目を指定した文字列を作成し、create_engine 関数に渡すことで DB へ接続します。

diarect+driver://user:password@host:port/dbname

各パーツの意味については以下の通りです。

名称意味
diarectデータベースの種類
driverドライバ
user接続ユーザー
passwordパスワード
hostホスト名
portポート番号
dbname接続するデータベース名

diarect と driver には、使用する DB の種類によって以下のように指定します。

diarectdriver
sqlitepysqlite（省略可）
mysqlmysqlconnector
mysqlpymysql
mysqloursql
postgresqlpsycopg2
postgresqlpypostgresql

以降ではエンジンレイヤ、SQL表現言語、ORMのそれぞれのレベルでのプログラミング例について紹介していきます。

説明に使用するデータベースとしては SQLite を使用します。接続表現を変更すれば他の種類のデータベースでも同様です。

SQLite 自体の使い方については「SQLiteの基本的な使い方」を参考にしてください。

エンジンレイヤとしての使用

エンジンレイヤとして SQLAlchemy を使用する方法を紹介していきます。エンジンレイヤとしての使用では、具体的な SQL で最も低レベルでデータベースへのアクセスを行います。

実装例

CREATE TABLE、INSERT、SELECT、UPDATE、DELETE を実行する実装例を紹介します。詳細については、以降で説明していきます。

import sqlalchemy as sa
from sqlalchemy import text

# engine = sa.create_engine("sqlite+pysqlite:///test_engine.db")
# engine = sa.create_engine("sqlite+pysqlite:///test_engine.db", echo=True)
engine = sa.create_engine("sqlite+pysqlite:///:memory:")

print("===== CREATE TABLE =====")
with engine.connect() as conn:
    create_table_sql = text(
        "CREATE TABLE IF NOT EXISTS person ("
        "id INTEGER PRIMARY KEY AUTOINCREMENT"
        ",name VARCHAR"
        ",age INTEGER)"
    )
    conn.execute(create_table_sql)
    conn.commit()

print("\n==== INSERT =====")
with engine.connect() as conn:
    insert_sql = text("INSERT INTO person(name, age) VALUES(:name, :age)")
    result = conn.execute(insert_sql, {"name": "TARO", "age": 30})
    conn.commit()
    print(f"挿入件数: {result.rowcount}")

with engine.connect() as conn:
    insert_sql = text("INSERT INTO person(name, age) VALUES(:name, :age)")
    input_data = [
        {"name": "JIRO", "age": 20},
        {"name": "SABURO", "age": 10},
    ]
    result = conn.execute(insert_sql, input_data)
    conn.commit()
    print(f"挿入件数: {result.rowcount}")

print("\n===== SELECT =====")
with engine.connect() as conn:
    select_all_sql = text("SELECT * FROM person")
    rows = conn.execute(select_all_sql)
    for row in rows:
        print(row, type(row))
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

# 条件を指定する場合
print("=== 条件指定")
with engine.connect() as conn:
    select_sql = text("SELECT * FROM person WHERE age > :age")
    rows = conn.execute(select_sql, {"age": 20})
    for row in rows:
        print(row, type(row))
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

print("\n===== UPDATE =====")
with engine.connect() as conn:
    update_sql = text("UPDATE person SET name=:name, age=:age WHERE id = :id")
    result = conn.execute(update_sql, {"name": "SHIRO", "age": 40, "id": 1})
    conn.commit()
    print(f"更新行数: {result.rowcount}")

    # 更新後の結果表示
    select_all_sql = text("SELECT * FROM person")
    rows = conn.execute(select_all_sql)
    for row in rows:
        print(row, type(row))
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

# 複数行更新する場合
print("=== 複数行を更新")
with engine.connect() as conn:
    update_sql = text("UPDATE person SET name=:name, age=:age WHERE id = :id")
    update_data = [
        {"name": "GORO", "age": 50, "id": 2},
        {"name": "ROKURO", "age": 60, "id": 3},
    ]
    result = conn.execute(update_sql, update_data)
    conn.commit()
    print(f"更新行数: {result.rowcount}")

    # 更新後の結果表示
    select_all_sql = text("SELECT * FROM person")
    rows = conn.execute(select_all_sql)
    for row in rows:
        print(row, type(row))
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

print("\n===== DELETE =====")
with engine.connect() as conn:
    delete_sql = text("DELETE FROM person WHERE id = :id")
    result = conn.execute(delete_sql, {"id": 1})
    conn.commit()
    print(f"削除件数: {result.rowcount}")

    # 削除後の結果表示
    select_all_sql = text("SELECT * FROM person")
    rows = conn.execute(select_all_sql)
    for row in rows:
        print(row, type(row))
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

# 複数行を削除する場合
print("=== 複数行を削除")
with engine.connect() as conn:
    delete_sql = text("DELETE FROM person WHERE id = :id")
    delete_data = [
        {"id": 2},
        {"id": 3},
    ]
    result = conn.execute(delete_sql, delete_data)
    conn.commit()
    print(f"削除件数: {result.rowcount}")

    # 削除後の結果表示
    select_all_sql = text("SELECT * FROM person")
    rows = conn.execute(select_all_sql)
    for row in rows:
        print(row, type(row))
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")
【実行結果】
===== CREATE TABLE =====

==== INSERT =====
挿入件数: 1
挿入件数: 2

===== SELECT =====
(1, 'TARO', 30) <class 'sqlalchemy.engine.row.Row'>
id:1, name: TARO, age: 30
(2, 'JIRO', 20) <class 'sqlalchemy.engine.row.Row'>
id:2, name: JIRO, age: 20
(3, 'SABURO', 10) <class 'sqlalchemy.engine.row.Row'>
id:3, name: SABURO, age: 10
=== 条件指定
(1, 'TARO', 30) <class 'sqlalchemy.engine.row.Row'>
id:1, name: TARO, age: 30

===== UPDATE =====
更新行数: 1
(1, 'SHIRO', 40) <class 'sqlalchemy.engine.row.Row'>
id:1, name: SHIRO, age: 40
(2, 'JIRO', 20) <class 'sqlalchemy.engine.row.Row'>
id:2, name: JIRO, age: 20
(3, 'SABURO', 10) <class 'sqlalchemy.engine.row.Row'>
id:3, name: SABURO, age: 10
=== 複数行を更新
更新行数: 2
(1, 'SHIRO', 40) <class 'sqlalchemy.engine.row.Row'>
id:1, name: SHIRO, age: 40
(2, 'GORO', 50) <class 'sqlalchemy.engine.row.Row'>
id:2, name: GORO, age: 50
(3, 'ROKURO', 60) <class 'sqlalchemy.engine.row.Row'>
id:3, name: ROKURO, age: 60

===== DELETE =====
削除件数: 1
(2, 'GORO', 50) <class 'sqlalchemy.engine.row.Row'>
id:2, name: GORO, age: 50
(3, 'ROKURO', 60) <class 'sqlalchemy.engine.row.Row'>
id:3, name: ROKURO, age: 60
=== 複数行を削除
削除件数: 2

詳細説明

エンジンの作成
# engine = sa.create_engine("sqlite+pysqlite:///test_engine.db")
# engine = sa.create_engine("sqlite+pysqlite:///test_engine.db", echo=True)
engine = sa.create_engine("sqlite+pysqlite:///:memory:")

SQLAlchemy を使用するには、まず create_engine で接続するデータベース用のエンジンを作成します。create_engine に指定する接続表現文字列は上記で説明した内容を参考にしてください。

SQLAlchemy の動作結果を標準出力に出力する場合には echo 引数に True を設定します。SQLAlchemy がどのような SQL を発行しているかなどが表示されるため動作確認に便利です。

また、SQLite ではメモリ上にデータベースを作成する「:memory:」も使用できます。メモリ上で実行されるのでデータは保存されませんが動作確認には役に立ちます。

サンプルプログラムは何度実行しても同じ実行表示になるように「:memory:」の行を有効にしています。具体的にファイルに出力したり、echo=True を使う場合にはコメントアウトされている行を参考に変更して動作確認してみてください。

テーブルの作成 (CREATE TABLE)
# CREATE TABLE
print("===== CREATE TABLE =====")
with engine.connect() as conn:
    create_table_sql = text(
        "CREATE TABLE IF NOT EXISTS person ("
        "id INTEGER PRIMARY KEY AUTOINCREMENT"
        ",name VARCHAR"
        ",age INTEGER)"
    )
    conn.execute(create_table_sql)
    conn.commit()

SQLAlchemy で DB アクセスする際には、エンジンの connect メソッドを使用して、with 句で処理します。

CREATE TABLE の SQL を具体的に組み立てて実行していますが、SQLの文字列は text に渡して生成し、生成した SQL を実行するには execute メソッドを使用します。

with 句を抜ける際には接続が解放され、トランザクションが終了します。変更などの確定が必要な処理の場合は commit メソッドで確定するようにしてください。（CREATE TABLE は commit しなくてもテーブルが作成されますが、手順の流れが分かりやすいように commit を含めて記載しています。）

データの挿入 (INSERT)
print("\n==== INSERT =====")
with engine.connect() as conn:
    insert_sql = text("INSERT INTO person(name, age) VALUES(:name, :age)")
    result = conn.execute(insert_sql, {"name": "TARO", "age": 30})
    conn.commit()
    print(f"挿入件数: {result.rowcount}")

with engine.connect() as conn:
    insert_sql = text("INSERT INTO person(name, age) VALUES(:name, :age)")
    input_data = [
        {"name": "JIRO", "age": 20},
        {"name": "SABURO", "age": 10},
    ]
    result = conn.execute(insert_sql, input_data)
    conn.commit()
    print(f"挿入件数: {result.rowcount}")

データを挿入する際には text で INSERT の SQL 文字列を作成し、execute メソッドで実行します。SQL内でデータを指定する部分については「:変数」というようなプレースホルダー表現を用います。例では「:name」や「:age」の部分です。

execute メソッドには、INSERT 用 SQL とプレースホルダーに埋め込む値を指定した辞書を指定します。複数データを一括挿入したい場合には、辞書のリストを指定することで複数まとめて処理が可能です。

また、execute メソッドが返却するオブジェクトの rowcount で処理した件数を確認が可能です。

データの検索 (SELECT)
print("\n===== SELECT =====")
with engine.connect() as conn:
    select_all_sql = text("SELECT * FROM person")
    rows = conn.execute(select_all_sql)
    for row in rows:
        print(row, type(row))
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

# 条件を指定する場合
print("=== 条件指定")
with engine.connect() as conn:
    select_sql = text("SELECT * FROM person WHERE age > :age")
    rows = conn.execute(select_sql, {"age": 20})
    for row in rows:
        print(row, type(row))
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

データを検索する際には text で SELECT のSQL 文字列を作成し、execute メソッドで実行します。

execute メソッドの返却値は for により順にレコードを取得できます。type で型を表示すると <class 'sqlalchemy.engine.row.Row'> クラスのオブジェクトとなっていることが分かります。

各レコードの列の値にアクセスするには「row.id」「row.name」「row.age」のように「.」(ドット) を使って列名でアクセスできます。

条件を指定したい場合は、プレースホルダ―（上記の例では「:age」）を使うことで、条件に一致するレコードのみを抽出することができます。

データの更新 (UPDATE)
print("\n===== UPDATE =====")
with engine.connect() as conn:
    update_sql = text("UPDATE person SET name=:name, age=:age WHERE id = :id")
    result = conn.execute(update_sql, {"name": "SHIRO", "age": 40, "id": 1})
    conn.commit()
    print(f"更新行数: {result.rowcount}")

    # 更新後の結果表示
    select_all_sql = text("SELECT * FROM person")
    rows = conn.execute(select_all_sql)
    for row in rows:
        print(row, type(row))
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

# 複数行更新する場合
print("=== 複数行を更新")
with engine.connect() as conn:
    update_sql = text("UPDATE person SET name=:name, age=:age WHERE id = :id")
    update_data = [
        {"name": "GORO", "age": 50, "id": 2},
        {"name": "ROKURO", "age": 60, "id": 3},
    ]
    result = conn.execute(update_sql, update_data)
    conn.commit()
    print(f"更新行数: {result.rowcount}")

    # 更新後の結果表示
    select_all_sql = text("SELECT * FROM person")
    rows = conn.execute(select_all_sql)
    for row in rows:
        print(row, type(row))
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

データを更新する際には text で UPDATE の SQL 文字列を作成し、execute メソッドで実行します。

使い方は基本的に INSERT の場合と同じです。値を埋め込みたい部分をプレースホルダー表現にして、execute メソッドに SQL と辞書で値を指定します。複数行を更新したい場合は、辞書のリストで値を指定します。

データの削除 (DELETE)
print("\n===== DELETE =====")
with engine.connect() as conn:
    delete_sql = text("DELETE FROM person WHERE id = :id")
    result = conn.execute(delete_sql, {"id": 1})
    conn.commit()
    print(f"削除件数: {result.rowcount}")

    # 削除後の結果表示
    select_all_sql = text("SELECT * FROM person")
    rows = conn.execute(select_all_sql)
    for row in rows:
        print(row, type(row))
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

# 複数行を削除する場合
print("=== 複数行を削除")
with engine.connect() as conn:
    delete_sql = text("DELETE FROM person WHERE id = :id")
    delete_data = [
        {"id": 2},
        {"id": 3},
    ]
    result = conn.execute(delete_sql, delete_data)
    conn.commit()
    print(f"削除件数: {result.rowcount}")

    # 削除後の結果表示
    select_all_sql = text("SELECT * FROM person")
    rows = conn.execute(select_all_sql)
    for row in rows:
        print(row, type(row))
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

データを削除する際には text で DELETE の SQL 文字列を作成し、execute メソッドで実行します。

上記のように使い方は基本的に INSERT や UPDATE の場合と同じです。値を埋め込みたい部分をプレースホルダー表現にして、execute メソッドに SQL と辞書で削除条件を指定します。複数行を削除したい場合は、辞書のリストで値を指定します。

SQL 表現言語を用いた使用

SQL 表現言語を用いた SQLAlchemy を使用する方法を紹介していきます。

SQL 表現言語を用いた方法では、Python の式として、SQL 操作関数を使用することで、データベース間の SQL 表現の違いを吸収してデータベースへアクセスを行うことができます。上記で紹介した SQL を具体的に記載するエンジンレイヤと後述の ORM のちょうど中間に位置する表現方法になります。

実装例

CREATE TABLE、INSERT、SELECT、UPDATE、DELETE を実行する実装例を紹介します。詳細の説明は以降で行います。

import sqlalchemy as sa
from sqlalchemy import Column, Integer, MetaData, String, Table, bindparam

# engine = sa.create_engine("sqlite+pysqlite:///test_sql.db")
# engine = sa.create_engine("sqlite+pysqlite:///test_sql.db", echo=True)
engine = sa.create_engine("sqlite+pysqlite:///:memory:")

# メタデータの生成
meta = MetaData()

print("===== CREATE TABLE =====")
person = Table(
    "person",
    meta,
    Column("id", Integer, primary_key=True, autoincrement=True),
    Column("name", String(50)),
    Column("age", Integer),
)
meta.create_all(engine)

print("\n===== INSERT =====")
# 1行挿入する場合
with engine.connect() as conn:
    insert_sql = person.insert().values(name="TARO", age=30)
    result = conn.execute(insert_sql)
    conn.commit()
    print(f"挿入件数: {result.rowcount}")

# 複数行挿入する場合
with engine.connect() as conn:
    insert_sql = person.insert()
    input_data = [
        {"name": "JIRO", "age": 20},
        {"name": "SABURO", "age": 10},
    ]
    result = conn.execute(insert_sql, input_data)
    conn.commit()
    print(f"挿入件数: {result.rowcount}")

print("\n===== SELECT =====")
# 全て検索する場合
with engine.connect() as conn:
    select_all_sql = person.select()
    rows = conn.execute(select_all_sql)
    for row in rows:
        print(row, type(row))
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

# 条件で検索する場合
print("=== 条件指定")
with engine.connect() as conn:
    select_sql = person.select().where(person.c.age > 20)
    rows = conn.execute(select_sql)
    for row in rows:
        print(row, type(row))
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

print("\n===== UPDATE =====")
with engine.connect() as conn:
    update_sql = (
        person.update().where(person.c.id == 1).values(name="SHIRO", age=40)
    )
    result = conn.execute(update_sql)
    conn.commit()
    print(f"更新行数: {result.rowcount}")

    # 更新後の結果を確認
    select_all_sql = person.select()
    rows = conn.execute(select_all_sql)
    for row in rows:
        print(row, type(row))
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

# 複数行更新する場合
print("=== 複数行を更新")
with engine.connect() as conn:
    update_sql = (
        person.update()
        .where(person.c.id == bindparam("old_id"))
        .values(name=bindparam("new_name"), age=bindparam("new_age"))
    )
    update_data = [
        {"old_id": 2, "new_name": "GORO", "new_age": 50},
        {"old_id": 3, "new_name": "ROKURO", "new_age": 60},
    ]
    result = conn.execute(update_sql, update_data)
    conn.commit()
    print(f"更新行数: {result.rowcount}")

    # 更新後の結果を確認
    select_all_sql = person.select()
    rows = conn.execute(select_all_sql)
    for row in rows:
        print(row, type(row))
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

print("\n===== DELETE =====")
with engine.connect() as conn:
    delete_sql = person.delete().where(person.c.id == 1)
    result = conn.execute(delete_sql)
    conn.commit()
    print(f"削除件数: {result.rowcount}")

    # 削除後の結果を確認
    select_all_sql = person.select()
    rows = conn.execute(select_all_sql)
    for row in rows:
        print(row, type(row))
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

# 複数行を削除する場合
print("=== 複数行を削除")
with engine.connect() as conn:
    delete_sql = person.delete().where(person.c.id == bindparam("delete_id"))
    delete_data = [
        {"delete_id": 2},
        {"delete_id": 3},
    ]
    result = conn.execute(delete_sql, delete_data)
    conn.commit()
    print(f"削除件数: {result.rowcount}")

    # 削除後の結果を確認
    select_all_sql = person.select()
    rows = conn.execute(select_all_sql)
    for row in rows:
        print(row, type(row))
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")
【実行結果】===== CREATE TABLE =====

===== INSERT =====
挿入件数: 1
挿入件数: 2

===== SELECT =====
(1, 'TARO', 30) <class 'sqlalchemy.engine.row.Row'>
id:1, name: TARO, age: 30
(2, 'JIRO', 20) <class 'sqlalchemy.engine.row.Row'>
id:2, name: JIRO, age: 20
(3, 'SABURO', 10) <class 'sqlalchemy.engine.row.Row'>
id:3, name: SABURO, age: 10
=== 条件指定
(1, 'TARO', 30) <class 'sqlalchemy.engine.row.Row'>
id:1, name: TARO, age: 30

===== UPDATE =====
更新行数: 1
(1, 'SHIRO', 40) <class 'sqlalchemy.engine.row.Row'>
id:1, name: SHIRO, age: 40
(2, 'JIRO', 20) <class 'sqlalchemy.engine.row.Row'>
id:2, name: JIRO, age: 20
(3, 'SABURO', 10) <class 'sqlalchemy.engine.row.Row'>
id:3, name: SABURO, age: 10
=== 複数行を更新
更新行数: 2
(1, 'SHIRO', 40) <class 'sqlalchemy.engine.row.Row'>
id:1, name: SHIRO, age: 40
(2, 'GORO', 50) <class 'sqlalchemy.engine.row.Row'>
id:2, name: GORO, age: 50
(3, 'ROKURO', 60) <class 'sqlalchemy.engine.row.Row'>
id:3, name: ROKURO, age: 60

===== DELETE =====
削除件数: 1
(2, 'GORO', 50) <class 'sqlalchemy.engine.row.Row'>
id:2, name: GORO, age: 50
(3, 'ROKURO', 60) <class 'sqlalchemy.engine.row.Row'>
id:3, name: ROKURO, age: 60
=== 複数行を削除
削除件数: 2

詳細説明

エンジンの作成

データベースへ接続するためのエンジンの作成方法は「エンジンレイヤ」での使用と同じですので、そちらを参照してください。

テーブルの作成 (CREATE TABLE)
from sqlalchemy import Column, Integer, MetaData, String, Table, bindparam

...(途中省略)...

# メタデータの生成
meta = MetaData()

print("===== CREATE TABLE =====")
person = Table(
    "person",
    meta,
    Column("id", Integer, primary_key=True, autoincrement=True),
    Column("name", String(50)),
    Column("age", Integer),
)
meta.create_all(engine)

SQL 表現言語を用いる場合には、上記のように MetaData クラスのオブジェクトを生成し、テーブル定義には Table クラスを使用します。また、その他定義に必要なクラス等 (Column、Integer、MetaData、String、Table、bindparam) をインポートします。

Table クラスを生成する際の引数としてはテーブル名、メタデータオブジェクト、その後に各列を定義します。列は Column クラスで生成し、型は SQLAlchemy で定義されている String や Integer といったクラスを指定します。

また、主キー (primary_key) や自動インクリメント (autoincrement) といった引数を指定することも可能です。

create_engine の echo 引数を True にしておくと標準出力に以下のように発行した SQL を確認できます。例だと以下のようになります。

CREATE TABLE person (
	id INTEGER NOT NULL, 
	name VARCHAR(50), 
	age INTEGER, 
	PRIMARY KEY (id)
)

上記結果から String と Integer クラスはそれぞれ VARCHAR と INTEGER に該当していることが分かります。このように SQL 表現言語を使うと、具体的な SQL 文を意識しないでもデータベース操作が可能になります。

では、INSERT、SELECT、UPDATE、DELETE の使い方を順に見ていきましょう。

データの挿入 (INSERT)
print("\n===== INSERT =====")
# 1行挿入する場合
with engine.connect() as conn:
    insert_sql = person.insert().values(name="TARO", age=30)
    result = conn.execute(insert_sql)
    conn.commit()
    print(f"挿入件数: {result.rowcount}")

# 複数行挿入する場合
with engine.connect() as conn:
    insert_sql = person.insert()
    input_data = [
        {"name": "JIRO", "age": 20},
        {"name": "SABURO", "age": 10},
    ]
    result = conn.execute(insert_sql, input_data)
    conn.commit()
    print(f"挿入件数: {result.rowcount}")

SQL 表現言語でデータを挿入するには「テーブル.insert().values(列名=値, ...)」というように指定することでデータを挿入できます。また、複数データを一括で挿入したい場合には、辞書のリストを指定することで複数件まとめて挿入が可能です。

データの検索 (SELECT)
print("\n===== SELECT =====")
# 全て検索する場合
with engine.connect() as conn:
    select_all_sql = person.select()
    rows = conn.execute(select_all_sql)
    for row in rows:
        print(row, type(row))
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

# 条件で検索する場合
print("=== 条件指定")
with engine.connect() as conn:
    select_sql = person.select().where(person.c.age > 20)
    rows = conn.execute(select_sql)
    for row in rows:
        print(row, type(row))
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

SQL 表現言語でデータを検索するには、テーブルの select メソッドを使用します。select の引数を指定しない場合は全件取得の SQL を取得でき、execute メソッドに指定することで実行できます。

SELECT の検索条件を指定する場合は、「テーブル.select().where(条件)」のように where に検索条件を指定することで抽出ができます。なお、列名は「テーブル.c.列名」といった形式でアクセスできます。テーブルと列名の間に「c」が必要なので注意してください。

データの更新 (UPDATE)
print("\n===== UPDATE =====")
with engine.connect() as conn:
    update_sql = (
        person.update().where(person.c.id == 1).values(name="SHIRO", age=40)
    )
    result = conn.execute(update_sql)
    conn.commit()
    print(f"更新行数: {result.rowcount}")

    # 更新後の結果を確認
    select_all_sql = person.select()
    rows = conn.execute(select_all_sql)
    for row in rows:
        print(row, type(row))
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

# 複数行更新する場合
print("=== 複数行を更新")
with engine.connect() as conn:
    update_sql = (
        person.update()
        .where(person.c.id == bindparam("old_id"))
        .values(name=bindparam("new_name"), age=bindparam("new_age"))
    )
    update_data = [
        {"old_id": 2, "new_name": "GORO", "new_age": 50},
        {"old_id": 3, "new_name": "ROKURO", "new_age": 60},
    ]
    result = conn.execute(update_sql, update_data)
    conn.commit()
    print(f"更新行数: {result.rowcount}")

    # 更新後の結果を確認
    select_all_sql = person.select()
    rows = conn.execute(select_all_sql)
    for row in rows:
        print(row, type(row))
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

SQL 表現言語でデータを更新するには、テーブルの update メソッドを使用して「テーブル.update().where(条件).values(列名=値, ...)」という形式で更新情報を指定することで UPDATE の SQL を取得し、execute メソッドに指定することでデータ更新を実行します。

複数の条件と値で更新を実行したい場合には、sqlalchemy から bindparam をインポートし、プレースホルダーとなる引数名を指定します。後は、辞書のリストで条件や値に該当する情報を指定することで複数件をまとめて更新することができます。

データの削除 (DELETE)
print("\n===== DELETE =====")
with engine.connect() as conn:
    delete_sql = person.delete().where(person.c.id == 1)
    result = conn.execute(delete_sql)
    conn.commit()
    print(f"削除件数: {result.rowcount}")

    # 削除後の結果を確認
    select_all_sql = person.select()
    rows = conn.execute(select_all_sql)
    for row in rows:
        print(row, type(row))
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

# 複数行を削除する場合
print("=== 複数行を削除")
with engine.connect() as conn:
    delete_sql = person.delete().where(person.c.id == bindparam("delete_id"))
    delete_data = [
        {"delete_id": 2},
        {"delete_id": 3},
    ]
    result = conn.execute(delete_sql, delete_data)
    conn.commit()
    print(f"削除件数: {result.rowcount}")

    # 削除後の結果を確認
    select_all_sql = person.select()
    rows = conn.execute(select_all_sql)
    for row in rows:
        print(row, type(row))
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

SQL 表現言語でデータを削除するには、テーブルの delete メソッドを使用して「テーブル.delete().where(条件)」というような形式で削除対象を指定することで DELETE の SQL を取得し、execute メソッドに指定することでデータ削除を実行します。

複数の条件で削除したい場合には、sqlalchemy から bindparam をインポートして、プレースホルダーとなる引数名を指定します。後は、辞書のリストで条件に該当する情報を指定することで複数件をまとめて削除することができます。

テーブル情報を取得する方法 autoload_with

既に登録されているテーブル情報を取得するには以下のようにします。

import sqlalchemy as sa
from sqlalchemy import MetaData, Table

engine = sa.create_engine("sqlite+pysqlite:///test_sql.db")

# メタデータの生成
meta = MetaData()
# テーブル情報の取得
person = Table("person", meta, autoload_with=engine)

# ===== 各種person対するデータ処理を記載
# ......

テーブル情報を取得するには、対象 DB のエンジンを生成した後に Table クラスに、テーブル名とメタデータオブジェクトを指定し、autoload_with 引数に生成したエンジンを指定します。

このようにすることでデータベースに登録されている person テーブル情報を取得できます。後は上記で紹介した person テーブルに対する各種処理を実行できます。

オブジェクト関係マッピング（ORM）としての使用

オブジェクト関係マッピング (ORM: Object Relation Mapping) を用いた SQLAlchemy の使用例を紹介していきます。

ORM では、Python クラスのオブジェクトを定義し、オブジェクトとデータベース構造をマッピングすることによりデータベースアクセスを行う方法です。クラスのオブジェクトを操作することが直接データベースの値の操作に該当します。

実装例

ORM ではまずテーブルを表すクラスを作成する必要があります。クラスの定義方法とCREATE TABLE、INSERT、SELECT、UPDATE、DELETE を 1 通り実行する実装例を紹介します。詳細の説明は以降で行います。

クラス定義 sqlalchemy_orm_person.py
from sqlalchemy import Integer, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class Person(Base):
    __tablename__ = "person"

    # 属性名はテーブル名の列名に対応する
    id: Mapped[int] = mapped_column(
        Integer,
        primary_key=True,
        autoincrement=True,
    )
    name: Mapped[str] = mapped_column(String(50))
    age: Mapped[int] = mapped_column(Integer)

    def __repr__(self):
        return f"<Person({self.id}, {self.name}, {self.age})>"
クラスの呼び出し／使用
import sqlalchemy as sa
import sqlalchemy_orm_person
from sqlalchemy import delete, select, update
from sqlalchemy.orm import Session
from sqlalchemy_orm_person import Person

# engine = sa.create_engine("sqlite+pysqlite:///test_orm.db")
# engine = sa.create_engine("sqlite+pysqlite:///test_orm.db", echo=True)
engine = sa.create_engine("sqlite+pysqlite:///:memory:")

print("===== CREATE TABLE =====")
sqlalchemy_orm_person.Base.metadata.create_all(engine)

print("\n===== INSERT =====")
with Session(engine) as session:
    taro = Person(name="TARO", age=30)
    jiro = Person(name="JIRO", age=20)
    saburo = Person(name="SABURO", age=10)

    session.add_all([taro, jiro, saburo])
    session.commit()

print("\n===== SELECT =====")
with Session(engine) as session:
    select_all_sql = select(Person)
    rows = session.scalars(select_all_sql)
    for row in rows:
        print(row)
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

# 条件で検索する場合
print("=== 条件指定")
with Session(engine) as session:
    select_sql = select(Person).where(Person.age > 20)
    rows = session.scalars(select_sql)
    for row in rows:
        print(row)
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

print("\n===== UPDATE =====")
with Session(engine) as session:
    # 主キーを指定して取得する
    target_person = session.get(Person, 1)
    # # 以下でも同様
    # select_sql = select(Person).where(Person.id == 1)
    # target_person = session.scalars(select_sql).one()
    target_person.name = "SHIRO"
    target_person.age = 40
    session.commit()

    # 更新後の結果を確認
    select_all_sql = select(Person)
    rows = session.scalars(select_all_sql)
    for row in rows:
        print(row)
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

# 複数行更新する場合
print("=== 複数行を更新")
with Session(engine) as session:
    update_sql = update(Person)
    update_data = [
        {"id": 2, "name": "GORO", "age": 50},
        {"id": 3, "name": "ROKURO", "age": 60},
    ]
    session.execute(update_sql, update_data)
    session.commit()

    # 更新後の結果を確認
    select_all_sql = select(Person)
    rows = session.scalars(select_all_sql)
    for row in rows:
        print(row)
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

print("\n===== DELETE =====")
with Session(engine) as session:
    # 主キーを指定して取得する
    target_person = session.get(Person, 1)
    # # 以下でも同様
    # select_sql = select(Person).where(Person.id == 1)
    # target_person = session.scalars(select_sql).one()

    # データを削除する
    session.delete(target_person)
    session.commit()

    # 削除後の結果を確認
    select_all_sql = select(Person)
    rows = session.scalars(select_all_sql)
    for row in rows:
        print(row)
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

# 複数行削除する場合
print("=== 複数行を削除")
with Session(engine) as session:
    delete_sql = delete(Person).where(Person.id.in_([2, 3]))
    session.execute(delete_sql)
    session.commit()

    # 更新後の結果を確認
    select_all_sql = select(Person)
    rows = session.scalars(select_all_sql)
    for row in rows:
        print(row)
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")
【実行結果】
===== CREATE TABLE =====

===== INSERT =====

===== SELECT =====
<Person(1, TARO, 30)>
id:1, name: TARO, age: 30
<Person(2, JIRO, 20)>
id:2, name: JIRO, age: 20
<Person(3, SABURO, 10)>
id:3, name: SABURO, age: 10
=== 条件指定
<Person(1, TARO, 30)>
id:1, name: TARO, age: 30

===== UPDATE =====
<Person(1, SHIRO, 40)>
id:1, name: SHIRO, age: 40
<Person(2, JIRO, 20)>
id:2, name: JIRO, age: 20
<Person(3, SABURO, 10)>
id:3, name: SABURO, age: 10
=== 複数行を更新
<Person(1, SHIRO, 40)>
id:1, name: SHIRO, age: 40
<Person(2, GORO, 50)>
id:2, name: GORO, age: 50
<Person(3, ROKURO, 60)>
id:3, name: ROKURO, age: 60

===== DELETE =====
<Person(2, GORO, 50)>
id:2, name: GORO, age: 50
<Person(3, ROKURO, 60)>
id:3, name: ROKURO, age: 60
=== 複数行を削除

詳細説明

クラス定義
class Base(DeclarativeBase):
    pass


class Person(Base):
    __tablename__ = "person"

    # 属性名はテーブル名の列名に対応する
    id: Mapped[int] = mapped_column(
        Integer,
        primary_key=True,
        autoincrement=True,
    )
    name: Mapped[str] = mapped_column(String(50))
    age: Mapped[int] = mapped_column(Integer)

    def __repr__(self):
        return f"<Person({self.id}, {self.name}, {self.age})>"

ORM を使用する場合には、テーブルに相当するクラスを定義します。

まず、DeclarativeBase を継承した Base クラスを作成します。内容は pass で構いません。次に、テーブルに該当するクラスを定義した Base を継承したクラスとして定義します。今回の例では １ テーブルのクラスのみですが他にも Base クラスを継承したテーブルを作ってもらって構いません。

__tablename__ はテーブル名の文字列を指定します。また、列名は Mapped[型] といった型ヒントの形式で指定し、id や name、age は列名に該当します。mapped_column の引数には当該列の型を指定します。

__repr__ メソッドは、print に当該クラスのオブジェクトが渡されたときの表示結果を定義します。Person クラスのオブジェクトを print に渡すと実行結果を見てもわかるように「<Person(1, TARO, 30)>」のような表示結果となります。

エンジンの作成

データベースへ接続するためのエンジンの作成についてはエンジンレイヤとしての使用方法と同じですので、そちらを参照してください。

テーブルの作成 (CREATE TABLE)
import sqlalchemy_orm_person
from sqlalchemy_orm_person import Person

...(途中省略)...

print("===== CREATE TABLE =====")
sqlalchemy_orm_person.Base.metadata.create_all(engine)

今回は、クラス定義と実行処理を別ファイルにしているのでクラス定義をしているファイルをインポートして使用しています。

テーブルを作成する場合には、クラスで定義した Base クラスを用いて Base.metadata.create_all(engine) で生成します。Base を継承したテーブルが複数ある場合は、この 1 行で複数テーブルの CREATE TABLE を一括実行できます。

データの挿入 (INSERT)
print("\n===== INSERT =====")
with Session(engine) as session:
    taro = Person(name="TARO", age=30)
    jiro = Person(name="JIRO", age=20)
    saburo = Person(name="SABURO", age=10)

    session.add_all([taro, jiro, saburo])
    session.commit()

ORM を用いた処理の場合は、Session(engine) のようにデータベース処理のセッションを with 句で作成します。エンジンレイヤとしての使用や SQL 表現言語での使用の場合では、engine.connect() で with 句を生成していた点と少し異なります。

まず Person オブジェクトを作成しますが、この各オブジェクトがテーブルの 1 行に該当します。これらのデータを挿入する場合には「session.add_all([オブジェクト, ...])」でまとめて挿入できます。

データの検索 (SELECT)
print("\n===== SELECT =====")
with Session(engine) as session:
    select_all_sql = select(Person)
    rows = session.scalars(select_all_sql)
    for row in rows:
        print(row)
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

# 条件で検索する場合
print("=== 条件指定")
with Session(engine) as session:
    select_sql = select(Person).where(Person.age > 20)
    rows = session.scalars(select_sql)
    for row in rows:
        print(row)
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

ORM を用いた場合で全件のデータを取得する場合には、select メソッドに対象テーブルのクラスを指定することで全件取得の SQL を取得できます。SQL の実行では session.scalars に生成した SQL を指定することでデータ取得が実行できます。

条件指定したい場合は where で条件を指定します。この時の条件は「クラス名.列名」の形式で指定することが可能です。

データの更新 (UPDATE)
print("\n===== UPDATE =====")
with Session(engine) as session:
    # 主キーを指定して取得する
    target_person = session.get(Person, 1)
    # # 以下でも同様
    # select_sql = select(Person).where(Person.id == 1)
    # target_person = session.scalars(select_sql).one()
    target_person.name = "SHIRO"
    target_person.age = 40
    session.commit()

    # 更新後の結果を確認
    select_all_sql = select(Person)
    rows = session.scalars(select_all_sql)
    for row in rows:
        print(row)
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

# 複数行更新する場合
print("=== 複数行を更新")
with Session(engine) as session:
    update_sql = update(Person)
    update_data = [
        {"id": 2, "name": "GORO", "age": 50},
        {"id": 3, "name": "ROKURO", "age": 60},
    ]
    session.execute(update_sql, update_data)
    session.commit()

    # 更新後の結果を確認
    select_all_sql = select(Person)
    rows = session.scalars(select_all_sql)
    for row in rows:
        print(row)
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

ORM を用いた場合でデータを更新するには、まず対象のデータをオブジェクトとして取得してオブジェクトの値を直接変更します。

主キーを用いてデータを取得する場合には、session.get(クラス, 主キーの値) といった形式でデータを取得することができます。なお、select(クラス).where(条件) で条件を指定して取得しても構いません。

取得したオブジェクトの値を直接変更し、session.commit() をするだけでテーブルのデータが更新されます。

また、複数データを一括更新したい場合には、まず update(クラス) で update 文の SQL を生成し、辞書に主キーの値と他の列の設定値を指定して session.execute で実行することが可能です。この方法を使う場合は、主キーの指定が必要になります。

データの削除 (DELETE)
print("\n===== DELETE =====")
with Session(engine) as session:
    # 主キーを指定して取得する
    target_person = session.get(Person, 1)
    # # 以下でも同様
    # select_sql = select(Person).where(Person.id == 1)
    # target_person = session.scalars(select_sql).one()

    # データを削除する
    session.delete(target_person)
    session.commit()

    # 削除後の結果を確認
    select_all_sql = select(Person)
    rows = session.scalars(select_all_sql)
    for row in rows:
        print(row)
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

# 複数行削除する場合
print("=== 複数行を削除")
with Session(engine) as session:
    delete_sql = delete(Person).where(Person.id.in_([2, 3]))
    session.execute(delete_sql)
    session.commit()

    # 更新後の結果を確認
    select_all_sql = select(Person)
    rows = session.scalars(select_all_sql)
    for row in rows:
        print(row)
        print(f"id:{row.id}, name: {row.name}, age: {row.age}")

ORM を用いてデータを削除する場合は、更新の場合と似ています。対象データを取得し、session.delete の引数に対象オブジェクトを指定することで削除できます。

また、複数行を削除したい場合には、delete(クラス).where(条件) で対象を絞る SQL を作成し、session.execute に引数として渡すことで複数行をまとめて削除することが可能です。

SQLAlchemy をどのレベルで使用するか

SQLAlchemy を使ってエンジンレイヤ、SQL 表現言語、ORM といったレベルでの扱い方の基本的な使い方を紹介してきました。ただ、SQLAlchemy を使う場合に、どのレベルを使用してプログラミングをするべきかに悩むかもしれません。

エンジンレイヤ → SQL 表現言語 → ORM にいくほどより抽象化されたものになっていくので、SQL の細かな部分を意識しないでよくなります。そのため、ORM を選べばよいように思うかもしれませんが、プログラムは思うように動かなくなることがあるということを考えておく必要があります。

プログラムの動作で問題が起こった時に、ORM がどのような仕組みで動いているのかを理解できていないと、本当の意味で問題を解決できない場合があります。create_engine の echo 引数を True にして、ある程度動作は確認ができるものの細かな動きを把握することは難しい場合があるため、場合によってはエンジンレイヤや SQL 表現言語のレベルでの実装の方が解決が容易な場合があります。

SQLAlchemy を使用する際には、プログラムの複雑さなどにより使用するレベルを検討するとよいと思います。ORM レベルを使用する場合は、単純なアプリケーション開発に使うようにした方がよいかもしれません。

まとめ

Python のクロスデータベースライブラリの中でよく知られている SQLAlchemy について基本的な使い方を解説しました。

SQLAlchemy では「エンジンレイヤ」「SQL表現言語」「ORM」といった使用のレベルを使い分けることができ、それぞれの使用方法について説明しました。

エンジンレイヤ → SQL表現言語 → ORM にいくほどより抽象化された方法ですが、内部の動きを理解できていないと問題発生時に解決が難しくなる場合があります。SQLAlchemy を使用する場合には、どのレベルでプログラミングを行うかをよく検討しましょう。

SQLAlchemy の公式ページはこちらを参照してください。

ソースコード

上記で紹介しているソースコードについては GitHub にて公開しています。参考にしていただければと思います。

あわせて読みたい
【Python Tech】プログラミングガイド
【Python Tech】プログラミングガイド
ABOUT ME
ホッシー
ホッシー
システムエンジニア
Contact
はじめまして。当サイトをご覧いただきありがとうございます。 私は製造業のメーカーで、DX推進や業務システムの設計・開発・導入を担当しているシステムエンジニアです。これまでに転職も経験しており、以前は大手電機メーカーでシステム開発に携わっていました。

プログラミング言語はこれまでC、C++、JAVA等を扱ってきましたが、最近では特に機械学習等の分析でも注目されているPythonについてとても興味をもって取り組んでいます。これまでの経験をもとに、Pythonに興味を持つ方のお役に立てるような情報を発信していきたいと思います。どうぞよろしくお願いいたします。

※キャラクターデザイン：ゼイルン様
記事URLをコピーしました