在sqlalchemy orm中找到根引用

Question

有一個模型有自引用關係，我想查找引用的根節點/記錄，例如在下面的例子中，Package可能依賴於另一個包。

# myapp.py 

from flask import Flask 
from flask_sqlalchemy import SQLAlchemy 
from sqlalchemy.orm import relationship 

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


class Package(db.Model): 
    __tablename__ = "packages" 
    id = db.Column(db.Integer, primary_key=True) 
    name = db.Column(db.String(80), unique=True) 
    dep_on_id = db.Column(db.Integer, db.ForeignKey('packages.id')) 
    dep_on = relationship('Package', remote_side=[id]) 

    def __init__(self, name): 
     self.name = name 

    def __repr__(self): 
     return '<Package (%r)>' % self.name 

如果包d依賴於C，C依賴於B，B依賴於A，我想找出根源取決於d，這是一個的包，所以期望導致follwing測試代碼應是<Package (u'a')>，有沒有一種簡單的方法可以用sqlalchemy替代find_root_dep函數？

# test.py 

from myapp import db, Package 

db.drop_all() 
db.create_all() 

a = Package('a') 
b = Package('b') 
c = Package('c') 
d = Package('d') 
b.dep_on = a 
c.dep_on = b 
d.dep_on = c 

for p in [a, b, c, d]: 
    db.session.add(p) 
db.session.commit() 

def find_root_dep(package): 
    dep_on = package.dep_on 
    while dep_on: 
     dep = dep_on.dep_on 
     if dep: 
      dep_on = dep 
     else: 
      break 
    return dep_on 

print find_root_dep(d) 

Answer 1

走樹木和圖表可以在SQL使用SQLAlchemy的一個recursive CTE，或Query.cte()來完成。

def find_root_dep(package): 
    # The initial step. Find the 1st dependency of the Package passed 
    # as the argument. 
    cte = db.session.query(Package).\ 
     filter_by(id=package.dep_on_id).\ 
     cte(recursive=True) 

    # The iterative step. Find Packages that found packages 
    # depend on. Iteration stops when the query results in 
    # an empty set, since no Package has NULL id. 
    cte = cte.union_all(
     db.session.query(Package). 
      filter_by(id=cte.c.dep_on_id)) 

    # Create an alias for returning an entity object.     
    result_alias = db.aliased(Package, cte) 

    # The root depends on nothing.   
    return db.session.query(result_alias).\ 
     filter_by(dep_on_id=None).\ 
     one_or_none() 

您最初的實現將返回None，如果傳遞了一個根包，所以SQL的執行直接通過查找第一個依賴，這將導致對根包空集開始。