SQLite数据迁移至PostgreSQL失败：数据未写入排查求助

以下是代码中导致数据未写入的核心问题及修复方案：

1. 未提交PostgreSQL事务 #

psycopg2默认采用手动提交模式，执行INSERT语句后如果不主动提交事务，连接关闭时会自动回滚所有操作，导致数据无法持久化。

修复：在INSERT语句执行后添加事务提交：

cursor.execute(f"""INSERT INTO content.film_work ({column_names_str}) VALUES {bind_values} """)
conn.commit()  # 新增提交操作

2. __post_init__方法未绑定到FilmWork类 #

当前代码中__post_init__是独立函数，没有缩进在FilmWork类内部，导致实例化FilmWork时不会执行字段初始化逻辑，可能出现字段类型不匹配（比如日期格式错误），引发隐性插入失败。

修复：将__post_init__缩进为FilmWork的类方法，同时优化日期类型为PostgreSQL兼容的datetime对象：

@dataclass
class FilmWork:
    created_at: date = None
    updated_at: date = None
    id: uuid.UUID = field(default_factory=uuid.uuid4)
    title: str = ''
    description: str = ''
    creation_date: date = None
    rating: float = 0.0
    type: str = ''
    file_path: str = ''

    def __post_init__(self):
        if self.creation_date is None:
            self.creation_date = datetime.now()
        if self.created_at is None:
            self.created_at = datetime.now()
        if self.updated_at is None:
            self.updated_at = datetime.now()
        if self.description is None:
            self.description = 'Нет описания'
        if self.rating is None:
            self.rating = 0.0

3. 字段映射与批量插入的潜在问题 #

PostgreSQL表字段为created、modified，而FilmWork类对应字段是created_at、updated_at，虽然当前列顺序刚好匹配，但批量插入时若films为空，会生成语法错误的SQL语句，且无错误提示。

优化：添加空数据判断与SQL调试输出：

def save_film_work_to_postgres(films: list):
    # ... 原有代码 ...
    with conn.cursor() as cursor:
        # ... 获取列名代码 ...
        if not films:
            print("没有需要插入的数据")
            return
        
        col_count = ', '.join(['%s'] * len(column_names_list)) 
        bind_values = ','.join(cursor.mogrify(f"({col_count})", astuple(film)).decode('utf-8') for film in films)
        
        # 调试时打印SQL语句，排查语法问题
        sql = f"""INSERT INTO content.film_work ({column_names_str}) VALUES {bind_values} """
        print(f"执行SQL: {sql}")
        
        cursor.execute(sql)
        conn.commit()

4. SQLite数据转FilmWork的异常捕获 #

SQLite的id字段是TEXT类型，若其中存在非法UUID字符串，实例化FilmWork时会抛出异常，导致后续数据插入中断且无提示。

优化：添加数据转换异常捕获：

def copy_from_sqlite():
    with conn_context(db_path) as connection:
        cursor = connection.cursor()
        cursor.execute("SELECT * FROM film_work;")
        result = cursor.fetchall()
        
        films = []
        for film in result:
            try:
                films.append(FilmWork(**dict(film)))
            except Exception as e:
                print(f"处理数据行失败: {dict(film)}, 错误: {e}")
        
        save_film_work_to_postgres(films)

内容的提问来源于stack exchange，提问作者Артём