SQLAlchemy能否像COPY命令一样智能完成数据库数据导入？

1. 场景还原：创建示例库表 #

create database sample;
create table data(id serial,f1 boolean);

2. COPY命令自动转换布尔值 #

准备data.txt文件，内容仅为字符0：

vim data.txt
0

执行wq保存退出后，执行COPY导入命令：

\copy data(f1) from 'data.txt' 

COPY命令会自动将字符串0转换为布尔值false，查询结果如下：

sample=# select * from data;
 id | f1 
----+----
  1 | f
(1 row)

3. SQLAlchemy直接导入报错及单字段处理 #

使用SQLAlchemy的add方法导入时，代码如下：

# 初始化数据库连接
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine, Column, Integer, Boolean

db_pass = 'xxxxx'
db_ip = '127.0.0.1'
engine = create_engine("postgresql://postgres:{}@{}/sample".format(db_pass,db_ip),future=True)

Base = declarative_base()
class Tdata(Base):
    __tablename__ = "data" 
    id = Column(Integer, primary_key=True, autoincrement=True)
    f1 = Column(Boolean, nullable=True)
    def __init__(self, f1):
        self.f1 = f1 

Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
sess = Session()

# 读取数据
fh = open('/tmp/data.txt','r')
raw = fh.readline().strip()
row = Tdata(raw)

# 尝试写入
sess.add(row)
sess.commit()

执行后会触发类型错误：

raise TypeError("Not a boolean value: %r" % (value,))
sqlalchemy.exc.StatementError: (builtins.TypeError) Not a boolean value: '0'

此时需手动转换类型：

sess.rollback()
raw = bool(int(raw))
row = Tdata(raw)
sess.add(row)
sess.commit()    

多布尔字段的高效处理方案 #

当表中存在多个布尔类型字段时，可通过以下方式避免逐个手动转换：

方式一：模型初始化方法统一处理 #

修改模型类的__init__方法，自动识别布尔字段并转换字符串值：

class Tdata(Base):
    __tablename__ = "data" 
    id = Column(Integer, primary_key=True, autoincrement=True)
    f1 = Column(Boolean, nullable=True)
    f2 = Column(Boolean, nullable=True)
    f3 = Column(Boolean, nullable=True)

    def __init__(self, **kwargs):
        # 遍历字段，自动转换布尔类型的字符串值
        for key, value in kwargs.items():
            column = self.__table__.columns.get(key)
            if column and isinstance(column.type, Boolean) and isinstance(value, str):
                kwargs[key] = bool(int(value))
        super().__init__(**kwargs)

使用时直接传入原始字符串即可：

row = Tdata(f1='0', f2='1', f3='0')
sess.add(row)
sess.commit()

方式二：自定义布尔类型转换器 #

创建自定义布尔类型，绑定自动转换逻辑：

from sqlalchemy.types import TypeDecorator, Boolean

class StringToBoolean(TypeDecorator):
    impl = Boolean

    def process_bind_param(self, value, dialect):
        if isinstance(value, str):
            return bool(int(value))
        return value

在模型中使用该自定义类型：

class Tdata(Base):
    __tablename__ = "data" 
    id = Column(Integer, primary_key=True, autoincrement=True)
    f1 = Column(StringToBoolean, nullable=True)
    f2 = Column(StringToBoolean, nullable=True)
    f3 = Column(StringToBoolean, nullable=True)

后续传入字符串0/1时会自动转换为对应布尔值。

方式三：批量转换原始数据 #

读取原始数据后，统一处理指定布尔字段：

# 假设原始数据为包含多字段的字典
raw_data = {'f1': '0', 'f2': '1', 'f3': '0'}
# 定义需要转换的布尔字段列表
bool_fields = ['f1', 'f2', 'f3']
for field in bool_fields:
    if field in raw_data:
        raw_data[field] = bool(int(raw_data[field]))

row = Tdata(**raw_data)
sess.add(row)
sess.commit()

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