sqlalchemy_mptt

MPTT (nested sets) INSERT

Library for implementing Modified Preorder Tree Traversal with your SQLAlchemy Models and working with trees of Model instances, like django-mptt. The nested set model is a particular technique for representing nested sets (also known as trees or hierarchies) in relational databases.

API:

sqlalchemy_mptt package

Events

Base events

SQLAlchemy events extension

sqlalchemy_mptt.events.mptt_before_insert(mapper, connection, instance)[source]

Based on example https://bitbucket.org/zzzeek/sqlalchemy/src/73095b353124/examples/nested_sets/nested_sets.py?at=master

sqlalchemy_mptt.events.mptt_before_delete(mapper, connection, instance, delete=True)[source]
sqlalchemy_mptt.events.mptt_before_update(mapper, connection, instance)[source]

Based on this example: http://stackoverflow.com/questions/889527/move-node-in-nested-set

class sqlalchemy_mptt.events.TreesManager(base_class)[source]

Manages events dispatching for all subclasses of a given class.

Hidden method
sqlalchemy_mptt.events._insert_subtree(table, connection, node_size, node_pos_left, node_pos_right, parent_pos_left, parent_pos_right, subtree, parent_tree_id, parent_level, node_level, left_sibling, table_pk)[source]

Mixins

SQLAlchemy nested sets mixin

class sqlalchemy_mptt.mixins.BaseNestedSets[source]

Base mixin for MPTT model.

Example:

from sqlalchemy import Boolean, Column, create_engine, Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

from sqlalchemy_mptt.mixins import BaseNestedSets

Base = declarative_base()


class Tree(Base, BaseNestedSets):
    __tablename__ = "tree"

    id = Column(Integer, primary_key=True)
    visible = Column(Boolean)

    def __repr__(self):
        return "<Node (%s)>" % self.id
tree_id()

Represents a column in a database table.

parent_id
parent
left()

Represents a column in a database table.

right()

Represents a column in a database table.

level()

Represents a column in a database table.

drilldown_tree(session=None, json=False, json_fields=None)[source]

This method generate a branch from a tree, begining with current node.

For example:

node7.drilldown_tree()

level           Nested sets example
1                    1(1)22       ---------------------
        _______________|_________|_________            |
       |               |         |         |           |
2    2(2)5           6(4)11      |      12(7)21        |
       |               ^         |         ^           |
3    3(3)4       7(5)8   9(6)10  | 13(8)16   17(10)20  |
                                 |    |          |     |
4                                | 14(9)15   18(11)19  |
                                 |                     |
                                  ---------------------

Example in tests:

  • sqlalchemy_mptt.tests.cases.get_tree.test_drilldown_tree
get_children(session=None)[source]

https://github.com/uralbash/sqlalchemy_mptt/issues/64 https://github.com/django-mptt/django-mptt/blob/fd76a816e05feb5fb0fc23126d33e514460a0ead/mptt/models.py#L563

Returns a query containing the immediate children of this model instance, in tree order.

For example:

node7.get_children() -> [Node(8), Node(10)]

level           Nested sets example

1                   1(1)22
        ______________|____________________
       |              |                    |
       |              |                    |
2    2(2)5          6(4)11              12(7)21
       |              ^                /                       3    3(3)4      7(5)8   9(6)10        /                                                            13(8)16   17(10)20
                                      |         |
4                                  14(9)15   18(11)19
classmethod get_default_level()[source]

Compatibility with Django MPTT: level value for root node. See https://github.com/uralbash/sqlalchemy_mptt/issues/56

get_siblings(include_self=False, session=None)[source]

https://github.com/uralbash/sqlalchemy_mptt/issues/64 https://django-mptt.readthedocs.io/en/latest/models.html#get-siblings-include-self-false

Creates a query containing siblings of this model instance. Root nodes are considered to be siblings of other root nodes.

For example:

node10.get_siblings() -> [Node(8)]

Only one node is sibling of node10

level           Nested sets example

1                   1(1)22
        ______________|____________________
       |              |                    |
       |              |                    |
