sandbox/async_sqlalchemy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

- add example of select filters

Browse Source
This commit is contained in:
marys
2025-05-27 12:40:07 +02:00
parent 61cbc536b6
commit 9f995f090b
2 changed files with 22 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
main.py
View File

@@ -23,7 +23,17 @@ async def main():
#
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_all_nodes_for_testing ====")
print(get_node_by_id.as_dict())
if __name__ == "__main__":
loop = asyncio.run(main())

33
utils/db/db_mtr_network_nodes.py
View File

@@ -14,31 +14,22 @@ class MtrAdditionalNodeQuery:
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
return (await session.execute(select(MtrAdditionalNodesModel))).scalars().all()
def get_all_nodes_for_testing(self) -> list:
async 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())
async with (self.db() as session):
stmt = select(MtrAdditionalNodesModel) \
.where(MtrAdditionalNodesModel.test_servers == False) \
.where(MtrAdditionalNodesModel.enabled == True)
return (await session.execute(stmt)).scalars().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()
async def get_node_by_id(self, node_id: int) -> MtrAdditionalNodesModel:
async with self.db() as session:
stmt = select(MtrAdditionalNodesModel).where(MtrAdditionalNodesModel.id == node_id)
return (await session.execute(stmt)).scalars().first()