sqlalchemyのORMじゃない方の機能(expression language api)

使い方忘れることが多いのでメモ。

準備

テキトウにデータを作って準備する。

テキトウにテーブル作成。

CREATE TABLE groups (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  name TEXT
);
CREATE TABLE users (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  group_id INTEGER,
  name TEXT,
  FOREIGN KEY(group_id) REFERENCES groups(id)
);
CREATE TABLE skills (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  user_id INTEGER,
  name TEXT,
  FOREIGN KEY(user_id) REFERENCES users(id)
);

テキトウにデータを追加。

INSERT INTO groups (id, name) VALUES(NULL, 'X');
INSERT INTO groups (id, name) VALUES(NULL, 'Y');
INSERT INTO groups (id, name) VALUES(NULL, 'Z');
INSERT INTO users (id, group_id, name) VALUES(NULL, 1, 'foo');
INSERT INTO users (id, group_id, name) VALUES(NULL, 1, 'bar');
INSERT INTO users (id, group_id, name) VALUES(NULL, 2, 'boo');
INSERT INTO skills (id, user_id, name) VALUES(NULL, 1, 'a');
INSERT INTO skills (id, user_id, name) VALUES(NULL, 1, 'b');
INSERT INTO skills (id, user_id, name) VALUES(NULL, 1, 'c');
INSERT INTO skills (id, user_id, name) VALUES(NULL, 2, 'a');
INSERT INTO skills (id, user_id, name) VALUES(NULL, 2, 'b');
INSERT INTO skills (id, user_id, name) VALUES(NULL, 2, 'c');
INSERT INTO skills (id, user_id, name) VALUES(NULL, 3, 'a');
INSERT INTO skills (id, user_id, name) VALUES(NULL, 3, 'b');
INSERT INTO skills (id, user_id, name) VALUES(NULL, 3, 'c');

groups -* users -* skills

みたいな形の構造

使う

transaction気にしなければ以下の様な感じ。まじめに自分で定義したくなければmetadataのreflectを使う。

import sqlalchemy as sa


def run(url, *, echo):
    config = {"url": url}
    engine = sa.engine_from_config(config, prefix="")
    metadata = sa.MetaData(bind=engine)
    metadata.reflect(engine)
    engine.echo = echo

    with engine.connect() as conn:
        usedb(conn, metadata.tables)

def usedb(conn, tables):
    for row in conn.execute(tables["users"].select()):
        print(row)

url = "sqlite:///../src/groups.db"
run(url, echo=True)

2017-07-05 22:19:36,252 INFO sqlalchemy.engine.base.Engine SELECT users.id, users.group_id, users.name 
FROM users
2017-07-05 22:19:36,252 INFO sqlalchemy.engine.base.Engine ()
(1, 1, 'foo')
(2, 1, 'bar')
(3, 2, 'boo')

join

普通にjoinもできる。

qs = tables["users"].join(tables["skills"], tables["users"].c.id == tables["skills"].c.user_id)
for row in conn.execute(qs.select().where(tables["users"].c.id == 1)):
print(row)

(1, 1, 'foo', 1, 1, 'a')
(1, 1, 'foo', 2, 1, 'b')
(1, 1, 'foo', 3, 1, 'c')

select(field,…)

もちろん、select句を直接指定できる。

qs = tables["users"].join(tables["skills"], tables["users"].c.id == tables["skills"].c.user_id)
for row in conn.execute(
    sa.sql.select([tables["users"].c.name, tables["skills"].c.name]).select_from(qs).where(tables["users"].c.id == 1)
):
    print(row)

('foo', 'a')
('foo', 'b')
('foo', 'c')

fetch many

limitとは別に内部的なapiがbuffering的なものをサポートしていた場合にはfetchmanyが使える。

def chunked(cursor, *, n):
    while True:
        rows = cursor.fetchmany(n)
        if not rows:
            break
        yield rows


qs = tables["users"].join(tables["skills"], tables["users"].c.id == tables["skills"].c.user_id)
for rows in chunked(conn.execute(qs.select().where(tables["users"].c.id == 1)), n=2):
    print(rows)

[(1, 1, 'foo', 1, 1, 'a'), (1, 1, 'foo', 2, 1, 'b')]
[(1, 1, 'foo', 3, 1, 'c')]