2    2(2)5          6(4)11              12(7)21
       |              ^                /                       3    3(3)4      7(5)8   9(6)10        /                                                            13(8)16   17(10)20
                                      |         |
4                                  14(9)15   18(11)19
classmethod get_tree(session=None, json=False, json_fields=None, query=None)[source]

This method generate tree of current node table in dict or json format. You can make custom query with attribute query. By default it return all nodes in table.

Args:
session (sqlalchemy.orm.session.Session): SQLAlchemy session
Kwargs:

json (bool): if True return JSON jqTree format json_fields (function): append custom fields in JSON query (function): it takes sqlalchemy.orm.query.Query object as an argument, and returns in a modified form

def query(nodes):
    return nodes.filter(node.__class__.tree_id.is_(node.tree_id))

node.get_tree(session=DBSession, json=True, query=query)

Example:

  • sqlalchemy_mptt.tests.cases.get_tree.test_get_tree
  • sqlalchemy_mptt.tests.cases.get_tree.test_get_json_tree
  • sqlalchemy_mptt.tests.cases.get_tree.test_get_json_tree_with_custom_field
is_ancestor_of(other, inclusive=False)[source]

class or instance level method which returns True if self is ancestor (closer to root) of other else False. Optional flag inclusive on whether or not to treat self as ancestor of self.

For example see:

  • sqlalchemy_mptt.tests.cases.integrity.test_hierarchy_structure
is_descendant_of(other, inclusive=False)[source]

class or instance level method which returns True if self is descendant (farther from root) of other else False. Optional flag inclusive on whether or not to treat self as descendant of self.

For example see:

  • sqlalchemy_mptt.tests.cases.integrity.test_hierarchy_structure
leftsibling_in_level()[source]

Node to the left of the current node at the same level

For example see sqlalchemy_mptt.tests.cases.get_tree.test_leftsibling_in_level

move_after(node_id)[source]

Moving one node of tree after another

For example see sqlalchemy_mptt.tests.cases.move_node.test_move_after_function

move_before(node_id)[source]

Moving one node of tree before another

For example see:

  • sqlalchemy_mptt.tests.cases.move_node.test_move_before_function
  • sqlalchemy_mptt.tests.cases.move_node.test_move_before_to_other_tree
  • sqlalchemy_mptt.tests.cases.move_node.test_move_before_to_top_level
move_inside(parent_id)[source]

Moving one node of tree inside another

For example see:

  • sqlalchemy_mptt.tests.cases.move_node.test_move_inside_function
  • sqlalchemy_mptt.tests.cases.move_node.test_move_inside_to_the_same_parent_function
path_to_root(session=None, order=<function desc>)[source]

Generate path from a leaf or intermediate node to the root.

For example:

node11.path_to_root()

level           Nested sets example

                 -----------------------------------------
1               |    1(1)22                               |
        ________|______|_____________________             |
       |        |      |                     |            |
       |         ------+---------            |            |
2    2(2)5           6(4)11      | --     12(7)21         |
       |               ^             |   /      \         |
3    3(3)4       7(5)8   9(6)10      ---/----    \        |
                                    13(8)16 |  17(10)20   |
                                       |    |     |       |
4                                   14(9)15 | 18(11)19    |
                                            |             |
                                             -------------
classmethod rebuild(session, tree_id=None)[source]

This function rebuid tree.

Args:
session (sqlalchemy.orm.session.Session): SQLAlchemy session
Kwargs:
tree_id (int or str): id of tree, default None

Example:

  • sqlalchemy_mptt.tests.TestTree.test_rebuild
classmethod rebuild_tree(session, tree_id)[source]

This method rebuid tree.

Args:
session (sqlalchemy.orm.session.Session): SQLAlchemy session tree_id (int or str): id of tree

Example:

  • sqlalchemy_mptt.tests.cases.get_tree.test_rebuild

Tests

Base tests

Where used

Manual

Initalize

Create model with MPTT mixin:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
from sqlalchemy import Column, Integer, Boolean
from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy_mptt.mixins import BaseNestedSets

Base = declarative_base()


class Tree(Base, BaseNestedSets):
    __tablename__ = "tree"

    id = Column(Integer, primary_key=True)
    visible = Column(Boolean)  # you custom field

    def __repr__(self):
        return "<Node (%s)>" % self.id

