使用SQLAlchemy向Google Cloud SQL(PostgreSQL)插入数据失败排查

• 优先检查事务提交
  SQLAlchemy默认开启事务，所有写操作（建表、插入）如果没有显式提交，事务结束后会自动回滚。代码里能查询到结果只是当前未提交事务内的临时可见，不会写入实际数据库。

  • 用ORM会话时，操作完成后必须调用 session.commit()
  • 用核心引擎执行SQL时，要手动管理事务：

    with engine.connect() as conn:
        trans = conn.begin()
        try:
            # 执行建表、插入语句
            conn.execute(text("CREATE TABLE test (id INT)"))
            conn.execute(text("INSERT INTO test VALUES (1)"))
            trans.commit()  # 提交事务才会持久化
        except Exception as e:
            trans.rollback()
            raise e
    

• 确认连接的数据库一致性
  检查Python代码的连接字符串，确保和命令行操作的是同一个Cloud SQL实例、同一个数据库。重点核对host、dbname参数，避免误连到其他测试库或实例。

• 验证SQL执行语句的一致性
  开启SQLAlchemy的echo=True参数（创建引擎时添加），打印出实际执行的SQL语句，和命令行执行的对比，确认没有语法差异或执行对象不同。

• 权限与Cloud SQL配置排查（可能性较低）
  虽然命令行能正常执行，但仍可快速确认：

  • Python使用的数据库账号是否拥有CREATE TABLE和INSERT权限
  • Cloud SQL的授权网络是否允许Python运行环境的IP（如果代码无连接报错，这条大概率没问题）

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