クエリビルダ
Peeweeの高レベルModel APIとField APIは、低レベルのTableとColumnに対応するものの上に構築されています。これらの低レベルAPIは、高レベルAPIほど詳細に文書化されていませんが、このドキュメントでは、実験できるようにするための概要と例を紹介します。
以下のスキーマを使用します。
CREATE TABLE "person" (
"id" INTEGER NOT NULL PRIMARY KEY,
"first" TEXT NOT NULL,
"last" TEXT NOT NULL);
CREATE TABLE "note" (
"id" INTEGER NOT NULL PRIMARY KEY,
"person_id" INTEGER NOT NULL,
"content" TEXT NOT NULL,
"timestamp" DATETIME NOT NULL,
FOREIGN KEY ("person_id") REFERENCES "person" ("id"));
CREATE TABLE "reminder" (
"id" INTEGER NOT NULL PRIMARY KEY,
"note_id" INTEGER NOT NULL,
"alarm" DATETIME NOT NULL,
FOREIGN KEY ("note_id") REFERENCES "note" ("id"));
テーブルの宣言
これらのテーブルを操作するためのTableオブジェクトを宣言するには、2つの方法があります。
# Explicitly declare columns
Person = Table('person', ('id', 'first', 'last'))
Note = Table('note', ('id', 'person_id', 'content', 'timestamp'))
# Do not declare columns, they will be accessed using magic ".c" attribute
Reminder = Table('reminder')
通常、テーブルをデータベースにbind()します。これにより、テーブルに対してクエリを実行するたびにデータベースを明示的に渡す必要がなくなります。
db = SqliteDatabase('my_app.db')
Person = Person.bind(db)
Note = Note.bind(db)
Reminder = Reminder.bind(db)
SELECTクエリ
最初の3つのメモを選択してその内容を出力するには、次のように記述します。
query = Note.select().order_by(Note.timestamp).limit(3)
for note_dict in query:
print(note_dict['content'])
注記
デフォルトでは、行は辞書として返されます。必要に応じて、tuples()、namedtuples()、またはobjects()メソッドを使用して、行データの異なるコンテナを指定できます。
列を指定していないため、メモのTableコンストラクタで定義したすべての列が選択されます。Reminderでは、列をまったく指定していないため、これは機能しません。
2018年に公開されたすべてのメモを作成者の名前とともに選択するには、join()を使用します。また、行を *namedtuple* オブジェクトとして返すように要求します。
query = (Note
.select(Note.content, Note.timestamp, Person.first, Person.last)
.join(Person, on=(Note.person_id == Person.id))
.where(Note.timestamp >= datetime.date(2018, 1, 1))
.order_by(Note.timestamp)
.namedtuples())
for row in query:
print(row.timestamp, '-', row.content, '-', row.first, row.last)
最も多くのメモを作成した人、つまり最も多作な人をクエリしてみましょう。これは、SQL関数（COUNT）の呼び出しを導入します。これは、fnオブジェクトを使用して実現されます。
name = Person.first.concat(' ').concat(Person.last)
query = (Person
.select(name.alias('name'), fn.COUNT(Note.id).alias('count'))
.join(Note, JOIN.LEFT_OUTER, on=(Note.person_id == Person.id))
.group_by(name)
.order_by(fn.COUNT(Note.id).desc()))
for row in query:
print(row['name'], row['count'])
上記のクエリには、注意すべき点がいくつかあります。
式を変数（
name）に格納し、それをクエリで使用します。SQL関数を
fn.<function>(...)を使用して呼び出し、通常のPython関数であるかのように引数を渡します。alias()メソッドは、列または計算に使用される名前を指定するために使用されます。
より複雑な例として、すべての人々と、彼らが最後に公開したメモの内容とタイムスタンプのリストを生成します。これを行うには、同じクエリ内で異なるコンテキストでNoteテーブルを2回使用することになります。そのため、テーブルエイリアスを使用する必要があります。
# Start with the query that calculates the timestamp of the most recent
# note for each person.
NA = Note.alias('na')
max_note = (NA
.select(NA.person_id, fn.MAX(NA.timestamp).alias('max_ts'))
.group_by(NA.person_id)
.alias('max_note'))
# Now we'll select from the note table, joining on both the subquery and
# on the person table to construct the result set.
query = (Note
.select(Note.content, Note.timestamp, Person.first, Person.last)
.join(max_note, on=((max_note.c.person_id == Note.person_id) &
(max_note.c.max_ts == Note.timestamp)))
.join(Person, on=(Note.person_id == Person.id))
.order_by(Person.first, Person.last))
for row in query.namedtuples():
print(row.first, row.last, ':', row.timestamp, '-', row.content)
*max_note* サブクエリへの結合の結合述語では、魔法の ".c" 属性を使用してサブクエリの列を参照できます。したがって、 *max_note.c.max_ts* は "max_noteサブクエリからのmax_ts列の値" に変換されます。
また、 ".c" マジック属性を使用して、Reminderテーブルで行ったように、列を明示的に定義していないテーブルの列にアクセスすることもできます。今日のすべてのリマインダーと、それに関連付けられたメモの内容を取得するための簡単なクエリを次に示します。
today = datetime.date.today()
tomorrow = today + datetime.timedelta(days=1)
query = (Reminder
.select(Reminder.c.alarm, Note.content)
.join(Note, on=(Reminder.c.note_id == Note.id))
.where(Reminder.c.alarm.between(today, tomorrow))
.order_by(Reminder.c.alarm))
for row in query:
print(row['alarm'], row['content'])
注記
混乱を防ぐため、 ".c" 属性は列を明示的に定義するテーブルでは機能しません。
INSERTクエリ
データの挿入は簡単です。insert() に指定するデータは、2つの異なる方法で指定できます（どちらの場合も、新しい行のIDが返されます）。
# Using keyword arguments:
zaizee_id = Person.insert(first='zaizee', last='cat').execute()
# Using column: value mappings:
Note.insert({
Note.person_id: zaizee_id,
Note.content: 'meeeeowwww',
Note.timestamp: datetime.datetime.now()}).execute()
一括挿入は簡単で、次のいずれかを渡すだけです。
辞書のリスト（すべて同じキー/列を持っている必要があります）。
列が明示的に指定されている場合は、タプルのリスト。
例
people = [
{'first': 'Bob', 'last': 'Foo'},
{'first': 'Herb', 'last': 'Bar'},
{'first': 'Nuggie', 'last': 'Bar'}]
# Inserting multiple rows returns the ID of the last-inserted row.
last_id = Person.insert(people).execute()
# We can also specify row tuples, so long as we tell Peewee which
# columns the tuple values correspond to:
people = [
('Bob', 'Foo'),
('Herb', 'Bar'),
('Nuggie', 'Bar')]
Person.insert(people, columns=[Person.first, Person.last]).execute()
UPDATEクエリ
update()クエリは、insert()と同様に、キーワード引数または列から値へのマッピング辞書を受け入れます。
例
# "Bob" changed his last name from "Foo" to "Baze".
nrows = (Person
.update(last='Baze')
.where((Person.first == 'Bob') &
(Person.last == 'Foo'))
.execute())
# Use dictionary mapping column to value.
nrows = (Person
.update({Person.last: 'Baze'})
.where((Person.first == 'Bob') &
(Person.last == 'Foo'))
.execute())
また、式を値として使用して、アトミック更新を実行することもできます。 *PageView* テーブルがあり、一部のURLのページビューカウントをアトミックにインクリメントする必要があるとします。
# Do an atomic update:
(PageView
.update({PageView.count: PageView.count + 1})
.where(PageView.url == some_url)
.execute())
DELETEクエリ
delete()クエリは、引数を受け入れないため、最も単純です。
# Delete all notes created before 2018, returning number deleted.
n = Note.delete().where(Note.timestamp < datetime.date(2018, 1, 1)).execute()
DELETE（およびUPDATE）クエリは結合をサポートしていないため、サブクエリを使用して関連テーブルの値に基づいて行を削除できます。たとえば、姓が "Foo" であるすべての人のメモを削除する方法は次のとおりです。
# Get the id of all people whose last name is "Foo".
foo_people = Person.select(Person.id).where(Person.last == 'Foo')
# Delete all notes by any person whose ID is in the previous query.
Note.delete().where(Note.person_id.in_(foo_people)).execute()
クエリオブジェクト
Peewee 2.xによって提供される抽象化の基本的な制限事項の1つは、特定のモデルクラスとの関係を持たない構造化クエリを表すクラスがないことでした。
この例としては、サブクエリに対する集計値の計算が挙げられます。たとえば、任意のクエリの行数を返すcount()メソッドは、クエリをラップすることで実装されます。
SELECT COUNT(1) FROM (...)
Peeweeでこれを実現するために、実装はこのように記述されています。
def count(query):
# Select([source1, ... sourcen], [column1, ...columnn])
wrapped = Select(from_list=[query], columns=[fn.COUNT(SQL('1'))])
curs = wrapped.tuples().execute(db)
return curs[0][0] # Return first column from first row of result.
実際には、集計クエリから値を返すのに適したscalar()メソッドを使用して、これをより簡潔に表現できます。
def count(query):
wrapped = Select(from_list=[query], columns=[fn.COUNT(SQL('1'))])
return wrapped.scalar(db)
クエリ例ドキュメントには、予約済みの利用可能なスロット数が最も多い施設のクエリを記述する、より複雑な例があります。
表現したいSQLは次のとおりです。
SELECT facid, total FROM (
SELECT facid, SUM(slots) AS total,
rank() OVER (order by SUM(slots) DESC) AS rank
FROM bookings
GROUP BY facid
) AS ranked
WHERE rank = 1
外部クエリにプレーンSelectを使用することで、これをかなりエレガントに表現できます。
# Store rank expression in variable for readability.
rank_expr = fn.rank().over(order_by=[fn.SUM(Booking.slots).desc()])
subq = (Booking
.select(Booking.facility, fn.SUM(Booking.slots).alias('total'),
rank_expr.alias('rank'))
.group_by(Booking.facility))
# Use a plain "Select" to create outer query.
query = (Select(columns=[subq.c.facid, subq.c.total])
.from_(subq)
.where(subq.c.rank == 1)
.tuples())
# Iterate over the resulting facility ID(s) and total(s):
for facid, total in query.execute(db):
print(facid, total)
別の例として、再帰的な共通テーブル式を作成して、最初の10個のフィボナッチ数を計算してみましょう。
base = Select(columns=(
Value(1).alias('n'),
Value(0).alias('fib_n'),
Value(1).alias('next_fib_n'))).cte('fibonacci', recursive=True)
n = (base.c.n + 1).alias('n')
recursive_term = Select(columns=(
n,
base.c.next_fib_n,
base.c.fib_n + base.c.next_fib_n)).from_(base).where(n < 10)
fibonacci = base.union_all(recursive_term)
query = fibonacci.select_from(fibonacci.c.n, fibonacci.c.fib_n)
results = list(query.execute(db))
# Generates the following result list:
[{'fib_n': 0, 'n': 1},
{'fib_n': 1, 'n': 2},
{'fib_n': 1, 'n': 3},
{'fib_n': 2, 'n': 4},
{'fib_n': 3, 'n': 5},
{'fib_n': 5, 'n': 6},
{'fib_n': 8, 'n': 7},
{'fib_n': 13, 'n': 8},
{'fib_n': 21, 'n': 9},
{'fib_n': 34, 'n': 10}]
その他
SQL ASTを記述するために使用されるさまざまなクラスの説明については、クエリビルダAPIドキュメントを参照してください。
詳細を知りたい場合は、プロジェクトのソースコードもご覧ください。