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
|
from pytest import mark
from promise import Promise
from promise.dataloader import DataLoader
def id_loader(**options):
load_calls = []
resolve = options.pop("resolve", Promise.resolve)
def fn(keys):
load_calls.append(keys)
return resolve(keys)
identity_loader = DataLoader(fn, **options)
return identity_loader, load_calls
@mark.asyncio
async def test_await_dataloader():
identity_loader, load_calls = id_loader()
async def load_multiple(identity_loader):
one = identity_loader.load("load1")
two = identity_loader.load("load2")
return await Promise.all([one, two])
result = await load_multiple(identity_loader)
assert result == ["load1", "load2"]
assert load_calls == [["load1"], ["load2"]]
@mark.asyncio
async def test_await_dataloader_safe_promise():
identity_loader, load_calls = id_loader()
@Promise.safe
async def load_multiple(identity_loader):
one = identity_loader.load("load1")
two = identity_loader.load("load2")
return await Promise.all([one, two])
result = await load_multiple(identity_loader)
assert result == ["load1", "load2"]
assert load_calls == [["load1"], ["load2"]]
@mark.asyncio
async def test_await_dataloader_individual():
identity_loader, load_calls = id_loader()
async def load_one_then_two(identity_loader):
one = await identity_loader.load("load1")
two = await identity_loader.load("load2")
return [one, two]
result = await load_one_then_two(identity_loader)
assert result == ["load1", "load2"]
assert load_calls == [["load1"], ["load2"]]
@mark.asyncio
async def test_await_dataloader_individual_safe_promise():
identity_loader, load_calls = id_loader()
@Promise.safe
async def load_one_then_two(identity_loader):
one = await identity_loader.load("load1")
two = await identity_loader.load("load2")
return [one, two]
result = await load_one_then_two(identity_loader)
assert result == ["load1", "load2"]
assert load_calls == [["load1"], ["load2"]]
@mark.asyncio
async def test_await_dataloader_two():
identity_loader, load_calls = id_loader()
async def load_one_then_two(identity_loader):
one = await identity_loader.load("load1")
two = await identity_loader.load("load2")
return (one, two)
result12 = await Promise.all([load_one_then_two(identity_loader)])
@mark.asyncio
async def test_await_dataloader_two_safe_promise():
identity_loader, load_calls = id_loader()
@Promise.safe
async def load_one_then_two(identity_loader):
one = await identity_loader.load("load1")
two = await identity_loader.load("load2")
return (one, two)
result12 = await Promise.all([load_one_then_two(identity_loader)])
|
