kamikazechaser/cic-internal-integration
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Update queue/syncer module structure

Browse Source
This commit is contained in:
Louis Holbrook
2021-06-03 13:51:55 +00:00
parent 69ae9b7c07
commit 096ed9bc27
20 changed files with 50 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
apps/cic-eth/cic_eth/queue/query.py
View File

@@ -5,7 +5,7 @@ import datetime
import celery
from chainlib.chain import ChainSpec
from chainlib.eth.tx import unpack
import chainqueue.query
import chainqueue.sql.query
from chainqueue.db.enum import (
StatusEnum,
is_alive,
@@ -28,7 +28,7 @@ celery_app = celery.current_app
def get_tx_cache(chain_spec_dict, tx_hash):
chain_spec = ChainSpec.from_dict(chain_spec_dict)
session = SessionBase.create_session()
r = chainqueue.query.get_tx_cache(chain_spec, tx_hash, session=session)
r = chainqueue.sql.query.get_tx_cache(chain_spec, tx_hash, session=session)
session.close()
return r
@@ -37,7 +37,7 @@ def get_tx_cache(chain_spec_dict, tx_hash):
def get_tx(chain_spec_dict, tx_hash):
chain_spec = ChainSpec.from_dict(chain_spec_dict)
session = SessionBase.create_session()
r = chainqueue.query.get_tx(chain_spec, tx_hash, session=session)
r = chainqueue.sql.query.get_tx(chain_spec, tx_hash, session=session)
session.close()
return r
@@ -46,7 +46,7 @@ def get_tx(chain_spec_dict, tx_hash):
def get_account_tx(chain_spec_dict, address, as_sender=True, as_recipient=True, counterpart=None):
chain_spec = ChainSpec.from_dict(chain_spec_dict)
session = SessionBase.create_session()
r = chainqueue.query.get_account_tx(chain_spec, address, as_sender=True, as_recipient=True, counterpart=None, session=session)
r = chainqueue.sql.query.get_account_tx(chain_spec, address, as_sender=True, as_recipient=True, counterpart=None, session=session)
session.close()
return r
@@ -55,17 +55,17 @@ def get_account_tx(chain_spec_dict, address, as_sender=True, as_recipient=True,
def get_upcoming_tx_nolock(chain_spec_dict, status=StatusEnum.READYSEND, not_status=None, recipient=None, before=None, limit=0, session=None):
chain_spec = ChainSpec.from_dict(chain_spec_dict)
session = SessionBase.create_session()
r = chainqueue.query.get_upcoming_tx(chain_spec, status, not_status=not_status, recipient=recipient, before=before, limit=limit, session=session, decoder=unpack)
r = chainqueue.sql.query.get_upcoming_tx(chain_spec, status, not_status=not_status, recipient=recipient, before=before, limit=limit, session=session, decoder=unpack)
session.close()
return r
def get_status_tx(chain_spec, status, not_status=None, before=None, exact=False, limit=0, session=None):
return chainqueue.query.get_status_tx_cache(chain_spec, status, not_status=not_status, before=before, exact=exact, limit=limit, session=session, decoder=unpack)
return chainqueue.sql.query.get_status_tx_cache(chain_spec, status, not_status=not_status, before=before, exact=exact, limit=limit, session=session, decoder=unpack)
def get_paused_tx(chain_spec, status=None, sender=None, session=None, decoder=None):
return chainqueue.query.get_paused_tx_cache(chain_spec, status=status, sender=sender, session=session, decoder=unpack)
return chainqueue.sql.query.get_paused_tx_cache(chain_spec, status=status, sender=sender, session=session, decoder=unpack)
def get_nonce_tx(chain_spec, nonce, sender):

24
apps/cic-eth/cic_eth/queue/state.py
View File

@@ -1,6 +1,6 @@
# external imports
from chainlib.chain import ChainSpec
import chainqueue.state
import chainqueue.sql.state
# local imports
import celery
@@ -14,7 +14,7 @@ celery_app = celery.current_app
def set_sent(chain_spec_dict, tx_hash, fail=False):
chain_spec = ChainSpec.from_dict(chain_spec_dict)
session = SessionBase.create_session()
r = chainqueue.state.set_sent(chain_spec, tx_hash, fail, session=session)
r = chainqueue.sql.state.set_sent(chain_spec, tx_hash, fail, session=session)
session.close()
return r
@@ -23,7 +23,7 @@ def set_sent(chain_spec_dict, tx_hash, fail=False):
def set_final(chain_spec_dict, tx_hash, block=None, tx_index=None, fail=False):
chain_spec = ChainSpec.from_dict(chain_spec_dict)
session = SessionBase.create_session()
r = chainqueue.state.set_final(chain_spec, tx_hash, block=block, tx_index=tx_index, fail=fail, session=session)
r = chainqueue.sql.state.set_final(chain_spec, tx_hash, block=block, tx_index=tx_index, fail=fail, session=session)
session.close()
return r
@@ -32,7 +32,7 @@ def set_final(chain_spec_dict, tx_hash, block=None, tx_index=None, fail=False):
def set_cancel(chain_spec_dict, tx_hash, manual=False):
chain_spec = ChainSpec.from_dict(chain_spec_dict)
session = SessionBase.create_session()
r = chainqueue.state.set_cancel(chain_spec, tx_hash, manual, session=session)
r = chainqueue.sql.state.set_cancel(chain_spec, tx_hash, manual, session=session)
session.close()
return r
@@ -41,7 +41,7 @@ def set_cancel(chain_spec_dict, tx_hash, manual=False):
def set_rejected(chain_spec_dict, tx_hash):
chain_spec = ChainSpec.from_dict(chain_spec_dict)
session = SessionBase.create_session()
r = chainqueue.state.set_rejected(chain_spec, tx_hash, session=session)
r = chainqueue.sql.state.set_rejected(chain_spec, tx_hash, session=session)
session.close()
return r
@@ -50,7 +50,7 @@ def set_rejected(chain_spec_dict, tx_hash):
def set_fubar(chain_spec_dict, tx_hash):
chain_spec = ChainSpec.from_dict(chain_spec_dict)
session = SessionBase.create_session()
r = chainqueue.state.set_fubar(chain_spec, tx_hash, session=session)
r = chainqueue.sql.state.set_fubar(chain_spec, tx_hash, session=session)
session.close()
return r
@@ -59,7 +59,7 @@ def set_fubar(chain_spec_dict, tx_hash):
def set_manual(chain_spec_dict, tx_hash):
chain_spec = ChainSpec.from_dict(chain_spec_dict)
session = SessionBase.create_session()
r = chainqueue.state.set_manual(chain_spec, tx_hash, session=session)
r = chainqueue.sql.state.set_manual(chain_spec, tx_hash, session=session)
session.close()
return r
@@ -68,7 +68,7 @@ def set_manual(chain_spec_dict, tx_hash):
def set_ready(chain_spec_dict, tx_hash):
chain_spec = ChainSpec.from_dict(chain_spec_dict)
session = SessionBase.create_session()
r = chainqueue.state.set_ready(chain_spec, tx_hash, session=session)
r = chainqueue.sql.state.set_ready(chain_spec, tx_hash, session=session)
session.close()
return r
@@ -77,7 +77,7 @@ def set_ready(chain_spec_dict, tx_hash):
def set_reserved(chain_spec_dict, tx_hash):
chain_spec = ChainSpec.from_dict(chain_spec_dict)
session = SessionBase.create_session()
r = chainqueue.state.set_reserved(chain_spec, tx_hash, session=session)
r = chainqueue.sql.state.set_reserved(chain_spec, tx_hash, session=session)
session.close()
return r
@@ -86,7 +86,7 @@ def set_reserved(chain_spec_dict, tx_hash):
def set_waitforgas(chain_spec_dict, tx_hash):
chain_spec = ChainSpec.from_dict(chain_spec_dict)
session = SessionBase.create_session()
r = chainqueue.state.set_waitforgas(chain_spec, tx_hash, session=session)
r = chainqueue.sql.state.set_waitforgas(chain_spec, tx_hash, session=session)
session.close()
return r
@@ -95,7 +95,7 @@ def set_waitforgas(chain_spec_dict, tx_hash):
def get_state_log(chain_spec_dict, tx_hash):
chain_spec = ChainSpec.from_dict(chain_spec_dict)
session = SessionBase.create_session()
r = chainqueue.state.get_state_log(chain_spec, tx_hash, session=session)
r = chainqueue.sql.state.get_state_log(chain_spec, tx_hash, session=session)
session.close()
return r
@@ -104,6 +104,6 @@ def get_state_log(chain_spec_dict, tx_hash):
def obsolete(chain_spec_dict, tx_hash, final):
chain_spec = ChainSpec.from_dict(chain_spec_dict)
session = SessionBase.create_session()
r = chainqueue.state.obsolete_by_cache(chain_spec, tx_hash, final, session=session)
r = chainqueue.sql.state.obsolete_by_cache(chain_spec, tx_hash, final, session=session)
session.close()
return r

4
apps/cic-eth/cic_eth/queue/tx.py
View File

@@ -15,14 +15,14 @@ from sqlalchemy import tuple_
from sqlalchemy import func
from chainlib.chain import ChainSpec
from chainlib.eth.tx import unpack
import chainqueue.state
import chainqueue.sql.state
from chainqueue.db.enum import (
StatusEnum,
StatusBits,
is_alive,
dead,
)
from chainqueue.tx import create
from chainqueue.sql.tx import create
from chainqueue.error import NotLocalTxError
from chainqueue.db.enum import status_str