Session factory wrapper

For the automatic tree maintainance triggered after session flush to work correctly, wrap the Session factory with sqlalchemy_mptt.mptt_sessionmaker

1
2
3
4
5
6
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy_mptt import mptt_sessionmaker

engine = create_engine('...')
Session = mptt_sessionmaker(sessionmaker(bind=engine))

Using session factory wrapper with flask_sqlalchemy

If you use Flask and SQLAlchemy, you probably use also flask_sqlalchemy extension for integration. In that case the Session creation is not directly accessible. The following allows you to use the wrapper:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
from sqlalchemy_mptt import mptt_sessionmaker
from flask_sqlalchemy import SQLAlchemy

# instead of creating db object directly
db = SQLAlchemy()

# subclass the db manager and insert the wrapper at session creation
class CustomSQLAlchemy(SQLAlchemy):
    """A custom SQLAlchemy manager, to have control on session creation"""

    def create_session(self, options):
        """Override the original session factory creation"""
        Session = super().create_session(options)
        # Use wrapper from sqlalchemy_mptt that manage tree tables
        return mptt_sessionmaker(Session)

# then
db = CustomSQLAlchemy()

Events

The tree manager automatically registers events. But you can do it manually:

from sqlalchemy_mptt import tree_manager

tree_manager.register_events()  # register events before_insert,
                                # before_update and before_delete

Or disable events if it required:

from sqlalchemy_mptt import tree_manager

tree_manager.register_events(remove=True)  # remove events before_insert,
                                           # before_update and before_delete

Data structure

Fill table with records, for example, as shown in the picture

SQLAlchemy MPTT (nested sets)

Represented data of tree like dict

tree = (
    {'id':  '1',                  'parent_id': None},

    {'id':  '2', 'visible': True, 'parent_id':  '1'},
    {'id':  '3', 'visible': True, 'parent_id':  '2'},

    {'id':  '4', 'visible': True, 'parent_id':  '1'},
    {'id':  '5', 'visible': True, 'parent_id':  '4'},
    {'id':  '6', 'visible': True, 'parent_id':  '4'},

    {'id':  '7', 'visible': True, 'parent_id':  '1'},
    {'id':  '8', 'visible': True, 'parent_id':  '7'},
    {'id':  '9',                  'parent_id':  '8'},
    {'id': '10',                  'parent_id':  '7'},
    {'id': '11',                  'parent_id': '10'},
)

Initializing a tree with data

When you add nodes to the table, the tree manager subsequently updates the level, left and right attributes in the reset of the table. This is done very quickly if the tree already exists in the database, but for initializing the tree, it might become a big overhead. In this case, it is recommended to deactivate automatic tree management, fill in the data, reactivate automatic tree management and finally call manually a rebuild of the tree once at the end:

from sqlalchemy_mptt import tree_manager

...

tree_manager.register_events(remove=True) # Disable MPTT events

# Fill tree
for item in items:
    item.left = 0
    item.right = 0
    item.tree_id = 'my_tree_1'
    db.session.add(item)
db.session.commit()

...

tree_manager.register_events() # enabled MPTT events back
models.MyModelTree.rebuild_tree(db.session, 'my_tree_1') # rebuild lft, rgt value automatically

After an initial table with tree you can use mptt features.

CRUD

INSERT

Insert node with parent_id==6

node = Tree(parent_id=6)
session.add(node)

Tree state before insert

level           Before INSERT
1                    1(1)22
        _______________|___________________
       |               |                   |
2    2(2)5           6(4)11             12(7)21
       |               ^                   ^
3    3(3)4       7(5)8   9(6)10    13(8)16   17(10)20
                                      |          |
4                                  14(9)15   18(11)19

After insert

level           After INSERT
1                    1(1)24
        _______________|_________________
       |               |                 |
2    2(2)5           6(4)13           14(7)23
       |           ____|___          ____|____
       |          |        |        |         |
3    3(3)4      7(5)8    9(6)12  15(8)18   19(10)22
                           |        |         |
4                      10(23)11  16(9)17   20(11)21

