/
04_relationships_tips.py
429 lines (357 loc) · 17.5 KB
/
04_relationships_tips.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
"""relationships に関する tips
ref: https://docs.sqlalchemy.org/en/20/orm/queryguide/relationships.html#relationship-loading-techniques
""" # noqa
import pytest
from sqlalchemy import select
from sqlalchemy.exc import InvalidRequestError
from sqlalchemy.orm import (
Load,
aliased,
contains_eager,
joinedload,
raiseload,
selectinload,
)
from sqlalchemy.orm.exc import DetachedInstanceError
import models_club # noqa
from db import Session
from models import Clazz, Email, Student, StudentClazz
def test_basic_relationship():
with Session() as session:
student1 = Student(name="name", gender=1, address="address", score=50)
# Student モデルに emails のリレーションを宣言しているのでアクセスできる (現状は空配列)
print(f"### student1.emails[{student1.emails}]")
# もちろん追加できる
email1 = Email(email="username1@example.com")
student1.emails.append(email1)
print(f"### student1.emails[{student1.emails}]")
# モデルで次のように宣言しているため、追加すると逆アクセスもできるようになる
# - Student モデルに emails のリレーションを宣言、および back_populates で依存先の属性(`student`)を宣言している
# - Email モデルに student のリレーションを宣言している
print(f"### email1.student[{email1.student}]")
# お試しとして Email モデルでは back_populates を宣言していないため、
# Email 側で追加しても逆アクセスはできない
email2 = Email(email="username2@example.com")
print(f"### email2.student[{email2.student}]")
email2.student = student1
print(f"### email2.student[{email2.student}]")
print(f"### student1.emails[{student1.emails}]") # email2 が追加されていない
print(f"### student1.id[{student1.id}]") # ここでは id は None
session.add(student1)
print(f"### student1 in session[{student1 in session}]")
print(f"### email1 in session[{email1 in session}]")
print(f"### email2 in session[{email2 in session}]")
session.commit()
# => student1 と email1 はコミットするが email2 はコミットしない
# commit するとオブジェクトをリフレッシュするため、オブジェクトの attribute にアクセスすると
# select クエリを発行して id を含めて再取得する
print(f"### student1.id[{student1.id}]")
# emails にもアクセスできるが遅延ロードとなる(アクセス時にクエリを発行する)
print(f"### student1.emails[{student1.emails}]")
def test_detached_instance_error():
"""コミット後に session 外でアクセスすると DetachedInstanceError となる
このため session 外でアクセスする場合には事前に session 内でアクセスしておくと良い。
"""
with Session() as session:
student = Student(name="name", gender=1, address="address", score=50)
session.add(student)
session.commit()
with pytest.raises(DetachedInstanceError) as e:
print(f"### student1.id[{student.id}]")
print(f"### ***ERROR*** [{e}]")
def test_join_by_relationship():
"""
ref: https://docs.sqlalchemy.org/en/20/tutorial/orm_related_objects.html#using-relationships-to-join
ref: https://docs.sqlalchemy.org/en/20/orm/queryguide/select.html#simple-relationship-joins
NOTE: join にリレーションを指定すると、自動的に外部キーで join してくれる
TODO: 外部キーがない場合はエラーになるのか?
""" # noqa
with Session() as session:
stmt = select(Student, StudentClazz).join(Student.clazz)
result = session.execute(stmt)
students = result.all()
for record in students:
# record が tuple なのでモデル名でアクセスできる
print(
f"### student.id[{record.Student.id}]"
f" student.name[{record.Student.name}]"
f" student_clazz.class_id[{record.StudentClazz.class_id}]"
)
# tuple が返るので展開して受け取ることもできる
for student, student_clazz in students:
print(
f"### student.id[{student.id}]"
f" student.name[{student.name}]"
f" student_clazz.class_id[{student_clazz.class_id}]"
)
def test_multi_join_by_relationship():
"""
ref: https://docs.sqlalchemy.org/en/20/orm/queryguide/select.html#chaining-multiple-joins
""" # noqa
with Session() as session:
stmt = select(Student, Clazz).join(Student.clazz).join(StudentClazz.clazz)
result = session.execute(stmt)
students = result.all()
for student, clazz in students:
print(
f"### student.id[{student.id}]"
f" student.name[{student.name}]"
f" clazz.id[{clazz.id}]"
f" clazz.name[{clazz.name}]"
)
def test_lazy_load():
"""joinやeagerロードを使わずに複数回のクエリを発行するケース
ローディングのデフォルト。
ref: https://docs.sqlalchemy.org/en/20/orm/queryguide/relationships.html#lazy-loading
NOTE: `student.clazz.class_id` を取得する際に select 文を毎度発行するため N+1 問題となるケース
""" # noqa
with Session() as session:
stmt = select(Student).limit(3)
result = session.execute(stmt)
students = result.scalars().all()
for student in students:
print(
f"### student.id[{student.id}]"
f" student.name[{student.name}]"
f" student.clazz.class_id[{student.clazz.class_id}]"
)
def test_selectin_load():
"""リレーション先を in 句で eager ロードする
次の2つの select を発行する。
- リレーション元の select
- リレーション元の ID を in 句にしたリレーション先の select
一対多、多対多では selectin load を推奨している。
ref: https://docs.sqlalchemy.org/en/20/tutorial/orm_related_objects.html#selectin-load
ref: https://docs.sqlalchemy.org/en/20/orm/queryguide/relationships.html#select-in-loading
ref: https://docs.sqlalchemy.org/en/20/orm/queryguide/relationships.html#what-kind-of-loading-to-use
ref: https://docs.sqlalchemy.org/en/20/orm/queryguide/relationships.html#sqlalchemy.orm.selectinload
NOTE: いくつかの制約あり。
- in 句を使うため上限制約を考慮する必要がある
- 複合主キーについては DB 依存
""" # noqa
with Session() as session:
stmt = select(Student).options(selectinload(Student.emails)).limit(3)
result = session.execute(stmt)
for student in result.scalars():
print(
f"### student.id[{student.id}]"
f" student.name[{student.name}]"
f" student.emails[{[email.email for email in student.emails]}]"
)
def test_joined_load_with_many_to_one_relationship():
"""多対一のリレーションに対する joined load による eager ロード
join によるロードのためクエリを1回発行する。
これは Email(多側) が基準のケース。
多対一 では joined load を推奨している。
ref: https://docs.sqlalchemy.org/en/20/tutorial/orm_related_objects.html#joined-load
ref: https://docs.sqlalchemy.org/en/20/orm/queryguide/relationships.html#joined-eager-loading
ref: https://docs.sqlalchemy.org/en/20/orm/queryguide/relationships.html#what-kind-of-loading-to-use
ref: https://docs.sqlalchemy.org/en/20/orm/queryguide/relationships.html#sqlalchemy.orm.joinedload
""" # noqa
with Session() as session:
stmt = (
select(Email).options(joinedload(Email.student, innerjoin=False)).limit(3)
)
result = session.execute(stmt)
for email in result.scalars().all():
print(
f"### student.id[{email.student.id}]"
f" student.name[{email.student.name}]"
f" student.emails[{email.email}]"
)
def test_joined_load_with_one_to_many_relationship():
"""一対多のリレーションに対する joined load による eager ロード
これは Student(一側) が基準のケース。
ref: https://docs.sqlalchemy.org/en/20/orm/queryguide/relationships.html#joined-eager-loading
NOTE: 一対多の関係を joined load する場合は `unique()` する必要がある。 (by queryguide の tips より)
理由は join で結合するため一側が複数行となるため。
`unique()` がないと `sqlalchemy.exc.InvalidRequestError` の例外となる
""" # noqa
with Session() as session:
stmt = (
select(Student)
.options(joinedload(Student.emails, innerjoin=False))
.limit(3)
)
result = session.execute(stmt)
for student in result.scalars().unique().all():
print(
f"### student.id[{student.id}]"
f" student.name[{student.name}]"
f" student.emails[{[email.email for email in student.emails]}]"
)
# `unique()` がない場合
result = session.execute(stmt)
with pytest.raises(InvalidRequestError) as e:
result.scalars().all()
print(f"### ***ERROR*** [{e}]")
def test_contains_eager():
"""join した結果を流用した eager ロード
ref: https://docs.sqlalchemy.org/en/20/tutorial/orm_related_objects.html#explicit-join-eager-load
ref: https://docs.sqlalchemy.org/en/20/orm/queryguide/relationships.html#routing-explicit-joins-statements-into-eagerly-loaded-collections
NOTE: join の際に `options(contains_eager(xxx))` の宣言がない場合、バッファリングされていない student が必要な際に
select のクエリを発行するため N+1 問題となり得る。
TODO: リレーション先をフィルタリングしたい場合などは contains_eager を使う感じ?
""" # noqa
with Session() as session:
stmt = (
select(Email)
.join(Email.student)
.options(contains_eager(Email.student))
.limit(3)
)
result = session.execute(stmt)
for email in result.scalars().all():
print(
f"### student.id[{email.student.id}]"
f" student.name[{email.student.name}]"
f" student.emails[{email.email}]"
)
def test_contains_eager_with_alias_and_join():
"""
お試しとして Email -> Student -> Email の Eager ロードを検証する。
結果として次のようなレコード群になるため `unique()` が必要となる。
[student1のeamil1, student1, student1のeamil1]
[student1のeamil1, student1, student1のeamil2]
[student1のeamil2, student1, student1のeamil1]
[student1のeamil2, student1, student1のeamil2]
ref: https://github.com/sqlalchemy/sqlalchemy/discussions/6876
ref: https://docs.sqlalchemy.org/en/20/orm/queryguide/select.html#using-relationship-to-join-between-aliased-targets
ref: https://docs.sqlalchemy.org/en/20/orm/internals.html#sqlalchemy.orm.PropComparator.of_type
""" # noqa
with Session() as session:
eamil_alias = aliased(Email)
stmt = (
select(Email)
.outerjoin(Email.student)
.outerjoin(Student.emails.of_type(eamil_alias))
.options(
contains_eager(Email.student).options(
contains_eager(Student.emails.of_type(eamil_alias))
)
)
)
result = session.execute(stmt)
for email in result.scalars().unique().all():
emails = [email.email for email in email.student.emails]
print(
f"### email.student.id[{email.student.id}]"
f" email.student.name[{email.student.name}]"
f" email.student.emails[{emails}]"
)
def test_contains_eager_with_subquery_and_alias_and_join():
"""生徒からのリレーションでメールアドレスを昇順で一番上にくるもののみを取得する
`03_select_tips.py::test_join_with_subquery_and_alias()` の応用版。
not exists とサブクエリを使って最新(or 最古)のデータのみを join するテクニック。
ref: https://takakisan.com/sql-max-in-each-group/
""" # noqa
with Session() as session:
mail_alias = aliased(Email)
mail_sub_query = (
select(Email)
.where(
~(
select(1)
.select_from(mail_alias)
.where(
mail_alias.student_id == Email.student_id,
mail_alias.email < Email.email,
)
).exists()
)
.subquery()
)
mail_alias = aliased(Email, mail_sub_query)
stmt = (
select(Student)
.join(Student.emails.of_type(mail_alias))
.options(contains_eager(Student.emails.of_type(mail_alias)))
)
for student in session.execute(stmt).scalars().unique():
emails = [email.email for email in student.emails]
print(
f"### student.id[{student.id}]"
f" student.name[{student.name}]"
f" student.emails[{emails}]"
)
def test_raiseload():
"""N+1 問題は絶対にゆるさんぞい
ref: https://docs.sqlalchemy.org/en/20/orm/queryguide/relationships.html#preventing-unwanted-lazy-loads-using-raiseload
""" # noqa
with Session() as session:
stmt = select(Student).options(joinedload(Student.clazz), raiseload("*"))
result = session.execute(stmt)
student = result.scalars().first()
print(f"### student.id[{student.id}]")
print(f"### student.name[{student.name}]")
# Student -> StudentClazz は joinedload しているので OK
print(f"### student.class[{student.clazz}]")
# Student -> StudentClazz -> Clazz は OUT
with pytest.raises(InvalidRequestError) as e:
print(f"### student.class.class[{student.clazz.clazz}]")
print(f"### ***ERROR*** [{e}]")
# Student -> Email は OUT
with pytest.raises(InvalidRequestError) as e:
print(f"### student.email[{student.emails}]")
print(f"### ***ERROR*** [{e}]")
def test_raiseload_特定の遅延ロード禁止を設定する():
with Session() as session:
# Student -> StudentClazz -> Clazz は OK
# Student -> Email は OUT
stmt = (
select(Student)
.options(joinedload(Student.clazz), Load(Student).raiseload("*"))
.limit(1)
)
result = session.execute(stmt)
student = result.scalars().first()
print(f"### student.class.class[{student.clazz.clazz}]")
with pytest.raises(InvalidRequestError) as e:
print(f"### student.email[{student.emails}]")
print(f"### ***ERROR*** [{e}]")
# Student に対する select の設定をクリアするために session を再発行
with Session() as session:
# Student -> StudentClazz -> Clazz は OUT
# Student -> Email は OK
stmt = (
select(Student).options(joinedload(Student.clazz).raiseload("*")).limit(1)
)
result = session.execute(stmt)
student = result.scalars().first()
print(f"### student.email[{student.emails}]")
with pytest.raises(InvalidRequestError) as e:
print(f"### student.class.class[{student.clazz.clazz}]")
print(f"### ***ERROR*** [{e}]")
def test_multi_joinedload():
"""
ref: https://docs.sqlalchemy.org/en/20/orm/queryguide/relationships.html#joined-eager-loading
""" # noqa
with Session() as session:
stmt = select(Student).options(
joinedload(Student.emails, innerjoin=False),
joinedload(Student.clazz, innerjoin=False),
)
result = session.execute(stmt)
student = result.scalar()
print(f"### student.id[{student.id}]")
print(f"### student.name[{student.name}]")
print(f"### student.emails[{[email.email for email in student.emails]}]")
print(f"### student.clazz.class_id[{student.clazz.class_id}]")
def test_eager_loading_specific_columns():
"""特定カラムの eager load
ref: https://docs.sqlalchemy.org/en/20/orm/queryguide/columns.html#using-load-only-on-related-objects-and-collections
""" # noqa
with Session() as session:
stmt = select(StudentClazz).options(
# raiseload を付与して name 以外の lazy load を禁止する
joinedload(StudentClazz.student).load_only(Student.name, raiseload=True)
)
result = session.execute(stmt)
student_clazz = result.scalar()
print(f"### student_clazz.class_id[{student_clazz.class_id}]")
print(f"### student_clazz.student_id[{student_clazz.student_id}]")
print(f"### student_clazz.student.id[{student_clazz.student.id}]")
print(f"### student_clazz.student.name[{student_clazz.student.name}]")
with pytest.raises(InvalidRequestError) as e:
print(f"### student_clazz.student.address[{student_clazz.student.address}]")
print(f"### ***ERROR*** [{e}]")