SQLAlchemy

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

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

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の部分の種類としては以下のようなものがあります。

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句を抜ける際には接続が解放され、ROLLBACKが発行されてトランザクションが終了します。そのため、確定が必要な処理の場合は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表現言語でデータを削除するには、personテーブルの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を一通り実行する実装例を紹介します。詳細の説明は以降で行います。

クラス定義 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を継承したテーブルが複数ある場合はこの一行で複数テーブルの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オブジェクトを作成しますが、この各オブジェクトがテーブルの１行に該当します。これらのデータを挿入する場合には、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を使用する場合には、どのレベルでプログラミングを行うかをよく検討しましょう。

Note

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

ソースコード

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

Python Tech Pythonプログラミングガイド
【Python Tech】Pythonプログラミングガイド本サイトで紹介しているPython入門、応用に関する記事のまとめページです。Python文法の入門から高度な内容、おすすめの書籍についても紹介しています。また、機械学習等のデータ分析やGUI開発、近年話題の生成AI等のトピックスについても含んでいます。...
ORM SQLite SQL表現言語 エンジンレイヤ オブジェクト関係マッピング