UPDATE

Set parent_id=5 for node with id==8

node = session.query(Tree).filter(Tree.id == 8).one()
node.parent_id = 5
session.add(node)

Tree state before update

level           Before UPDATE
1                    1(1)22
        _______________|___________________
       |               |                   |
2    2(2)5           6(4)11             12(7)21
       |               ^                   ^
3    3(3)4       7(5)8   9(6)10    13(8)16   17(10)20
                                      |          |
4                                  14(9)15   18(11)19

After update

level               Move 8 - > 5
    1                     1(1)22
             _______________|__________________
            |               |                  |
    2     2(2)5           6(4)15            16(7)21
            |               ^                  |
    3     3(3)4      7(5)12   13(6)14      17(10)20
                       |                       |
    4                8(8)11                18(11)19
                       |
    5                9(9)10

DELETE

Delete node with id==4

node = session.query(Tree).filter(Tree.id == 4).one()
session.delete(node)

Tree state before delete

level           Before DELETE
1                    1(1)22
        _______________|___________________
       |               |                   |
2    2(2)5           6(4)11             12(7)21
       |               ^                   ^
3    3(3)4       7(5)8   9(6)10    13(8)16   17(10)20
                                      |          |
4                                  14(9)15   18(11)19

After delete

level         Delete node == 4
1                    1(1)16
        _______________|_____
       |                     |
2    2(2)5                 6(7)15
       |                     ^
3    3(3)4            7(8)10   11(10)14
                        |          |
4                     8(9)9    12(11)13

For more example see sqlalchemy_mptt.tests.TestTree

Tutorial

Flask

Initialize Flask app and sqlalchemy

from pprint import pprint
from flask import Flask
from flask_sqlalchemy import SQLAlchemy

from sqlalchemy_mptt.mixins import BaseNestedSets

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:////tmp/test.db'
db = SQLAlchemy(app)

Make models.

class Category(db.Model, BaseNestedSets):
    __tablename__ = 'categories'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(400), index=True, unique=True)
    items = db.relationship("Product", backref='item', lazy='dynamic')

    def __repr__(self):
        return '<Category {}>'.format(self.name)


class Product(db.Model):
    __tablename__ = 'products'
    id = db.Column(db.Integer, primary_key=True)
    category_id = db.Column(db.Integer, db.ForeignKey('categories.id'))
    name = db.Column(db.String(475), index=True)

Represent data of tree in table

Add data to table with tree.

db.session.add(Category(name="root"))  # root node
db.session.add_all(  # first branch of tree
    [
        Category(name="foo", parent_id=1),
        Category(name="bar", parent_id=2),
        Category(name="baz", parent_id=3),
    ]
)
db.session.add_all(  # second branch of tree
    [
        Category(name="foo1", parent_id=1),
        Category(name="bar1", parent_id=5),
        Category(name="baz1", parent_id=5),
    ]
)

db.drop_all()
db.create_all()
db.session.commit()

The database entries are added:

"id"  "name"  "lft"  "rgt"  "level"  "parent_id"  "tree_id"
1     "root"  1      14     1        1
2     "foo"   2      7      2        1            1
3     "bar"   3      6      3        2            1
4     "baz"   4      5      4        3            1
5     "foo1"  8      13     2        1            1
6     "bar1"  9      10     3        5            1
7     "baz1"  11     12     3        5            1

Lft of root element every time \(1\).

\(root_{lft} = 1\)

Rgt of root element always equal 2 * quantity of tree nodes.

\(root_{rgt} = 2 * | P |\)

\(root_{rgt} = 2 * 7 = 14\)

The tree that displays the records in the database is represented schematically below:

level
  1                  1(root)14
                         |
                ---------------------
                |                   |
  2          2(foo)7             8(foo1)13
                |               /         \
  3          3(bar)6        9(bar1)10   11(baz1)12
                |
  4          4(baz)5

Drilldown

Drilldown tree for a given node.

A drilldown tree consists of a node’s ancestors, itself and its immediate children. For example, a drilldown tree for a foo1 category might look something like:

Drilldown for foo1 node

