Coverage for databases/tests/test_batch.py: 100%

1import pytest 

2 

3import prisma 

4from prisma import Prisma 

5 

6from ..utils import RawQueries, DatabaseConfig 

7 

8 

9@pytest.mark.asyncio 

10async def test_base_usage(client: Prisma) -> None: 

11 """Basic non context manager usage""" 

12 batcher = client.batch_() 

13 batcher.user.create({'name': 'Robert'}) 

14 batcher.user.create({'name': 'Tegan'}) 

15 await batcher.commit() 

16 

17 user = await client.user.find_first(where={'name': 'Robert'}) 

18 assert user is not None 

19 assert user.name == 'Robert' 

20 

21 user = await client.user.find_first(where={'name': 'Tegan'}) 

22 assert user is not None 

23 assert user.name == 'Tegan' 

24 

25 

26@pytest.mark.asyncio 

27async def test_context_manager(client: Prisma) -> None: 

28 """Basic usage with a context manager""" 

29 async with client.batch_() as batcher: 

30 batcher.user.create({'name': 'Robert'}) 

31 batcher.user.create({'name': 'Tegan'}) 

32 

33 user = await client.user.find_first(where={'name': 'Robert'}) 

34 assert user is not None 

35 assert user.name == 'Robert' 

36 

37 user = await client.user.find_first(where={'name': 'Tegan'}) 

38 assert user is not None 

39 assert user.name == 'Tegan' 

40 

41 

42@pytest.mark.asyncio 

43async def test_batch_error(client: Prisma) -> None: 

44 """Error while committing does not commit any records""" 

45 with pytest.raises(prisma.errors.UniqueViolationError) as exc: 

46 batcher = client.batch_() 

47 batcher.user.create({'id': 'abc', 'name': 'Robert'}) 

48 batcher.user.create({'id': 'abc', 'name': 'Robert 2'}) 

49 await batcher.commit() 

50 

51 assert exc.match(r'Unique constraint failed') 

52 assert await client.user.count() == 0 

53 

54 

55@pytest.mark.asyncio 

56async def test_context_manager_error(client: Prisma) -> None: 

57 """Error exiting context manager does not commit any records""" 

58 with pytest.raises(prisma.errors.UniqueViolationError) as exc: 

59 async with client.batch_() as batcher: 

60 batcher.user.create({'id': 'abc', 'name': 'Robert'}) 

61 batcher.user.create({'id': 'abc', 'name': 'Robert 2'}) 

62 

63 assert exc.match(r'Unique constraint failed') 

64 assert await client.user.count() == 0 

65 

66 

67@pytest.mark.asyncio 

68async def test_context_manager_throws_error(client: Prisma) -> None: 

69 """Context manager respects errors""" 

70 with pytest.raises(RuntimeError) as exc: 

71 async with client.batch_() as batcher: 

72 batcher.user.create({'name': 'Robert'}) 

73 raise RuntimeError('Example error') 

74 

75 assert exc.match('Example error') 

76 assert await client.user.count() == 0 

77 

78 

79@pytest.mark.asyncio 

80async def test_mixing_models(client: Prisma) -> None: 

81 """Batching queries to multiple models works as intended""" 

82 async with client.batch_() as batcher: 

83 # NOTE: this is just to test functionality, the better method 

84 # for acheiving this is to use nested writes with user.create 

85 # client.user.create({'name': 'Robert', 'profile': {'create': {'bio': 'Robert\'s profile'}}}) 

86 batcher.user.create({'id': 'abc', 'name': 'Robert'}) 

87 batcher.profile.create( 

88 { 

89 'user': {'connect': {'id': 'abc'}}, 

90 'description': "Robert's profile", 

91 'country': 'Scotland', 

92 } 

93 ) 

94 

95 user = await client.user.find_first(where={'name': 'Robert'}, include={'profile': True}) 

96 assert user is not None 

97 assert user.name == 'Robert' 

98 assert user.profile is not None 

99 assert user.profile.description == "Robert's profile" 

100 

101 assert await client.user.count() == 1 

102 assert await client.profile.count() == 1 

103 

104 

105@pytest.mark.asyncio 

106async def test_mixing_actions(client: Prisma) -> None: 

107 """Batching queries to different operations works as intended""" 

108 async with client.batch_() as batcher: 

109 batcher.user.create({'name': 'Robert'}) 

110 batcher.user.delete_many(where={'name': 'Robert'}) 

111 

112 assert await client.user.count() == 0 

113 

114 

115@pytest.mark.asyncio 

116async def test_reusing_batcher(client: Prisma) -> None: 

117 """Reusing the same batcher does not commit the same query multiple times""" 

118 batcher = client.batch_() 

119 batcher.user.create({'name': 'Robert'}) 

120 await batcher.commit() 

121 

122 assert await client.user.count() == 1 

123 

124 batcher.user.create({'name': 'Robert 2'}) 

125 await batcher.commit() 

126 

127 assert await client.user.count() == 2 

128 

129 

130@pytest.mark.asyncio 

131async def test_large_query(client: Prisma) -> None: 

132 """Batching a lot of queries works""" 

133 async with client.batch_() as batcher: 

134 for i in range(1000): 

135 batcher.user.create({'name': f'User {i}'}) 

136 

