# Copyright (c) 2012 Web Notes Technologies Pvt Ltd (http://erpnext.com) # # MIT License (MIT) # # Permission is hereby granted, free of charge, to any person obtaining a # copy of this software and associated documentation files (the "Software"), # to deal in the Software without restriction, including without limitation # the rights to use, copy, modify, merge, publish, distribute, sublicense, # and/or sell copies of the Software, and to permit persons to whom the # Software is furnished to do so, subject to the following conditions: # # The above copyright notice and this permission notice shall be included in # all copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, # INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A # PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT # HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF # CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE # OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. # # Tree (Hierarchical) Nested Set Model (nsm) # # To use the nested set model, # use the following pattern # 1. name your parent field as "parent_node" if not have a property nsm_parent_field as your field name in the document class # 2. have a field called "old_parent" in your fields list - this identifies whether the parent has been changed # 3. call update_nsm(doc_obj) in the on_upate method # ------------------------------------------ from __future__ import unicode_literals # uncomment below lines for testing # import sys, unittest # sys.path.append("lib/py") # sys.path.append(".") import webnotes, unittest from webnotes import msgprint class TestNSM(unittest.TestCase): def setUp(self): from webnotes.model.doc import Document self.root = Document(fielddata={'doctype':'nsmtest', 'name':'T001', 'parent':None}) self.first_child = Document(fielddata={'doctype':'nsmtest', 'name':'C001', 'parent_node':'T001'}) self.first_sibling = Document(fielddata={'doctype':'nsmtest', 'name':'C002', 'parent_node':'T001'}) self.grand_child = Document(fielddata={'doctype':'nsmtest', 'name':'GC001', 'parent_node':'C001'}) webnotes.conn.sql(""" create table `tabnsmtest` ( name varchar(120) not null primary key, creation datetime, modified datetime, modified_by varchar(40), owner varchar(40), docstatus int(1) default '0', parent varchar(120), parentfield varchar(120), parenttype varchar(120), idx int(8), parent_node varchar(180), old_parent varchar(180), lft int, rgt int) ENGINE=InnoDB""") def test_root(self): self.root.save(1) update_nsm(self.root) self.assertTrue(self.root.lft==1) self.assertTrue(self.root.rgt==2) def test_first_child(self): self.root.save(1) update_nsm(self.root) self.first_child.save(1) update_nsm(self.first_child) self.root._loadfromdb() self.assertTrue(self.root.lft==1) self.assertTrue(self.first_child.lft==2) self.assertTrue(self.first_child.rgt==3) self.assertTrue(self.root.rgt==4) def test_sibling(self): self.test_first_child() self.first_sibling.save(1) update_nsm(self.first_sibling) self.root._loadfromdb() self.first_child._loadfromdb() self.assertTrue(self.root.lft==1) self.assertTrue(self.first_child.lft==2) self.assertTrue(self.first_child.rgt==3) self.assertTrue(self.first_sibling.lft==4) self.assertTrue(self.first_sibling.rgt==5) self.assertTrue(self.root.rgt==6) def test_remove_sibling(self): self.test_sibling() self.first_sibling.parent_node = '' update_nsm(self.first_sibling) self.root._loadfromdb() self.first_child._loadfromdb() self.assertTrue(self.root.lft==1) self.assertTrue(self.first_child.lft==2) self.assertTrue(self.first_child.rgt==3) self.assertTrue(self.root.rgt==4) self.assertTrue(self.first_sibling.lft==5) self.assertTrue(self.first_sibling.rgt==6) def test_change_parent(self): self.test_sibling() # add grand child self.grand_child.save(1) update_nsm(self.grand_child) # check lft rgt self.assertTrue(self.grand_child.lft==3) self.assertTrue(self.grand_child.rgt==4) # change parent self.grand_child.parent_node = 'C002' self.grand_child.save() # update update_nsm(self.grand_child) # check lft rgt self.assertTrue(self.grand_child.lft==5) self.assertTrue(self.grand_child.rgt==6) def tearDown(self): webnotes.conn.sql("drop table tabnsmtest") # called in the on_update method def update_nsm(doc_obj): # get fields, data from the DocType pf, opf = 'parent_node', 'old_parent' if str(doc_obj.__class__)=='webnotes.model.doc.Document': # passed as a Document object d = doc_obj else: # passed as a DocType object d = doc_obj.doc if hasattr(doc_obj,'nsm_parent_field'): pf = doc_obj.nsm_parent_field if hasattr(doc_obj,'nsm_oldparent_field'): opf = doc_obj.nsm_oldparent_field p, op = d.fields.get(pf, ''), d.fields.get(opf, '') # has parent changed (?) or parent is None (root) if not d.lft and not d.rgt: update_add_node(d.doctype, d.name, p or '', pf) elif op != p: update_remove_node(d.doctype, d.name) update_add_node(d.doctype, d.name, p or '', pf) # set old parent webnotes.conn.set(d, opf, p or '') # reload d._loadfromdb() def rebuild_tree(doctype, parent_field): """ call rebuild_node for all root nodes """ # get all roots right = 1 result = webnotes.conn.sql("SELECT name FROM `tab%s` WHERE `%s`='' or `%s` IS NULL ORDER BY name ASC" % (doctype, parent_field, parent_field)) for r in result: right = rebuild_node(doctype, r[0], right, parent_field) webnotes.conn.sql("commit") webnotes.conn.sql("start transaction") def rebuild_node(doctype, parent, left, parent_field, cnt = 0): """ reset lft, rgt and recursive call for all children """ from webnotes.utils import now n = now() # the right value of this node is the left value + 1 right = left+1 # get all children of this node result = webnotes.conn.sql("SELECT name FROM `tab%s` WHERE `%s`='%s'" % (doctype, parent_field, parent)) for r in result: right = rebuild_node(doctype, r[0], right, parent_field, cnt) # we've got the left value, and now that we've processed # the children of this node we also know the right value webnotes.conn.sql("UPDATE `tab%s` SET lft=%s, rgt=%s, modified='%s' WHERE name='%s'" % (doctype,left,right,n,parent)) # commit after every 100 cnt += 1 if cnt % 100 == 0: cnt = 0 webnotes.conn.sql("commit") webnotes.conn.sql("start transaction") #return the right value of this node + 1 return right+1 def update_add_node(doctype, name, parent, parent_field): """ insert a new node """ from webnotes.utils import now n = now() # get the last sibling of the parent if parent: right = webnotes.conn.sql("select rgt from `tab%s` where name='%s'" % (doctype, parent))[0][0] else: # root right = webnotes.conn.sql("select ifnull(max(rgt),0)+1 from `tab%s` where ifnull(`%s`,'') =''" % (doctype, parent_field))[0][0] right = right or 1 # update all on the right webnotes.conn.sql("update `tab%s` set rgt = rgt+2, modified='%s' where rgt >= %s" %(doctype,n,right)) webnotes.conn.sql("update `tab%s` set lft = lft+2, modified='%s' where lft >= %s" %(doctype,n,right)) # update index of new node if webnotes.conn.sql("select * from `tab%s` where lft=%s or rgt=%s"% (doctype, right, right+1)): webnotes.msgprint("Nested set error. Please send mail to support") raise Exception webnotes.conn.sql("update `tab%s` set lft=%s, rgt=%s, modified='%s' where name='%s'" % (doctype,right,right+1,n,name)) return right def update_remove_node(doctype, name): """ remove a node """ from webnotes.utils import now n = now() left = webnotes.conn.sql("select lft from `tab%s` where name='%s'" % (doctype,name)) if left[0][0]: # reset this node webnotes.conn.sql("update `tab%s` set lft=0, rgt=0, modified='%s' where name='%s'" % (doctype,n,name)) # update all on the right webnotes.conn.sql("update `tab%s` set rgt = rgt-2, modified='%s' where rgt > %s" %(doctype,n,left[0][0])) webnotes.conn.sql("update `tab%s` set lft = lft-2, modified='%s' where lft > %s" %(doctype,n,left[0][0])) class DocTypeNestedSet: def on_update(self): update_nsm(self) def on_trash(self): update_remove_node(self.doc.doctype, self.doc.name) def validate_root_details(self, root, parent_field): #does not exists parent if self.doc.name==root and self.doc.fields.get(parent_field): msgprint("You can not assign parent for root: %s" % (root, ), raise_exception=1) elif self.doc.name!=root and not self.doc.parent_account: msgprint("Parent is mandatory for %s" % (self.doc.name, ), raise_exception=1) if __name__=="__main__": webnotes.connect() unittest.main()