level
  1                  1(root)14
                         |
                ---------------------
                |         ----------|---------------
  2          2(foo)7      |      8(foo1)13         |
                |         |     /         \        |
  3          3(bar)6      | 9(bar1)10   11(baz1)12 |
                |         --------------------------
  4          4(baz)5
categories = Category.query.all()

for item in categories:
    print(item)
    pprint(item.drilldown_tree())
    print()
<Category root>
[{'children': [{'children': [{'children': [{'node': <Category baz>}],
                              'node': <Category bar>}],
                'node': <Category foo>},
               {'children': [{'node': <Category bar1>},
                             {'node': <Category baz1>}],
                'node': <Category foo1>}],
  'node': <Category root>}]

<Category foo>
[{'children': [{'children': [{'node': <Category baz>}],
                'node': <Category bar>}],
  'node': <Category foo>}]

<Category bar>
[{'children': [{'node': <Category baz>}], 'node': <Category bar>}]

<Category baz>
[{'node': <Category baz>}]

<Category foo1>
[{'children': [{'node': <Category bar1>}, {'node': <Category baz1>}],
  'node': <Category foo1>}]

<Category bar1>
[{'node': <Category bar1>}]

<Category baz1>
[{'node': <Category baz1>}]

Represent it to JSON format:

def cat_to_json(item):
    return {
        'id': item.id,
        'name': item.name
    }

for item in categories:
    pprint(item.drilldown_tree(json=True, json_fields=cat_to_json))
    print()
[{'children': [{'children': [{'children': [{'id': 4,
                                            'label': '<Category baz>',
                                            'name': 'baz'}],
                              'id': 3,
                              'label': '<Category bar>',
                              'name': 'bar'}],
                'id': 2,
                'label': '<Category foo>',
                'name': 'foo'},
               {'children': [{'id': 6,
                              'label': '<Category bar1>',
                              'name': 'bar1'},
                             {'id': 7,
                              'label': '<Category baz1>',
                              'name': 'baz1'}],
                'id': 5,
                'label': '<Category foo1>',
                'name': 'foo1'}],
  'id': 1,
  'label': '<Category root>',
  'name': 'root'}]

[{'children': [{'children': [{'id': 4,
                              'label': '<Category baz>',
                              'name': 'baz'}],
                'id': 3,
                'label': '<Category bar>',
                'name': 'bar'}],
  'id': 2,
  'label': '<Category foo>',
  'name': 'foo'}]

[{'children': [{'id': 4, 'label': '<Category baz>', 'name': 'baz'}],
  'id': 3,
  'label': '<Category bar>',
  'name': 'bar'}]

[{'id': 4, 'label': '<Category baz>', 'name': 'baz'}]

[{'children': [{'id': 6, 'label': '<Category bar1>', 'name': 'bar1'},
               {'id': 7, 'label': '<Category baz1>', 'name': 'baz1'}],
  'id': 5,
  'label': '<Category foo1>',
  'name': 'foo1'}]

[{'id': 6, 'label': '<Category bar1>', 'name': 'bar1'}]

[{'id': 7, 'label': '<Category baz1>', 'name': 'baz1'}]

Path to root

Returns a list containing the ancestors and the node itself in tree order.

Path to root of bar node

level      ---------------------
  1        |         1(root)14 |
           |             |     |
           |    ---------------|-----
           |    |    -----------    |
  2        | 2(foo)7 |           8(foo1)13
           |    |    |          /         \
  3        | 3(bar)6 |      9(bar1)10   11(baz1)12
           -----|-----
  4          4(baz)5
for item in categories:
    print(item)
    print(item.path_to_root()[-1])  # get root
                                    # last element in list
    pprint(item.path_to_root().all())
    print()
<Category root>
<Category root>
[<Category root>]

<Category foo>
<Category root>
[<Category foo>, <Category root>]

<Category bar>
<Category root>
[<Category bar>, <Category foo>, <Category root>]

<Category baz>
<Category root>
[<Category baz>, <Category bar>, <Category foo>, <Category root>]

<Category foo1>
<Category root>
[<Category foo1>, <Category root>]

<Category bar1>
<Category root>
[<Category bar1>, <Category foo1>, <Category root>]

