cic-internal-integration/apps/cic-ussd/cic_ussd/account/transaction.py

178 lines
6.9 KiB
Python

# 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')
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')
recipient_transaction_data = {
"token_symbol": destination_token_symbol,
"token_value": destination_token_value,
"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,
"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)