137 assert await client.user.count() == 1000 

138 

139 

140@pytest.mark.asyncio 

141async def test_delete(client: Prisma) -> None: 

142 """delete action works as suggested""" 

143 user = await client.user.create({'name': 'Robert'}) 

144 assert await client.user.find_first(where={'id': user.id}) is not None 

145 

146 async with client.batch_() as batcher: 

147 batcher.user.delete(where={'id': user.id}) 

148 

149 assert await client.user.find_first(where={'id': user.id}) is None 

150 

151 

152@pytest.mark.asyncio 

153async def test_update(client: Prisma) -> None: 

154 """update action works as suggested""" 

155 user = await client.user.create({'name': 'Robert'}) 

156 assert await client.user.find_first(where={'id': user.id}) is not None 

157 

158 async with client.batch_() as batcher: 

159 batcher.user.update(where={'id': user.id}, data={'name': 'Roberto'}) 

160 

161 new = await client.user.find_first(where={'id': user.id}) 

162 assert new is not None 

163 assert new.id == user.id 

164 assert new.name == 'Roberto' 

165 

166 

167@pytest.mark.asyncio 

168async def test_upsert(client: Prisma) -> None: 

169 """upsert action works as suggested""" 

170 user_id = 'abc123' 

171 assert await client.user.find_unique(where={'id': user_id}) is None 

172 

173 async with client.batch_() as batcher: 

174 batcher.user.upsert( 

175 where={'id': user_id}, 

176 data={ 

177 'create': {'id': user_id, 'name': 'Robert'}, 

178 'update': {'name': 'Robert'}, 

179 }, 

180 ) 

181 

182 user = await client.user.find_unique(where={'id': user_id}) 

183 assert user is not None 

184 assert user.id == user_id 

185 assert user.name == 'Robert' 

186 

187 async with client.batch_() as batcher: 

188 batcher.user.upsert( 

189 where={'id': user_id}, 

190 data={ 

191 'create': {'id': user_id, 'name': 'Robert'}, 

192 'update': {'name': 'Roberto'}, 

193 }, 

194 ) 

195 

196 user = await client.user.find_unique(where={'id': user_id}) 

197 assert user is not None 

198 assert user.id == user_id 

199 assert user.name == 'Roberto' 

200 assert await client.user.count() == 1 

201 

202 

203@pytest.mark.asyncio 

204async def test_update_many(client: Prisma) -> None: 

205 """update_many action works as suggested""" 

206 await client.user.create({'name': 'Robert'}) 

207 await client.user.create({'name': 'Robert 2'}) 

208 

209 async with client.batch_() as batcher: 

210 batcher.user.update_many(where={'name': {'startswith': 'Robert'}}, data={'name': 'Robert'}) 

211 

212 users = await client.user.find_many() 

213 assert len(users) == 2 

214 assert users[0].name == 'Robert' 

215 assert users[1].name == 'Robert' 

216 

217 

218@pytest.mark.asyncio 

219async def test_delete_many(client: Prisma) -> None: 

220 """delete_many action works as suggested""" 

221 await client.user.create({'name': 'Robert'}) 

222 await client.user.create({'name': 'Robert 2'}) 

223 assert await client.user.count() == 2 

224 

225 async with client.batch_() as batcher: 

226 batcher.user.delete_many(where={'name': {'startswith': 'Robert'}}) 

227 

228 assert await client.user.count() == 0 

229 

230 

231@pytest.mark.asyncio 

232async def test_execute_raw(client: Prisma, raw_queries: RawQueries) -> None: 

233 """execute_raw action can be used to execute raw SQL queries""" 

234 post1 = await client.post.create( 

235 { 

236 'title': 'My first post!', 

237 'published': False, 

238 } 

239 ) 

240 post2 = await client.post.create( 

241 { 

242 'title': 'My 2nd post.', 

243 'published': False, 

244 } 

245 ) 

246 

247 async with client.batch_() as batcher: 

248 batcher.execute_raw( 

249 raw_queries.update_unique_post_title, 

250 post1.id, 

251 ) 

252 batcher.execute_raw( 

253 raw_queries.update_unique_post_new_title, 

254 post2.id, 

255 ) 

256 

257 found = await client.post.find_unique(where={'id': post1.id}) 

258 assert found is not None 

259 assert found.id == post1.id 

260 assert found.title == 'My edited title' 

261 

262 found = await client.post.find_unique(where={'id': post2.id}) 

263 assert found is not None 

264 assert found.id == post2.id 

265 assert found.title == 'My new title' 

266 

267 

268@pytest.mark.asyncio 

269async def test_create_many_skip_duplicates_unsupported( 

270 client: Prisma, 

271 config: DatabaseConfig, 

272) -> None: 

273 """Cannot call create_many with skip_duplicates on databases that do not support it""" 

274 if 'create_many_skip_duplicates' not in config.unsupported_features: 

275 pytest.skip('The create_many skip_duplicates argument is supported by the current database') 

276 

277 with pytest.raises(prisma.errors.UnsupportedDatabaseError) as exc: 

278 async with client.batch_() as batcher: 

279 batcher.user.create_many([{'name': 'Robert'}], skip_duplicates=True) 

280 

281 assert exc.match(r'skip_duplicates is not supported')