Coverage for django_napse/core/models/connections/connection.py: 58%

1from django.db import models 

2 

3from django_napse.core.models.connections.managers import ConnectionManager 

4from django_napse.core.models.transactions.transaction import Transaction 

5from django_napse.utils.constants import TRANSACTION_TYPES 

6from django_napse.utils.usefull_functions import process_value_from_type 

7 

8 

9class Connection(models.Model): 

10 """Link between a bot & a wallet.""" 

11 

12 owner = models.ForeignKey("Wallet", on_delete=models.CASCADE, related_name="connections") 

13 bot = models.ForeignKey("Bot", on_delete=models.CASCADE, related_name="connections") 

14 

15 created_at = models.DateTimeField(auto_now_add=True, blank=True) 

16 

17 objects = ConnectionManager() 

18 

19 class Meta: # noqa: D106 

20 unique_together = ("owner", "bot") 

21 

22 def __str__(self) -> str: # pragma: no cover 

23 return f"CONNECTION: {self.pk=}" 

24 

25 def info(self, verbose: bool = True, beacon: str = "") -> str: # noqa: FBT001, FBT002 

26 """Give info on the Connection instance.""" 

27 string = "" 

28 string += f"{beacon}Connection {self.pk}:\n" 

29 string += f"{beacon}Args:\n" 

30 string += f"{beacon}\t{self.owner=}\n" 

31 string += f"{beacon}\t{self.bot=}\n" 

32 string += f"{beacon}\t{self.created_at=}\n" 

33 

34 string += f"{beacon}Wallet:\n" 

35 new_beacon = beacon + "\t" 

36 string += f"{self.wallet.info(verbose=False, beacon=new_beacon)}\n" 

37 

38 string += f"{beacon}ConnectionSpecificArgs:\n" 

39 query = self.specific_args.all() 

40 if query.count() == 0: 

41 string += f"{beacon}\tNone\n" 

42 else: 

43 for connection_specific_arg in query: 

44 string += f"{connection_specific_arg.info(verbose=False, beacon=new_beacon)}\n" 

45 

46 if verbose: # pragma: no cover 

47 print(string) 

48 return string 

49 

50 @property 

51 def testing(self) -> bool: 

52 """Return true if the connection's space is in testing.""" 

53 return self.space.testing 

54 

55 @property 

56 def space(self) -> "NapseSpace": # noqa: F821 

57 """Return the space of the connection.""" 

58 return self.owner.find().space 

59 

60 def deposit(self, ticker: str, amount: int) -> Transaction: 

61 """Make a transaction from owner's wallet to bot's wallet.""" 

62 return Transaction.objects.create( 

63 from_wallet=self.owner, 

64 to_wallet=self.wallet, 

65 amount=amount, 

66 ticker=ticker, 

67 transaction_type=TRANSACTION_TYPES.CONNECTION_DEPOSIT, 

68 ) 

69 

70 def withdraw(self, ticker: str, amount: int) -> Transaction: 

71 """Make a transaction from bot's wallet to owner's wallet.""" 

72 return Transaction.objects.create( 

73 from_wallet=self.wallet, 

74 to_wallet=self.owner, 

75 amount=amount, 

76 ticker=ticker, 

77 transaction_type=TRANSACTION_TYPES.CONNECTION_WITHDRAW, 

78 ) 

79 

80 def to_dict(self) -> dict[str, any]: 

81 """Return a connection to a dictionary.""" 

82 return { 

83 "connection": self, 

84 "wallet": self.wallet.find().to_dict(), 

85 "connection_specific_args": {arg.key: arg for arg in self.specific_args.all()}, 

86 } 

87 

88 

89class ConnectionSpecificArgs(models.Model): 

90 """Argument of a connection. 

91 

92 A connection can have a infinite number of arguments. 

93 """ 

94 

95 connection = models.ForeignKey("Connection", on_delete=models.CASCADE, related_name="specific_args") 

96 key = models.CharField(max_length=100) 

97 value = models.CharField(max_length=100, default="None") 

98 target_type = models.CharField(max_length=100, default="None") 

99 

100 class Meta: # noqa: D106 

101 unique_together = ("connection", "key") 

102 

103 def __str__(self) -> str: # pragma: no cover 

104 string = f"CONNECTION_SPECIFIC_ARGS: {self.pk=}," 

105 return string + f"connection__pk={self.connection.pk}, key={self.key}, value={self.value}, target_type={self.target_type}" 

106 

107 def info(self, verbose: bool = True, beacon: str = "") -> str: # noqa: FBT001, FBT002 

108 """Give info on the ConnectionSpecificArgs instance.""" 

109 string = "" 

110 string += f"{beacon}ConnectionSpecificArgs {self.pk}:\n" 

111 string += f"{beacon}Args:\n" 

112 string += f"{beacon}\t{self.connection=}\n" 

113 string += f"{beacon}\t{self.key=}\n" 

114 string += f"{beacon}\t{self.value=}\n" 

115 string += f"{beacon}\t{self.target_type=}\n" 

116 if verbose: # pragma: no cover 

117 print(string) 

118 return string 

119 

120 def get_value(self) -> any: 

121 """Return the value in the correct type.""" 

122 return process_value_from_type(self.value, self.target_type)