45 lines
1.5 KiB
Python
45 lines
1.5 KiB
Python
from utils.db.db import DbConnector
|
|
from models.model_mtr_network_nodes import MtrAdditionalNodesModel
|
|
|
|
from sqlalchemy import select
|
|
|
|
|
|
class MtrAdditionalNodeQuery:
|
|
def __init__(self):
|
|
db_con = DbConnector("spirent_mysql")
|
|
self.db = db_con.get_db()
|
|
|
|
async def get_all_nodes(self) -> list:
|
|
"""
|
|
Retrieve all entries from the mtr_network_nodes database table.
|
|
"""
|
|
async with self.db() as session:
|
|
result = await session.execute(select(MtrAdditionalNodesModel))
|
|
all_nodes = result.scalars().all()
|
|
#await session.close()
|
|
return all_nodes
|
|
|
|
def get_all_nodes_for_testing(self) -> list:
|
|
"""
|
|
Retrieve all entries from the mtr_network_nodes database table, exclude rows used for storing parameters for
|
|
testing TAS servers.
|
|
"""
|
|
|
|
all_nodes = (self.db.query(MtrAdditionalNodesModel)
|
|
.filter(MtrAdditionalNodesModel.test_servers == False)
|
|
.filter(MtrAdditionalNodesModel.enabled == True)
|
|
.all())
|
|
|
|
return all_nodes
|
|
|
|
def get_mtr_config_for_servers(self) -> list:
|
|
all_nodes = (self.db.query(MtrAdditionalNodesModel)
|
|
.filter(MtrAdditionalNodesModel.test_servers == True)
|
|
.filter(MtrAdditionalNodesModel.enabled == True)
|
|
.all())
|
|
|
|
return all_nodes
|
|
|
|
def get_node_by_id(self, node_id: int) -> MtrAdditionalNodesModel:
|
|
return self.db.query(MtrAdditionalNodesModel).filter_by(id=node_id).one()
|