39 lines
1.2 KiB
Python
39 lines
1.2 KiB
Python
import asyncio
|
|
import os
|
|
import logging
|
|
|
|
from utils.db.db import DbConnector
|
|
from utils.db.db_mtr_network_nodes import MtrAdditionalNodeQuery
|
|
|
|
# Change log level for each part of code
|
|
logging.getLogger('').setLevel(logging.DEBUG)
|
|
#logging.basicConfig(level=logging.DEBUG)
|
|
|
|
DB_TYPE=os.environ.setdefault('DB_TYPE', 'mysql')
|
|
DB_NAME=os.getenv("DB_NAME")
|
|
DB_HOST=os.getenv("DB_HOST")
|
|
DB_PASS=os.getenv("DB_PASS")
|
|
DB_PORT=os.getenv("DB_PORT")
|
|
DB_USER=os.getenv("DB_USER")
|
|
|
|
async def main():
|
|
logging.debug("Connecting to LANDSLIDE MYSQL DATABASE")
|
|
db = DbConnector("spirent_mysql")
|
|
db.init(DB_TYPE, DB_HOST, int(DB_PORT), DB_NAME, DB_USER, DB_PASS)
|
|
#
|
|
mtr_additional_query = MtrAdditionalNodeQuery()
|
|
all_nodes = await mtr_additional_query.get_all_nodes()
|
|
print("==== get_all_nodes ====")
|
|
[print(i.as_dict()) for i in all_nodes]
|
|
|
|
get_all_nodes_for_testing = await mtr_additional_query.get_all_nodes_for_testing()
|
|
print("==== get_all_nodes_for_testing ====")
|
|
[print(i.as_dict()) for i in get_all_nodes_for_testing]
|
|
|
|
get_node_by_id = await mtr_additional_query.get_node_by_id(21)
|
|
print("==== get_node_by_id ====")
|
|
print(get_node_by_id.as_dict())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
loop = asyncio.run(main()) |