Add tx data normalizer to sql backend

chainqueue/db/models/otx.py

@@ -44,7 +44,7 @@ class Otx(SessionBase):
     """Datetime when record was created"""
     date_updated = Column(DateTime, default=datetime.datetime.utcnow)
     """Datetime when record was last updated"""
-    tx_hash = Column(String(66))
+    tx_hash = Column(String())
     """Tranasction hash"""
     signed_tx = Column(Text)
     """Signed raw transaction data"""

chainqueue/db/models/tx.py

@@ -54,13 +54,13 @@ class TxCache(SessionBase):
 
     otx_id = Column(Integer, ForeignKey('otx.id'))
     """Foreign key to chainqueue.db.models.otx.Otx"""
-    source_token_address = Column(String(42))
+    source_token_address = Column(String())
     """Contract address of token that sender spent from"""
-    destination_token_address = Column(String(42))
+    destination_token_address = Column(String())
     """Contract address of token that recipient will receive balance of"""
-    sender = Column(String(42))
+    sender = Column(String())
     """Ethereum address of transaction sender"""
-    recipient = Column(String(42))
+    recipient = Column(String())
     """Ethereum address of transaction beneficiary (e.g. token transfer recipient)"""
     from_value = Column(NUMERIC())
     """Amount of source tokens spent"""

chainqueue/sql/backend.py

@@ -1,6 +1,7 @@
 # standard imports
 import logging
 import urllib.error
+import copy
 
 # external imports
 from sqlalchemy.exc import (
@@ -11,11 +12,6 @@ from chainlib.error import (
         RPCNonceException,
         DefaultErrorParser,
         )
-from hexathon import (
-        add_0x,
-        strip_0x,
-        uniform as hex_uniform,
-        )
 
 # local imports
 from chainqueue.sql.tx import create as queue_create
@@ -33,6 +29,7 @@ from chainqueue.sql.state import (
         set_rejected,
         )
 from chainqueue.sql.tx import cache_tx_dict
+from chainqueue.encode import TxHexNormalize
 
 logg = logging.getLogger(__name__)
 
@@ -52,12 +49,15 @@ class SQLBackend:
     """
 
     #def __init__(self, conn_spec, error_parser=None, *args, **kwargs):
-    def __init__(self, conn_spec, error_parser=None, pool_size=0, debug=False, *args, **kwargs):
+    def __init__(self, conn_spec, tx_normalizer=None, error_parser=None, pool_size=0, debug=False, *args, **kwargs):
         #SessionBase.connect(conn_spec, pool_size=kwargs.get('poolsize', 0), debug=kwargs.get('debug', False))
         SessionBase.connect(conn_spec, pool_size=pool_size, debug=debug)
         if error_parser == None:
             error_parser = DefaultErrorParser()
         self.error_parser = error_parser
+        if tx_normalizer == None:
+            tx_normalizer = TxHexNormalize()
+        self.tx_normalizer = tx_normalizer
 
 
     def create(self, chain_spec, nonce, holder_address, tx_hash, signed_tx, obsolete_predecessors=True, session=None):
@@ -82,6 +82,8 @@ class SQLBackend:
         :rtype: int
         :returns: 0 if successfully added
         """
+        tx_hash = self.tx_normalizer.tx_hash(tx_hash)
+        signed_tx = self.tx_normalizer.tx_wire(signed_tx)
         try:
             queue_create(chain_spec, nonce, holder_address, tx_hash, signed_tx, obsolete_predecessors=True, session=session)
         except IntegrityError as e:
@@ -101,8 +103,15 @@ class SQLBackend:
         :rtype: int
         :returns: 0 if successful
         """
-        (tx, txc_id) = cache_tx_dict(tx, session=session)
-        logg.debug('cached {} db insert id {}'.format(tx, txc_id))
+        ntx = copy.copy(tx)
+        ntx['hash'] = self.tx_normalizer.tx_hash(ntx['hash'])
+        ntx['from'] = self.tx_normalizer.wallet_address(ntx['from'])
+        ntx['to'] = self.tx_normalizer.wallet_address(ntx['to'])
+        ntx['source_token'] = self.tx_normalizer.executable_address(ntx['source_token'])
+        ntx['destination_token'] = self.tx_normalizer.executable_address(ntx['destination_token'])
+
+        (tx_dict, txc_id) = cache_tx_dict(ntx, session=session)
+        logg.debug('cached {} db insert id {}'.format(tx_dict, txc_id))
         return 0
 
 
@@ -120,6 +129,7 @@ class SQLBackend:
         :rtype: dict
         :returns: otx record summary
         """
+        tx_hash = self.tx_normalizer.tx_hash(tx_hash)
         return backend_get_tx(chain_spec, tx_hash, session=session)
 
 
@@ -170,6 +180,7 @@ class SQLBackend:
         :rtype: int
         :returns: 0 if no error
         """
+        tx_hash = self.tx_normalizer.tx_hash(tx_hash)
         set_reserved(chain_spec, tx_hash, session=session)
         fail = False
         r = 1

chainqueue/sql/tx.py

@@ -1,6 +1,5 @@
 # standard imports
 import logging
-import copy
 
 # external imports
 from hexathon import (
@@ -95,24 +94,14 @@ def cache_tx_dict(tx_dict, session=None):
     """
     session = SessionBase.bind_session(session)
 
-    ntx = copy.copy(tx_dict)
-    for k in [
-        'hash',
-        'from',
-        'to',
-        'source_token',
-        'destination_token',
-        ]:
-        ntx[k] = add_0x(hex_uniform(strip_0x(ntx[k])))
-
     txc = TxCache(
-        ntx['hash'],
-        ntx['from'],
-        ntx['to'],
-        ntx['source_token'],
-        ntx['destination_token'],
-        ntx['from_value'],
-        ntx['to_value'],
+        tx_dict['hash'],
+        tx_dict['from'],
+        tx_dict['to'],
+        tx_dict['source_token'],
+        tx_dict['destination_token'],
+        tx_dict['from_value'],
+        tx_dict['to_value'],
         session=session
         )
     
@@ -123,4 +112,4 @@ def cache_tx_dict(tx_dict, session=None):
 
     SessionBase.release_session(session)
 
-    return (ntx, insert_id)
+    return (tx_dict, insert_id)