Coverage for docs_src / tutorial / transaction / tutorial000.py: 86%

1import os 6 ctx1abcdef

2from typing import Optional 6 ctx1abcdef

3 

4from mongodb_odm import Document, connect 6 ctx1abcdef

5from pymongo.errors import OperationFailure 6 ctx1abcdef

6 

7 

8class Player(Document): 6 ctx1abcdef

9 name: str 6 ctx1abcdef

10 country_code: str 6 ctx1abcdef

11 rating: Optional[int] = None 6 ctx1abcdef

12 

13 

14def create_documents(): 6 ctx1abcdef

15 with Document.start_session() as session: 6 ctx1abcdef

16 with session.start_transaction(): 6 ctx1abcdef

17 try: 6 ctx1abcdef

18 Player( 6 ctx1abcdef

19 name="Pelé", 

20 country_code="BRA", 

21 ).create(session=session) 

22 

23 maradona = Player(name="Diego Maradona", country_code="ARG") 6 ctx1abcdef

24 maradona.create(session=session) 6 ctx1abcdef

25 except OperationFailure: 

26 session.abort_transaction() 

27 

28 

29def update_documents(): 6 ctx1abcdef

30 with Document.start_session() as session: 6 ctx1abcdef

31 with session.start_transaction(): 6 ctx1abcdef

32 try: 6 ctx1abcdef

33 pele = Player.get({Player.name: "Pelé"}) 6 ctx1abcdef

34 pele.rating = 98 6 ctx1abcdef

35 pele.update(session=session) 6 ctx1abcdef

36 

37 maradona = Player.get({Player.name: "Diego Maradona"}) 6 ctx1abcdef

38 pele.rating = 97 6 ctx1abcdef

39 maradona.update(session=session) 6 ctx1abcdef

40 except OperationFailure: 

41 session.abort_transaction() 

42 

43 

44def delete_documents(): 6 ctx1abcdef

45 with Document.start_session() as session: 6 ctx1abcdef

46 with session.start_transaction(): 6 ctx1abcdef

47 try: 6 ctx1abcdef

48 pele = Player.get({Player.name: "Pelé"}) 6 ctx1abcdef

49 pele.delete(session=session) 6 ctx1abcdef

50 

51 maradona = Player.get({Player.name: "Diego Maradona"}) 6 ctx1abcdef

52 maradona.delete(session=session) 6 ctx1abcdef

53 except OperationFailure: 

54 session.abort_transaction() 

55 

56 

57def main(): 6 ctx1abcdef

58 # Use a database that has a replica set. 

59 connect(os.environ.get("MONGO_URL", "mongodb://localhost:27017/testdb")) 6 ctx1abcdef

60 

61 create_documents() 6 ctx1abcdef

62 update_documents() 6 ctx1abcdef

63 delete_documents() 6 ctx1abcdef

64 

65 

66if __name__ == "__main__": 6 ctx1abcdef

67 main()