如何在SQLAlchemy ORM中创建指定表及其所有外键依赖表？

现有如下SQLAlchemy ORM代码：

from sqlalchemy import (create_engine, Column, String,
                    Integer, Float, ForeignKeyConstraint)
from sqlalchemy.orm import (DeclarativeBase, relationship)


class Base(DeclarativeBase):
    pass


class Process(Base):

    __tablename__ = 'process'

    run_id = Column(Integer, primary_key=True)
    process_id = Column(String, primary_key=True)
    process_attrs = Column(String, nullable=True)


class ProcessWeekly(Base):

    __tablename__ = 'process_weekly'

    run_id = Column(Integer, primary_key=True)
    process_id = Column(String, primary_key=True)
    week = Column(Integer, primary_key=True)
    loss_factor = Column(Float, default=1.0)

    process = relationship(Process)

    __table_args__ = (
        ForeignKeyConstraint([run_id, process_id],
                         [Process.run_id, Process.process_id]),
    )


class Location(Base):

    __tablename__ = "location"

    run_id = Column(Integer, primary_key=True)
    location_id = Column(String, primary_key=True)


engine = create_engine("sqlite:///test.db")

Base.metadata.create_all(engine, tables=[ProcessWeekly.__table__])

执行上述代码仅创建了ProcessWeekly表，但ProcessWeekly依赖Process表的外键约束，需要自动创建指定表及其所有层级的外键依赖表（若Process还有下游依赖表也需一并创建），同时不创建无依赖关系的Location表，使用SQLAlchemy 2.0.25版本，如何修改代码实现需求？

要实现自动创建目标表及其所有依赖表，同时排除无关表，可以通过递归收集目标表的所有外键依赖表，再将这些表和目标表一起传入create_all方法。具体修改如下：

修改后的代码 #

from sqlalchemy import (create_engine, Column, String,
                    Integer, Float, ForeignKeyConstraint)
from sqlalchemy.orm import (DeclarativeBase, relationship)


class Base(DeclarativeBase):
    pass


class Process(Base):

    __tablename__ = 'process'

    run_id = Column(Integer, primary_key=True)
    process_id = Column(String, primary_key=True)
    process_attrs = Column(String, nullable=True)


class ProcessWeekly(Base):

    __tablename__ = 'process_weekly'

    run_id = Column(Integer, primary_key=True)
    process_id = Column(String, primary_key=True)
    week = Column(Integer, primary_key=True)
    loss_factor = Column(Float, default=1.0)

    process = relationship(Process)

    __table_args__ = (
        ForeignKeyConstraint([run_id, process_id],
                         [Process.run_id, Process.process_id]),
    )


class Location(Base):

    __tablename__ = "location"

    run_id = Column(Integer, primary_key=True)
    location_id = Column(String, primary_key=True)


def get_dependent_tables(target_table, collected=None):
    """递归收集目标表的所有外键依赖表"""
    if collected is None:
        collected = set()
    # 避免重复处理同一表
    if target_table in collected:
        return collected
    collected.add(target_table)
    # 遍历当前表的所有外键，获取关联的表
    for fk in target_table.foreign_keys:
        referenced_table = fk.column.table
        get_dependent_tables(referenced_table, collected)
    return collected


engine = create_engine("sqlite:///test.db")

# 获取ProcessWeekly及其所有依赖表
target_tables = get_dependent_tables(ProcessWeekly.__table__)
# 排除Location表（保险机制，避免后续逻辑变动导致误创建）
target_tables.discard(Location.__table__)

# 创建所有目标表
Base.metadata.create_all(engine, tables=list(target_tables))

说明 #

1. 递归收集依赖表：定义get_dependent_tables函数，遍历目标表的所有外键，递归获取关联的依赖表，用集合存储避免重复处理同一表。
2. 排除无关表：通过discard方法确保Location表不会被创建，即使后续依赖逻辑变动也能保证无关表不被包含。
3. 按顺序创建表：SQLAlchemy会自动识别表之间的依赖关系，按正确顺序创建（先创建被依赖的Process，再创建ProcessWeekly）。

这样修改后，执行代码会自动创建Process和ProcessWeekly表，同时不会创建Location表，完全满足需求。

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