diff options
| -rw-r--r-- | src-py/db_controller.py | 18 | ||||
| -rw-r--r-- | src-py/test_db_controller.py | 9 |
2 files changed, 23 insertions, 4 deletions
diff --git a/src-py/db_controller.py b/src-py/db_controller.py index 96de855d..1e80ff84 100644 --- a/src-py/db_controller.py +++ b/src-py/db_controller.py @@ -198,17 +198,27 @@ class DBO_load_link_set_by_src_or_dst_id_attributes(DB_op): return self.parse_single_query_response_data(data) class DBO_load_link_id_set(DB_op): - def __init__(self, filter_type=None, filter_attr_map=None): + def __init__(self, filter_type=None, filter_attr_map={}): """ - load a set of link ids + load an id-set of links @param filter_type: link type filter @param filter_attr_map: is a filter_key to filter_value_set map of attributes to match link properties against @return: a set of loaded link ids """ - filter_str = dbu.where_clause_from_filter_attr_map() - + super(DBO_load_link_id_set, self).__init__() + + q = "match ()-[r{filter_type} {filter_attr}]->() return id(r)" + q = cfmt(q, filter_type="" if not filter_type else ":" + filter_type) + q = cfmt(q, filter_attr=db_util.gen_clause_attr_filter_from_filter_attr_map(filter_attr_map)) + q_params = {k: v[0] for (k, v) in filter_attr_map.items()} # pass on only first value from each value set + + self.add_statement(q, q_params) + + def on_completion(self, data): + log.debug('loaded id-set: ' + str(data)) + return self.parse_single_query_response_data(data) class DB_Driver_Base(): pass diff --git a/src-py/test_db_controller.py b/src-py/test_db_controller.py index 446d9288..176fce4b 100644 --- a/src-py/test_db_controller.py +++ b/src-py/test_db_controller.py @@ -94,6 +94,15 @@ class TestDBController(unittest.TestCase): n_set = self.db_ctl.exec_op(dbc.DBO_load_node_set_by_id_attribute(['skill_00', 'person_01'])) self.assertEqual(len(n_set), 2) + def test_load_link_set_by_type(self): + op = dbc.DBO_load_link_id_set(filter_type='Knows') + id_set = self.db_ctl.exec_op(op) + self.assertEqual(len(id_set), 2) + + op = dbc.DBO_load_link_id_set(filter_type='Nan_Type') + id_set = self.db_ctl.exec_op(op) + self.assertEqual(len(id_set), 0) + def test_load_link_set_by_src_or_dst_id_attributes(self): op = dbc.DBO_load_link_set_by_src_or_dst_id_attributes(src_id='person_00', dst_id='skill_00') n_set = self.db_ctl.exec_op(op) |