SQLAlchemy 2.0：批量插入时如何转换关联数据为逗号分隔字符串？

针对批量插入上万条JSON数据时，需要将每条数据中嵌套的affiliations数组转换为逗号分隔字符串存入User模型的affiliation_list字段的需求，以下是三种可行的实现方案：

方法1：批量插入前预处理JSON数据 #

这是最直观且高效的方案，直接在Python层面完成数据转换，再传入ORM进行批量插入，性能可控。

示例代码 #

假设你的输入JSON结构如下：

[
  {"name": "Alice", "affiliations": ["MIT", "Stanford"]},
  {"name": "Bob", "affiliations": ["Harvard"]}
]

User模型定义：

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    affiliation_list: Mapped[str]  # 存储逗号分隔的字符串

预处理并批量插入的代码：

import json
from sqlalchemy import create_engine
from sqlalchemy.orm import Session

# 加载原始JSON数据
with open("users_data.json", "r") as f:
    raw_data = json.load(f)

# 预处理：将affiliations数组转为逗号分隔字符串
processed_data = []
for item in raw_data:
    processed_item = {
        "name": item["name"],
        "affiliation_list": ",".join(item.get("affiliations", []))
    }
    processed_data.append(processed_item)

# 执行批量插入
engine = create_engine("sqlite:///your_database.db")
Base.metadata.create_all(engine)

with Session(engine) as session:
    users = [User(**data) for data in processed_data]
    session.add_all(users)
    session.commit()

方法2：自定义字段类型（TypeDecorator） #

如果希望模型自动处理数组与字符串的双向转换，无需每次手动预处理，可以使用SQLAlchemy的TypeDecorator自定义字段类型。

示例代码 #

定义自定义字段类型：

from sqlalchemy import String
from sqlalchemy.types import TypeDecorator

class AffiliationList(TypeDecorator):
    impl = String
    cache_ok = True

    def process_bind_param(self, value, dialect):
        # 写入数据库时：数组转逗号分隔字符串
        return ",".join(value) if isinstance(value, list) else value

    def process_result_value(self, value, dialect):
        # 读取数据库时：字符串转回数组（可选，按需启用）
        return value.split(",") if value else []

更新User模型：

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    # 使用自定义字段类型，支持传入数组或字符串
    affiliation_list: Mapped[list[str] | str] = mapped_column(AffiliationList)

批量插入代码（无需预处理）：

with Session(engine) as session:
    users = [
        User(name=item["name"], affiliation_list=item["affiliations"])
        for item in raw_data
    ]
    session.add_all(users)
    session.commit()

方法3：使用ORM的before_insert事件监听 #

通过SQLAlchemy的事件监听机制，在插入对象前自动完成数据转换，适合需要在模型层面统一处理的场景。

示例代码 #

注册事件监听：

from sqlalchemy import event

@event.listens_for(User, "before_insert")
def convert_affiliations(mapper, connection, target):
    # 从传入的临时属性中获取affiliations数组，转换后赋值给affiliation_list
    if hasattr(target, "affiliations") and isinstance(target.affiliations, list):
        target.affiliation_list = ",".join(target.affiliations)
        delattr(target, "affiliations")  # 删除临时属性，避免冲突

批量插入代码：

with Session(engine) as session:
    users = [
        User(name=item["name"], affiliations=item["affiliations"])
        for item in raw_data
    ]
    session.add_all(users)
    session.commit()

性能建议 #

对于上万条数据的批量插入，方法1的预处理方案性能最优，因为它在Python层面一次性完成所有转换，避免了事件监听或自定义类型的额外开销。如果需要长期复用转换逻辑，方法2的自定义字段类型更具封装性。

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