# standard import import logging from math import trunc from typing import Dict, Tuple # external import from cic_eth.api import Api from sqlalchemy.orm.session import Session # local import from cic_ussd.db.models.account import Account from cic_ussd.db.models.base import SessionBase from cic_ussd.error import UnknownUssdRecipient from cic_ussd.translation import translation_for logg = logging.getLogger(__name__) def _add_tags(action_tag_key: str, preferred_language: str, direction_tag_key: str, transaction: dict): """ This function adds action and direction tags to a transaction data object. :param action_tag_key: Key mapping to a helper entry in the translation files describing an action. :type action_tag_key: str :param preferred_language: An account's set preferred language. :type preferred_language: str :param direction_tag_key: Key mapping to a helper entry in the translation files describing a transaction's direction relative to the transaction's subject account. :type direction_tag_key: str :param transaction: Parsed transaction data object. :type transaction: dict """ action_tag = translation_for(action_tag_key, preferred_language) direction_tag = translation_for(direction_tag_key, preferred_language) transaction['action_tag'] = action_tag transaction['direction_tag'] = direction_tag def aux_transaction_data(preferred_language: str, transaction: dict) -> dict: """This function adds auxiliary data to a transaction object offering contextual information relative to the subject account's role in the transaction. :param preferred_language: An account's set preferred language. :type preferred_language: str :param transaction: Parsed transaction data object. :type transaction: dict :return: Transaction object with contextual data. :rtype: dict """ role = transaction.get('role') if role == 'recipient': _add_tags('helpers.received', preferred_language, 'helpers.from', transaction) if role == 'sender': _add_tags('helpers.sent', preferred_language, 'helpers.to', transaction) return transaction def from_wei(decimals: int, value: int) -> float: """This function converts values in Wei to a token in the cic network. :param decimals: The decimals required for wei values. :type decimals: int :param value: Value in Wei :type value: int :return: SRF equivalent of value in Wei :rtype: float """ value = float(value) / (10**decimals) return truncate(value=value, decimals=2) def to_wei(decimals: int, value: int) -> int: """This functions converts values from a token in the cic network to Wei. :param decimals: The decimals required for wei values. :type decimals: int :param value: Value in SRF :type value: int :return: Wei equivalent of value in SRF :rtype: int """ return int(value * (10**decimals)) def truncate(value: float, decimals: int) -> float: """This function truncates a value to a specified number of decimals places. :param value: The value to be truncated. :type value: float :param decimals: The number of decimals for the value to be truncated to :type decimals: int :return: The truncated value. :rtype: int """ stepper = 10.0**decimals return trunc(stepper*value) / stepper def transaction_actors(transaction: dict) -> Tuple[Dict, Dict]: """ This function parses transaction data into a tuple of transaction data objects representative of of the source and destination account's involved in a transaction. :param transaction: Transaction data object. :type transaction: dict :return: Recipient and sender transaction data object :rtype: Tuple[Dict, Dict] """ destination_token_symbol = transaction.get('destination_token_symbol') destination_token_value = transaction.get('destination_token_value') or transaction.get('to_value') destination_token_decimals = transaction.get('destination_token_decimals') recipient_blockchain_address = transaction.get('recipient') sender_blockchain_address = transaction.get('sender') source_token_symbol = transaction.get('source_token_symbol') source_token_value = transaction.get('source_token_value') or transaction.get('from_value') source_token_decimals = transaction.get('source_token_decimals') recipient_transaction_data = { "token_symbol": destination_token_symbol, "token_value": destination_token_value, "token_decimals": destination_token_decimals, "blockchain_address": recipient_blockchain_address, "role": "recipient", } sender_transaction_data = { "blockchain_address": sender_blockchain_address, "token_symbol": source_token_symbol, "token_value": source_token_value, "token_decimals": source_token_decimals, "role": "sender", } return recipient_transaction_data, sender_transaction_data def validate_transaction_account(blockchain_address: str, role: str, session: Session) -> Account: """This function checks whether the blockchain address specified in a parsed transaction object resolves to an account object in the ussd system. :param blockchain_address: :type blockchain_address: :param role: :type role: :param session: :type session: :return: :rtype: """ session = SessionBase.bind_session(session) account = session.query(Account).filter_by(blockchain_address=blockchain_address).first() if not account: if role == 'recipient': raise UnknownUssdRecipient( f'Tx for recipient: {blockchain_address} has no matching account in the system.' ) if role == 'sender': logg.warning(f'Tx from sender: {blockchain_address} has no matching account in system.') SessionBase.release_session(session) return account class OutgoingTransaction: def __init__(self, chain_str: str, from_address: str, to_address: str): """ :param chain_str: The chain name and network id. :type chain_str: str :param from_address: Ethereum address of the sender :type from_address: str, 0x-hex :param to_address: Ethereum address of the recipient :type to_address: str, 0x-hex """ self.chain_str = chain_str self.cic_eth_api = Api(chain_str=chain_str) self.from_address = from_address self.to_address = to_address def transfer(self, amount: int, decimals: int, token_symbol: str): """This function initiates standard transfers between one account to another :param amount: The amount of tokens to be sent :type amount: int :param decimals: The decimals for the token being transferred. :type decimals: int :param token_symbol: ERC20 token symbol of token to send :type token_symbol: str """ self.cic_eth_api.transfer(from_address=self.from_address, to_address=self.to_address, value=to_wei(decimals=decimals, value=amount), token_symbol=token_symbol)