74 lines
2.2 KiB
Python
74 lines
2.2 KiB
Python
|
import mysql.connector
|
||
|
|
import os
|
||
|
|
from dotenv import load_dotenv
|
||
|
|
load_dotenv()
|
||
|
|
import indexerClasses
|
||
|
|
|
||
|
|
DB_HOST = os.getenv("DB_HOST")
|
||
|
|
DB_USER = os.getenv("DB_USER")
|
||
|
|
DB_PASS = os.getenv("DB_PASSWORD")
|
||
|
|
DB_NAME = os.getenv("DB_NAME")
|
||
|
|
|
||
|
|
class DBConnection:
|
||
|
|
def __init__(self):
|
||
|
|
self.conn = connect()
|
||
|
|
|
||
|
|
def __enter__(self):
|
||
|
|
return self.conn
|
||
|
|
|
||
|
|
def __exit__(self, exc_type, exc_value, traceback):
|
||
|
|
self.conn.close()
|
||
|
|
|
||
|
|
def getBlock(self, blockheight) -> indexerClasses.Block | None:
|
||
|
|
with self.conn.cursor() as cursor:
|
||
|
|
try:
|
||
|
|
blockheight = int(blockheight)
|
||
|
|
cursor.execute("SELECT * FROM blocks WHERE height = %s", (str(blockheight),))
|
||
|
|
result = cursor.fetchone()
|
||
|
|
if result:
|
||
|
|
return indexerClasses.Block(result)
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
print(f"Error: {e}")
|
||
|
|
|
||
|
|
return None
|
||
|
|
|
||
|
|
def getBlockHash(self, blockhash) -> indexerClasses.Block | None:
|
||
|
|
with self.conn.cursor() as cursor:
|
||
|
|
cursor.execute("SELECT * FROM blocks WHERE hash = %s", (blockhash,))
|
||
|
|
result = cursor.fetchone()
|
||
|
|
if result:
|
||
|
|
return indexerClasses.Block(result)
|
||
|
|
else:
|
||
|
|
return None
|
||
|
|
|
||
|
|
def getTransaction(self, txid) -> indexerClasses.Transaction | None:
|
||
|
|
with self.conn.cursor() as cursor:
|
||
|
|
cursor.execute("SELECT * FROM transactions WHERE hash = %s", (txid,))
|
||
|
|
result = cursor.fetchone()
|
||
|
|
if result:
|
||
|
|
return indexerClasses.Transaction(result)
|
||
|
|
else:
|
||
|
|
return None
|
||
|
|
|
||
|
|
def getNameByHash(self, hash) -> str | None:
|
||
|
|
with self.conn.cursor() as cursor:
|
||
|
|
cursor.execute("SELECT name FROM names WHERE namehash = %s", (hash,))
|
||
|
|
result = cursor.fetchone()
|
||
|
|
if result:
|
||
|
|
return result[0]
|
||
|
|
else:
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
def connect():
|
||
|
|
return mysql.connector.connect(
|
||
|
|
host=DB_HOST,
|
||
|
|
user=DB_USER,
|
||
|
|
password=DB_PASS,
|
||
|
|
database=DB_NAME,
|
||
|
|
charset='utf8mb4',
|
||
|
|
collation='utf8mb4_unicode_ci',
|
||
|
|
)
|