Створюємо власне Web App для Telegram-бота

 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

class StartWebAppAPIView(generics.RetrieveAPIView):
   """
   A `StartWebAppAPIView` class for retrieve information(such as total balance,
   balance by cryptocurrencies and transaction history) about a user and retrieve or create new balances.
   """

   queryset = Profile.objects.all()
   serializer_class = ProfileSerializer
   lookup_field = "telegram_id"

   def get_object(self) -> Profile:
       """ Returns a `Profile object` or creates a new one if it does not exist. """

       object_ = Profile.objects.filter(telegram_id=self.kwargs.get("telegram_id"))
       if object_.exists():
           return object_.first()
       else:
           object_ = Profile.objects.create(
               telegram_id=self.kwargs.get("telegram_id"),
               first_name=self.request.data.get("message", {}).get("from", {}).get("first_name", "No name"),
               last_name=self.request.data.get("message", {}).get("from", {}).get("last_name", "No last name"),
               user_name=self.request.data.get("message", {}).get("from", {}).get("username", "No username"),
               client_language=self.request.data.get("message", {}).get("from", {}).get("client_language", "uk"),
           )

           tokens = Token.objects.all().order_by("priority_level")
           for token in tokens:
               Balance.objects.create(
                   profile=object_,
                   token=token,
                   network=token.network.first(),
                   amount=0
               )

           return object_

   def retrieve(self, request, *args, **kwargs) -> Response:
       """
       Returns the user's serialized data, his total balance converted to USDT,
       balance by cryptocurrencies, and his transaction history.
       """

       user: Profile = self.get_object()
       _user_balances: Balance = user.balance_set.all()

       serializer: ProfileSerializer = self.get_serializer(instance=user)

       total_user_balance: float = UserHelper.get_total_balance_in_usdt(user_balances=_user_balances)
       balances_by_cryptocurrencies: list[dict] = UserHelper.get_balances_by_cryptocurrencies(
           user_balances=_user_balances
       )
       transactions_history: list[dict] = UserHelper.get_transaction_history(user=user)

       return Response(
           data={
               "user": serializer.data,
               "user_balance": total_user_balance,
               "token_name": "USDT",
               "cryptocurrency": balances_by_cryptocurrencies,
               "transaction_history": transactions_history
           },
           status=status.HTTP_200_OK
       )

 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

class ETHStrategy(BlockchainStrategy):
	def __init__(self, network: str):
    	self._blockchain = Blockchain.objects.get(symbol="ETH")
    	self._network = Network.objects.get(blockchain=self._blockchain, network_type=network)
    	self._network_type = network
    	self._chain_id = self._network.chain_id
    	self.web3 = Web3(Web3.HTTPProvider(endpoint_uri=self._network.url_rpc))

	def create_wallet(self):
    	new_account = self.web3.eth.account.create()
    	return new_account.address, new_account.key.hex()

	def _create_tx(self, from_address: Address, to_address: Address, amount: Decimal) -> dict:
    	tx = {
        	'from': from_address,
        	'to': to_address,
        	'gasPrice': self.web3.eth.gas_price,
        	'nonce': self.web3.eth.get_transaction_count(from_address),
        	'value': self.web3.to_wei(amount, 'ether'),
        	'chainId': self._chain_id,
    	}

    	gas = self.web3.eth.estimate_gas(tx)
    	tx['gas'] = gas
    	return tx

	def send_transaction(self,
                     	private_key: str,
                     	to_address: Address,
                     	amount: Decimal,
                     	contract_address: Address) -> str:

    	account = Account.from_key(private_key)

    	try:
        	if contract_address is None:
            	tx = self._create_tx(account.address, to_address, amount)
        	else:
            	token_helper = ERC20TokenHelper(smart_contract_address=contract_address, web3=self.web3)
            	tx = token_helper.transfer(from_address=account.address, to_address=to_address, amount=amount)

    	except ValueError as ex:
        	if 'gas required exceeds allowance' in ex.args[0]['message']:
            	raise TransactionGasPriceTooLow
        	raise

    	signed_tx = self.web3.eth.account.sign_transaction(tx, private_key)

    	tasks.send_broadcast_callback.apply_async(kwargs={'signed_tx':signed_tx.hash,
                                        	'raw_tx':signed_tx.rawTransaction,
                                        	'wallet_address':account.address,
                                        	'network_type':self._network_type},
                                   	queue='sending_callback_tasks',
                                   	routing_key='apps.api.tasks.sending_callback_tasks')

    	return signed_tx.hash.hex()