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

1import pytest 

2 

3import prisma 

4from prisma import Prisma 

5 

6from ..utils import RawQueries, DatabaseConfig 

7 

8 

9def test_base_usage(client: Prisma) -> None: 

10 """Basic non context manager usage""" 

11 batcher = client.batch_() 

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

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

14 batcher.commit() 

15 

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

17 assert user is not None 

18 assert user.name == 'Robert' 

19 

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

21 assert user is not None 

22 assert user.name == 'Tegan' 

23 

24 

25def test_context_manager(client: Prisma) -> None: 

26 """Basic usage with a context manager""" 

27 with client.batch_() as batcher: 

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

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

30 

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

32 assert user is not None 

33 assert user.name == 'Robert' 

34 

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

36 assert user is not None 

37 assert user.name == 'Tegan' 

38 

39 

40def test_batch_error(client: Prisma) -> None: 

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

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

43 batcher = client.batch_() 

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

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

46 batcher.commit() 

47 

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

49 assert client.user.count() == 0 

50 

51 

52def test_context_manager_error(client: Prisma) -> None: 

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

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

55 with client.batch_() as batcher: 

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

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

58 

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

60 assert client.user.count() == 0 

61 

62 

63def test_context_manager_throws_error(client: Prisma) -> None: 

64 """Context manager respects errors""" 

65 with pytest.raises(RuntimeError) as exc: 

66 with client.batch_() as batcher: 

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

68 raise RuntimeError('Example error') 

69 

70 assert exc.match('Example error') 

71 assert client.user.count() == 0 

72 

73 

74def test_mixing_models(client: Prisma) -> None: 

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

76 with client.batch_() as batcher: 

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

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

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

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

81 batcher.profile.create( 

82 { 

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

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

85 'country': 'Scotland', 

86 } 

87 ) 

88 

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

90 assert user is not None 

91 assert user.name == 'Robert' 

92 assert user.profile is not None 

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

94 

95 assert client.user.count() == 1 

96 assert client.profile.count() == 1 

97 

98 

99def test_mixing_actions(client: Prisma) -> None: 

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

101 with client.batch_() as batcher: 

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

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

104 

105 assert client.user.count() == 0 

106 

107 

108def test_reusing_batcher(client: Prisma) -> None: 

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

110 batcher = client.batch_() 

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

112 batcher.commit() 

113 

114 assert client.user.count() == 1 

115 

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

117 batcher.commit() 

118 

119 assert client.user.count() == 2 

120 

121 

122def test_large_query(client: Prisma) -> None: 

123 """Batching a lot of queries works""" 

124 with client.batch_() as batcher: 

125 for i in range(1000): 

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

127 

128 assert client.user.count() == 1000 

129 

130 

131def test_delete(client: Prisma) -> None: 

132 """delete action works as suggested""" 

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

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

135 

136 with client.batch_() as batcher: 

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

138 

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

140 

141 

142def test_update(client: Prisma) -> None: 

143 """update action works as suggested""" 

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

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

146 

147 with client.batch_() as batcher: 

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

149 

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

151 assert new is not None 

152 assert new.id == user.id 

153 assert new.name == 'Roberto' 

154 

155 

156def test_upsert(client: Prisma) -> None: 

157 """upsert action works as suggested""" 

158 user_id = 'abc123' 

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

160 

161 with client.batch_() as batcher: 

162 batcher.user.upsert( 

163 where={'id': user_id}, 

164 data={ 

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

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

167 }, 

168 ) 

169 

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

171 assert user is not None 

172 assert user.id == user_id 

173 assert user.name == 'Robert' 

174 

175 with client.batch_() as batcher: 

176 batcher.user.upsert( 

177 where={'id': user_id}, 

178 data={ 

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

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

181 }, 

182 ) 

183 

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

185 assert user is not None 

186 assert user.id == user_id 

187 assert user.name == 'Roberto' 

188 assert client.user.count() == 1 

189 

190 

191def test_update_many(client: Prisma) -> None: 

192 """update_many action works as suggested""" 

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

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

195 

196 with client.batch_() as batcher: 

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

198 

199 users = client.user.find_many() 

200 assert len(users) == 2 

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

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

203 

204 

205def test_delete_many(client: Prisma) -> None: 

206 """delete_many action works as suggested""" 

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

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

209 assert client.user.count() == 2 

210 

211 with client.batch_() as batcher: 

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

213 

214 assert client.user.count() == 0 

215 

216 

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

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

219 post1 = client.post.create( 

220 { 

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

222 'published': False, 

223 } 

224 ) 

225 post2 = client.post.create( 

226 { 

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

228 'published': False, 

229 } 

230 ) 

231 

232 with client.batch_() as batcher: 

233 batcher.execute_raw( 

234 raw_queries.update_unique_post_title, 

235 post1.id, 

236 ) 

237 batcher.execute_raw( 

238 raw_queries.update_unique_post_new_title, 

239 post2.id, 

240 ) 

241 

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

243 assert found is not None 

244 assert found.id == post1.id 

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

246 

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

248 assert found is not None 

249 assert found.id == post2.id 

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

251 

252 

253def test_create_many_skip_duplicates_unsupported( 

254 client: Prisma, 

255 config: DatabaseConfig, 

256) -> None: 

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

258 if 'create_many_skip_duplicates' not in config.unsupported_features: 

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

260 

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

262 with client.batch_() as batcher: 

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

264 

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