<Category baz1>
<Category root>
[<Category baz1>, <Category foo1>, <Category root>]

Full code

from pprint import pprint
from flask import Flask
from flask_sqlalchemy import SQLAlchemy

from sqlalchemy_mptt.mixins import BaseNestedSets

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:////tmp/test.db'
db = SQLAlchemy(app)


class Category(db.Model, BaseNestedSets):
    __tablename__ = 'categories'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(400), index=True, unique=True)
    items = db.relationship("Product", backref='item', lazy='dynamic')

    def __repr__(self):
        return '<Category {}>'.format(self.name)


class Product(db.Model):
    __tablename__ = 'products'
    id = db.Column(db.Integer, primary_key=True)
    category_id = db.Column(db.Integer, db.ForeignKey('categories.id'))
    name = db.Column(db.String(475), index=True)

db.session.add(Category(name="root"))  # root node
db.session.add_all(  # first branch of tree
    [
        Category(name="foo", parent_id=1),
        Category(name="bar", parent_id=2),
        Category(name="baz", parent_id=3),
    ]
)
db.session.add_all(  # second branch of tree
    [
        Category(name="foo1", parent_id=1),
        Category(name="bar1", parent_id=5),
        Category(name="baz1", parent_id=5),
    ]
)

'''
"id"  "name"  "lft"  "rgt"  "level"  "parent_id"  "tree_id"
1     "root"  1      14     1        1
2     "foo"   2      7      2        1            1
3     "bar"   3      6      3        2            1
4     "baz"   4      5      4        3            1
5     "foo1"  8      13     2        1            1
6     "bar1"  9      10     3        5            1
7     "baz1"  11     12     3        5            1

root lft everytime = 1
root rgt = qty_nodes * 2

level
  1                  1(root)14
                         |
                ---------------------
                |                   |
  2          2(foo)7             8(foo1)13
                |               /         \
  3          3(bar)6        9(bar1)10   11(baz1)12
                |
  4          4(baz)5
'''

db.drop_all()
db.create_all()
db.session.commit()

categories = Category.query.all()

for item in categories:
    print(item)
    pprint(item.drilldown_tree())
    print()

'''
<Category root>
[{'children': [{'children': [{'children': [{'node': <Category baz>}],
                              'node': <Category bar>}],
                'node': <Category foo>},
               {'children': [{'node': <Category bar1>},
                             {'node': <Category baz1>}],
                'node': <Category foo1>}],
  'node': <Category root>}]

<Category foo>
[{'children': [{'children': [{'node': <Category baz>}],
                'node': <Category bar>}],
  'node': <Category foo>}]

<Category bar>
[{'children': [{'node': <Category baz>}], 'node': <Category bar>}]

<Category baz>
[{'node': <Category baz>}]

<Category foo1>
[{'children': [{'node': <Category bar1>}, {'node': <Category baz1>}],
  'node': <Category foo1>}]

<Category bar1>
[{'node': <Category bar1>}]

<Category baz1>
[{'node': <Category baz1>}]
'''

for item in categories:
    print(item)
    print(item.path_to_root()[-1])
    pprint(item.path_to_root().all())
    print()

'''
<Category root>
<Category root>
[<Category root>]

<Category foo>
<Category root>
[<Category foo>, <Category root>]

<Category bar>
<Category root>
[<Category bar>, <Category foo>, <Category root>]

<Category baz>
<Category root>
[<Category baz>, <Category bar>, <Category foo>, <Category root>]

<Category foo1>
<Category root>
[<Category foo1>, <Category root>]

<Category bar1>
<Category root>
[<Category bar1>, <Category foo1>, <Category root>]

<Category baz1>
<Category root>
[<Category baz1>, <Category foo1>, <Category root>]
'''

A lot of examples and logic in sqlalchemy_mptt.tests.tree_testing_base

Support and Development

To report bugs, use the issue tracker.

We welcome any contribution: suggestions, ideas, commits with new futures, bug fixes, refactoring, docs, tests, translations, etc…

If you have question, contact me sacrud@uralbash.ru or #sacrud IRC channel IRC Freenode

Indices and tables