diff options
| -rw-r--r-- | src-py/db_controller.py | 40 | ||||
| -rw-r--r-- | src-py/model/model.py | 25 | ||||
| -rw-r--r-- | src-py/test_db_controller.py | 29 |
3 files changed, 67 insertions, 27 deletions
diff --git a/src-py/db_controller.py b/src-py/db_controller.py index 7cd1af7b..ceb5267c 100644 --- a/src-py/db_controller.py +++ b/src-py/db_controller.py @@ -264,28 +264,38 @@ class DBO_match_node_set_by_id_attribute(DBO_match_node_id_set): super(DBO_match_node_set_by_id_attribute, self).__init__(filter_attr_map={'id': id_set}) -class DBO_match_link_set_by_src_or_dst_id_attributes(DB_op): - def __init__(self, src_id=None, dst_id=None): +class DBO_load_link_set(DB_op): + def __init__(self, link_ptr_set): """ - match a set of links by source/target node id attributes + match a set of sets of links by source/target node id attributes + This class should be instantiated through a static factory function + + @link_ptr_set link pointer set @return: a set of loaded links """ - assert None != src_id or None != dst_id + super(DBO_load_link_set, self).__init__() + + for l_ptr in link_ptr_set: + if not l_ptr.src_id: + q = "match ()-[r]->({id: {dst_id}}) return r" + q_params = {'dst_id': l_ptr.dst_id} + elif not l_ptr.dst_id: + q = "match ({id: {src_id}})-[r]->() return r" + q_params = {'src_id': l_ptr.src_id} + else: + q = "match ({id: {src_id}})-[r]->({id: {dst_id}}) return r" + q_params = {'src_id': l_ptr.src_id, 'dst_id': l_ptr.dst_id} - super(DBO_match_link_set_by_src_or_dst_id_attributes, self).__init__() + self.add_statement(q, q_params) - if not src_id: - q = "match ()-[r]->({id: {dst_id}}) return r" - q_params = {'dst_id': dst_id} - elif not dst_id: - q = "match ({id: {src_id}})-[r]->() return r" - q_params = {'src_id': src_id} - else: - q = "match ({id: {src_id}})-[r]->({id: {dst_id}}) return r" - q_params = {'src_id': src_id, 'dst_id': dst_id} + @staticmethod + def init_from_link_ptr(l_ptr): + return DBO_load_link_set([l_ptr]) - self.add_statement(q, q_params) + @staticmethod + def init_from_link_ptr_set(l_ptr_set): + return DBO_load_link_set(l_ptr_set) class DBO_match_link_id_set(DB_op): def __init__(self, filter_type=None, filter_attr_map={}): diff --git a/src-py/model/model.py b/src-py/model/model.py index 9ee61397..701074ab 100644 --- a/src-py/model/model.py +++ b/src-py/model/model.py @@ -1,4 +1,4 @@ -class link(): +class Link(): """ documentation anchor - this class currently carries no implementation and only acts as a documentation anchor @@ -6,4 +6,25 @@ class link(): link['__src'] - meta attribute for link source link['__dst'] - meta attribute for link destination """ - pass + + class Link_Ptr(dict): + def __init__(self, src_id=None, dst_id=None): + assert None != src_id or None != dst_id + + self['__src'] = src_id + self['__dst'] = dst_id + + @property + def src_id(self): + return self['__src'] + + @property + def dst_id(self): + return self['__dst'] + + @staticmethod + def link_ptr(src_id=None, dst_id=None): + """ + init from src_id or dst_id attributes - at least one must be provided + """ + return Link.Link_Ptr(src_id, dst_id) diff --git a/src-py/test_db_controller.py b/src-py/test_db_controller.py index 4af4cabc..cdc0c5e9 100644 --- a/src-py/test_db_controller.py +++ b/src-py/test_db_controller.py @@ -122,19 +122,28 @@ class TestDBController(unittest.TestCase): id_set = self.db_ctl.exec_op(op) self.assertEqual(len(id_set), 0) - def test_match_link_set_by_src_or_dst_id_attributes(self): - op = dbc.DBO_match_link_set_by_src_or_dst_id_attributes(src_id='person_00', dst_id='skill_00') - n_set = self.db_ctl.exec_op(op) - self.assertEqual(len(n_set), 1) + def test_load_link_set(self): + l_ptr = Link.link_ptr(src_id='person_00', dst_id='skill_00') + op = dbc.DBO_load_link_set.init_from_link_ptr(l_ptr) + l_set = self.db_ctl.exec_op(op) + self.assertEqual(len(l_set), 1) - op = dbc.DBO_match_link_set_by_src_or_dst_id_attributes(src_id='person_00') - n_set = self.db_ctl.exec_op(op) - self.assertEqual(len(n_set), 2) + l_ptr = Link.link_ptr(src_id='person_00') + op = dbc.DBO_load_link_set.init_from_link_ptr(l_ptr) + l_set = self.db_ctl.exec_op(op) + self.assertEqual(len(l_set), 2) - op = dbc.DBO_match_link_set_by_src_or_dst_id_attributes(dst_id='skill_00') - n_set = self.db_ctl.exec_op(op) - self.assertEqual(len(n_set), 1) + l_ptr = Link.link_ptr(dst_id='skill_00') + op = dbc.DBO_load_link_set.init_from_link_ptr(l_ptr) + l_set = self.db_ctl.exec_op(op) + self.assertEqual(len(l_set), 1) + # load sets + l_ptr_set = [Link.link_ptr(s,d) for (s,d) in [('person_00', 'skill_00'), ('person_00', 'skill_01')]] + op = dbc.DBO_load_link_set.init_from_link_ptr_set(l_ptr_set) + l_set = self.db_ctl.exec_op(op) + self.assertEqual(len(l_set), 2) + def test_load_node_set_by_DB_id(self): """ test node DB id life cycle |