nathanwoodburn/fireexplorer-indexer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: Some select request formats

Browse Source
This commit is contained in:
Nathan Woodburn 2025-02-09 22:07:22 +11:00
parent 0012ecc77f
commit 2f2441822e
Signed by: nathanwoodburn
GPG Key ID: 203B000478AD0EF1
1 changed files with 21 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

56
main.py
View File

@ -54,7 +54,7 @@ dbSave = clickhouse_connect.create_client(
database=DB_NAME database=DB_NAME
) )
dbGet = Client( dbGet = clickhouse_connect.get_client(
host=DB_HOST, host=DB_HOST,
user=DB_USER, user=DB_USER,
password=DB_PASSWORD, password=DB_PASSWORD,
@ -149,54 +149,40 @@ def saveBlock(blockData):
def getNewestBlock() -> int: def getNewestBlock() -> int:
"""Returns the height of the newest block in the database""" """Returns the height of the newest block in the database"""
dbNB = Client( newestBlock = dbGet.query("SELECT height FROM blocks ORDER BY height DESC LIMIT 1").result
host=DB_HOST, return int(newestBlock[0][0]) if newestBlock else -1
user=DB_USER,
password=DB_PASSWORD,
database=DB_NAME
)
newestBlock = dbNB.execute("SELECT height FROM blocks ORDER BY height DESC LIMIT 1")
if newestBlock:
return int(newestBlock[0])
return -1
def dbCheck(): def dbCheck():
# For the first 100 blocks, check for transactions # For the first 100 blocks, check for transactions
for i in range(100): for i in range(100):
block = dbGet.execute("SELECT * FROM blocks WHERE height = %s", (i,)) block = dbGet.query(f"SELECT * FROM blocks WHERE height = {i}").result
if not block: if not block:
return return
block = Block(block) print(Block(block[0]))
print(block)
def getBlock(height) -> Block | None: def getBlock(height) -> Block | None:
block = dbGet.execute("SELECT * FROM blocks WHERE height = %s", (height,)) """Fetch a block by height"""
block = dbGet.query(f"SELECT * FROM blocks WHERE height = {height}").result
if not block: return Block(block[0]) if block else None
return None
return Block(block)
def getTransaction(tx_hash) -> Transaction | None:
"""Fetch a transaction by hash"""
tx = dbGet.query(f"SELECT * FROM transactions WHERE hash = '{tx_hash}'").result
return Transaction(tx[0]) if tx else None
def getTransaction(hash) -> Transaction | None:
tx = dbGet.execute("SELECT * FROM transactions WHERE hash = %s", (hash,))
if not tx:
return None
return Transaction(tx)
def getTransactions(height) -> list[Transaction] | None: def getTransactions(height) -> list[Transaction] | None:
txs = dbGet.execute("SELECT * FROM transactions WHERE block = %s", (height,)) """Fetch all transactions for a given block height"""
if not txs: txs = dbGet.query(f"SELECT * FROM transactions WHERE block = {height}").result
return None return [Transaction(tx) for tx in txs] if txs else None
# Convert to list of Transaction objects
return [Transaction(tx) for tx in txs]
def getNameFromHash(nameHash): def getNameFromHash(nameHash):
# Connect to db """Fetch a name record by nameHash"""
name = dbGet.execute("SELECT * FROM names WHERE nameHash = %s", (nameHash,)) name = dbGet.query(f"SELECT * FROM names WHERE nameHash = '{nameHash}'").result
if not name: return Name(name[0]) if name else -1
return -1
return Name(name)
def getNamesFromBlock(height): def getNamesFromBlock(height):
transactions = getTransactions(height) transactions = getTransactions(height)