如何构建SQLAlchemy邻接列表查询实现Flask网络设备域模型关联

嘿，你已经把模型的关联逻辑理得很清晰了！先假设你的模型定义大概是这样的（如果和你的实际代码有出入，微调下关联字段就行）：

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

class Domain(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50), unique=True, nullable=False)
    # 一对多关联子域
    subdomains = db.relationship('Subdomain', back_populates='domain')
    # 一对多关联设备
    devices = db.relationship('Device', back_populates='domain')

class Subdomain(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50), nullable=False)
    # 子域归属单个域，按你说的"子域到域为一对一"，如果是严格一个子域对应唯一域，可加 unique=True
    domain_id = db.Column(db.Integer, db.ForeignKey('domain.id'), nullable=False)
    domain = db.relationship('Domain', back_populates='subdomains')

class Device(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50), nullable=False)
    ip_address = db.Column(db.String(15), nullable=False)
    domain_id = db.Column(db.Integer, db.ForeignKey('domain.id'), nullable=False)
    domain = db.relationship('Domain', back_populates='devices')

下面针对你的关联关系，给出常用的SQLAlchemy查询示例：

1. 获取指定域下的所有子域 #

比如查询名为mpls-core的域下所有子域：

# 方式1：通过域对象的关联属性直接获取
core_domain = Domain.query.filter_by(name='mpls-core').first()
if core_domain:
    core_subdomains = core_domain.subdomains
    for sub in core_subdomains:
        print(f"子域：{sub.name}")

# 方式2：通过Join查询，适合复杂过滤场景
core_subdomains = Subdomain.query.join(Domain).filter(Domain.name == 'mpls-core').all()

2. 获取指定域下的所有设备 #

同样以mpls-core域为例：

# 方式1：域对象关联查询
core_domain = Domain.query.filter_by(name='mpls-core').first()
if core_domain:
    core_devices = core_domain.devices
    for dev in core_devices:
        print(f"设备：{dev.name} - {dev.ip_address}")

# 方式2：Join查询
core_devices = Device.query.join(Domain).filter(Domain.name == 'mpls-core').all()

3. 通过子域反查所属的父域 #

比如查询名为edge-west的子域对应的父域：

subdomain = Subdomain.query.filter_by(name='edge-west').first()
if subdomain:
    parent_domain = subdomain.domain
    print(f"子域 {subdomain.name} 所属域：{parent_domain.name}")

4. 通过设备反查所属的域 #

比如查询名为pe-router-01的设备所属域：

device = Device.query.filter_by(name='pe-router-01').first()
if device:
    device_domain = device.domain
    print(f"设备 {device.name} 所属域：{device_domain.name}")

5. 复杂关联查询：获取某子域所属域下的所有设备 #

比如找到edge-west子域后，获取其所属域的全部设备：

# 方式1：分步查询，逻辑直观
subdomain = Subdomain.query.filter_by(name='edge-west').first()
if subdomain:
    domain_devices = subdomain.domain.devices
    for dev in domain_devices:
        print(f"{subdomain.domain.name} 域下设备：{dev.name}")

# 方式2：一次Join完成，性能更优
domain_devices = Device.query.join(Domain).join(Subdomain)\
    .filter(Subdomain.name == 'edge-west').all()

6. 带过滤条件的精准查询 #

比如查询mpls-core域下IP以192.168.1.开头的设备：

filtered_devices = Device.query.join(Domain)\
    .filter(Domain.name == 'mpls-core', Device.ip_address.like('192.168.1.%')).all()

7. 预加载关联数据（避免N+1查询问题） #

如果需要一次性加载所有域、它们的子域和设备，避免多次查询数据库，用joinedload实现预加载：

from sqlalchemy.orm import joinedload

domains_with_relations = Domain.query.options(
    joinedload(Domain.subdomains),
    joinedload(Domain.devices)
).all()

for domain in domains_with_relations:
    print(f"\n=== 域：{domain.name} ===")
    print("子域列表：")
    for sub in domain.subdomains:
        print(f"  - {sub.name}")
    print("设备列表：")
    for dev in domain.devices:
        print(f"  - {dev.name} ({dev.ip_address})")

内容的提问来源于stack exchange，提问作者John Jensen