まともに OR マッパとして使うのではなく、ちょっとした Notebook やテストコードの中で SQLAlchemy の Engine オブジェクトをさっと使う方法を毎回忘れるのでメモ。だいたいここ https://docs.sqlalchemy.org/en/14/core/connections.html を読めば良い。

from sqlalchemy import create_engine
from sqlalchemy import text

DB 接続と connection 作成。engine.execute() のように直接 engine にクエリを発行する方法は 1.4 廃止される(参考: https://docs.sqlalchemy.org/en/14/core/connections.html#sqlalchemy.engine.Engine.execute) らしいので、ちゃんと connection を作る。

engine = create_engine("mysql+mysqlconnector://root@localhost/mydb")
connection = engine.connect()

テスト用のテーブルを作成

connection.execute("CREATE TABLE family (name varchar(256), age int)")

<sqlalchemy.engine.cursor.LegacyCursorResult at 0x10b7bed60>

テスト用のデータを作成。

connection.execute("""
INSERT INTO family VALUES
    ('Sazae', 24),
    ('Masuo', 28),
    ('Tarao', 3)
""")

<sqlalchemy.engine.cursor.LegacyCursorResult at 0x10b200f40>

SELECT を発行 text コンストラクタ (https://docs.sqlalchemy.org/en/14/core/sqlelement.html#sqlalchemy.sql.expression.text) を使えば、DB エンジン互換の方法でパラメータを指定出来る。text を使わないと DB ごとにパラメータ指定が違ってしまう。

result = connection.execute(text("SELECT * FROM family WHERE age > :threashold"), threashold = 20)

イテレータが返るので SELECT 結果を取り出す。

results = list(result)
results

[('Sazae', 24), ('Masuo', 28)]

結果は Row オブジェクト (https://docs.sqlalchemy.org/en/14/core/connections.html#sqlalchemy.engine.Row) というのに格納されている。

results[0]._fields

('name', 'age')

results[0].age

テスト用データを削除。connection も閉じる。with 文を使えば自動的に connection を閉じる事が出来る。

connection.execute("DROP TABLE family")
connection.close()