python sqlalchemy 向 mysql 插入数据的几种方法

以下是使用 SQLAlchemy 向 MySQL 数据库插入数据的详细步骤：

1. 安装依赖

pip install sqlalchemy pymysql

2. 基础配置

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker

# 创建引擎（替换为你的数据库信息）
engine = create_engine('mysql+pymysql://user:password@localhost/dbname', echo=True)

# 创建基类
Base = declarative_base()

# 创建会话类
Session = sessionmaker(bind=engine)
session = Session()

3. 定义数据模型

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    email = Column(String(100))
    
    def __repr__(self):
        return f"<User(name='{self.name}', email='{self.email}')>"

4. 插入数据方式

方式一：使用 ORM 模型插入

# 创建实例
new_user = User(name='Alice', email='alice@example.com')

# 添加到会话
session.add(new_user)

# 提交事务
session.commit()

方式二：使用 Core 直接插入

from sqlalchemy import insert

# 创建插入语句
stmt = insert(User).values(name='Bob', email='bob@example.com')

# 执行插入
with engine.connect() as conn:
    conn.execute(stmt)

5. 批量插入优化

# 生成批量数据
users_data = [
    {'name': 'Charlie', 'email': 'charlie@example.com'},
    {'name': 'David', 'email': 'david@example.com'},
    # 更多数据...
]

# 使用 add_all 批量添加
session.add_all([User(**data) for data in users_data])
session.commit()

# 或者使用 core 的批量插入
stmt = insert(User)
with engine.connect() as conn:
    conn.execute(stmt, users_data)

6. 完整示例

# 初始化表结构（如果不存在）
Base.metadata.create_all(engine)

# 插入单条数据
new_user = User(name='Eve', email='eve@example.com')
session.add(new_user)
session.commit()

# 查询验证
users = session.query(User).all()
print(users)  # 输出：[<User(name='Eve', email='eve@example.com')>, ...]

# 关闭会话
session.close()

注意事项：

1. 事务管理：使用 ORM 时，确保在修改数据后调用 commit()，发生异常时调用 rollback()
2. 连接池：生产环境建议使用连接池配置：

   engine = create_engine(
       'mysql+pymysql://user:pass@host/db',
       pool_size=10,
       max_overflow=20
   )
   

3. 数据类型映射：确保 Python 类型与 MySQL 类型正确对应（如 String 对应 VARCHAR）
4. 批量插入性能：对于大量数据（>1000条），建议分批提交（每批 500-1000 条）

建议在实际使用中结合异常处理机制，例如：

try:
    session.add(new_user)
    session.commit()
except Exception as e:
    session.rollback()
    print(f"Error: {str(e)}")