49 lines
2.3 KiB
Python
49 lines
2.3 KiB
Python
import os
|
|
from sqlalchemy import create_engine, func
|
|
from sqlalchemy.orm import sessionmaker
|
|
from models import SystemData, ActiveTable, ConsumedTable, TransferLogs, TransactionHistory, RejectedTransactionHistory, Base, ContractStructure, ContractBase, ContractParticipants, SystemBase, ActiveContracts, ContractAddressMapping, LatestCacheBase, ContractTransactionHistory, RejectedContractTransactionHistory, TokenContractAssociation, ContinuosContractBase, ContractStructure1, ContractParticipants1, ContractDeposits1, ContractTransactionHistory1, DatabaseTypeMapping
|
|
import pdb
|
|
|
|
|
|
def check_database_existence(type, parameters):
|
|
if type == 'token':
|
|
return os.path.isfile(f"./tokens/{parameters['token_name']}.db")
|
|
elif type == 'smart_contract':
|
|
return os.path.isfile(f"./smartContracts/{parameters['contract_name']}-{parameters['contract_address']}.db")
|
|
|
|
|
|
def create_database_connection(type, parameters):
|
|
if type == 'token':
|
|
engine = create_engine(f"sqlite:///tokens/{parameters['token_name']}.db", echo=True)
|
|
elif type == 'smart_contract':
|
|
engine = create_engine(f"sqlite:///smartContracts/{parameters['contract_name']}-{parameters['contract_address']}.db", echo=True)
|
|
elif type == 'system_dbs':
|
|
engine = create_engine(f"sqlite:///{parameters['db_name']}.db", echo=False)
|
|
|
|
connection = engine.connect()
|
|
return connection
|
|
|
|
|
|
def create_database_session_orm(type, parameters, base):
|
|
if type == 'token':
|
|
engine = create_engine(f"sqlite:///tokens/{parameters['token_name']}.db", echo=True)
|
|
base.metadata.create_all(bind=engine)
|
|
session = sessionmaker(bind=engine)()
|
|
|
|
elif type == 'smart_contract':
|
|
engine = create_engine(f"sqlite:///smartContracts/{parameters['contract_name']}-{parameters['contract_address']}.db", echo=True)
|
|
base.metadata.create_all(bind=engine)
|
|
session = sessionmaker(bind=engine)()
|
|
|
|
elif type == 'system_dbs':
|
|
engine = create_engine(f"sqlite:///{parameters['db_name']}.db", echo=False)
|
|
base.metadata.create_all(bind=engine)
|
|
session = sessionmaker(bind=engine)()
|
|
|
|
return session
|
|
|
|
|
|
session = create_database_session_orm('token', {'token_name': f"vivek"}, Base)
|
|
session.add(ActiveTable(address='sdf', parentid=0))
|
|
session.commit()
|
|
session.close() |