SQLAlchemy中如何通过主表属性获取其主键作为外键关联的子表

反向获取关联子表的通用方法 #

根据你的使用场景，有两种实现方案：

方案1：基于已配置的relationship属性获取（性能更高，推荐） #

只要你定义ORM映射时配置了relationship属性，不管单向还是双向关联，都可以直接通过属性拿到关联子表信息，通用实现如下：

from sqlalchemy.orm import RelationshipProperty

# 遍历获取User类所有关联的子表
for attr_name, attr in User.__mapper__.attrs.items():
    if isinstance(attr, RelationshipProperty):
        # 关联的子映射类
        child_cls = attr.entity.class_
        # 子表的Table对象
        child_table = child_cls.__table__
        # 子表名
        child_table_name = child_cls.__tablename__
        print(f"关联属性{attr_name}对应的子表：{child_table_name}")

# 已知关联属性名为address时可直接获取
address_table = User.address.property.entity.class_.__table__

该方法适配所有带relationship配置的关联关系，无需遍历全库元数据

方案2：基于外键约束遍历获取（无relationship配置时使用） #

如果没有配置ORM的relationship，可直接遍历元数据中所有表的外键约束，找到所有引用了父表的子表：

parent_table = User.__table__
child_tables = []
# 遍历所有已注册的表
for table in Base.metadata.tables.values():
    for fk_constraint in table.foreign_key_constraints:
        if fk_constraint.referred_table == parent_table:
            child_tables.append(table)
print(child_tables) # 输出包含address表

批量插入关联子表的完整实现 #

你现在只执行了User表的插入语句，SQLAlchemy Core层面不会自动处理ORM级联逻辑，所以需要手动处理关联子表的插入，完整流程如下：

import pandas as pd
from sqlalchemy.dialects.sqlite import insert

# 1. 处理User表的批量插入更新，返回生成的主键ID
user_table = User.__table__
# 提取User对应的字段数据
user_df = df[["name"]].copy()
user_records = user_df.to_dict(orient="records")

user_insert_stmt = insert(user_table).values(user_records).returning(user_table.c.id)
user_pks = [pk.name for pk in user_table.primary_key]
user_update_cols = {col.name: col for col in user_insert_stmt.excluded if col.name not in user_pks}
user_upsert_stmt = user_insert_stmt.on_conflict_do_update(
    index_elements=user_pks,
    set_=user_update_cols
)

with engine.connect() as conn:
    result = conn.execute(user_upsert_stmt)
    # 按插入顺序拿到所有用户主键ID
    user_ids = [row[0] for row in result]
    conn.commit()

# 2. 处理Address表的批量插入更新
address_table = User.address.property.entity.class_.__table__
# 组装Address数据，关联对应用户的主键
address_records = []
for idx, company_name in enumerate(df["company_name"]):
    address_records.append({
        "user_id": user_ids[idx],
        "company_name": company_name
    })

address_insert_stmt = insert(address_table).values(address_records)
address_pks = [pk.name for pk in address_table.primary_key]
address_update_cols = {col.name: col for col in address_insert_stmt.excluded if col.name not in address_pks}
address_upsert_stmt = address_insert_stmt.on_conflict_do_update(
    index_elements=address_pks,
    set_=address_update_cols
)

with engine.connect() as conn:
    conn.execute(address_upsert_stmt)
    conn.commit()

如果是一个User对应多个Address的场景，只需提前整理好同一个用户对应的多条Address数据，再关联user_id即可

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