Source code for examples.adjacency_list.adjacency_list
from__future__importannotationsfromtypingimportDictfromtypingimportOptionalfromsqlalchemyimportcreate_enginefromsqlalchemyimportForeignKeyfromsqlalchemyimportselectfromsqlalchemy.ormimportDeclarativeBasefromsqlalchemy.ormimportMappedfromsqlalchemy.ormimportmapped_columnfromsqlalchemy.ormimportMappedAsDataclassfromsqlalchemy.ormimportrelationshipfromsqlalchemy.ormimportselectinloadfromsqlalchemy.ormimportSessionfromsqlalchemy.orm.collectionsimportattribute_keyed_dictclassBase(DeclarativeBase):passclassTreeNode(MappedAsDataclass,Base):__tablename__="tree"id:Mapped[int]=mapped_column(primary_key=True,init=False)parent_id:Mapped[Optional[int]]=mapped_column(ForeignKey("tree.id"),init=False)name:Mapped[str]children:Mapped[Dict[str,TreeNode]]=relationship(cascade="all, delete-orphan",back_populates="parent",collection_class=attribute_keyed_dict("name"),init=False,repr=False,)parent:Mapped[Optional[TreeNode]]=relationship(back_populates="children",remote_side=id,default=None)defdump(self,_indent:int=0)->str:return(" "*_indent+repr(self)+"\n"+"".join([c.dump(_indent+1)forcinself.children.values()]))if__name__=="__main__":engine=create_engine("sqlite://",echo=True)print("Creating Tree Table:")Base.metadata.create_all(engine)withSession(engine)assession:node=TreeNode("rootnode")TreeNode("node1",parent=node)TreeNode("node3",parent=node)node2=TreeNode("node2")TreeNode("subnode1",parent=node2)node.children["node2"]=node2TreeNode("subnode2",parent=node.children["node2"])print(f"Created new tree structure:\n{node.dump()}")print("flush + commit:")session.add(node)session.commit()print(f"Tree after save:\n{node.dump()}")session.add_all([TreeNode("node4",parent=node),TreeNode("subnode3",parent=node.children["node4"]),TreeNode("subnode4",parent=node.children["node4"]),TreeNode("subsubnode1",parent=node.children["node4"].children["subnode3"],),])# remove node1 from the parent, which will trigger a delete# via the delete-orphan cascade.delnode.children["node1"]print("Removed node1. flush + commit:")session.commit()print("Tree after save, will unexpire all nodes:\n")print(f"{node.dump()}")withSession(engine)assession:print("Perform a full select of the root node, eagerly loading ""up to a depth of four")node=session.scalars(select(TreeNode).options(selectinload(TreeNode.children,recursion_depth=4)).filter(TreeNode.name=="rootnode")).one()print(f"Full Tree:\n{node.dump()}")print("Marking root node as deleted, flush + commit:")session.delete(node)session.commit()