### Python 快速生成单条数据并插入到 MySQL 的方法 以下是一个使用 `SQLAlchemy` 和原生 `PyMySQL` 的代码示例，展示如何快速生成一条数据并将其插入到 MySQL 数据库中。 #### 使用 SQLAlchemy ```python from sqlalchemy import create_engine, Column, Integer, String, Float from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker # 定义数据库模型 Base = declarative_base() class Product(Base): __tablename__ = 'products' id = Column(Integer, primary_key=True, autoincrement=True) name = Column(String(50), nullable=False) price = Column(Float, nullable=False) # 创建数据库连接 engine = create_engine("mysql+pymysql://root:password@127.0.0.1:3306/learningsql?charset=utf8", pool_size=5, max_overflow=0) Session = sessionmaker(bind=engine) session = Session() # 插入单条数据 def insert_single_data(name, price): product = Product(name=name, price=price) session.add(product) session.commit() print(f"Inserted data: Name={name}, Price={price}") if __name__ == "__main__": insert_single_data("SampleProduct", 99.99) ``` 上述代码通过 SQLAlchemy ORM 定义了一个 `Product` 模型，并创建了一条记录插入到数据库中[^1]。 #### 使用 PyMySQL 如果不想使用 ORM，可以直接使用 `PyMySQL` 库执行 SQL 插入语句。 ```python import pymysql def insert_single_data_pymysql(name, price): connection = pymysql.connect( host='127.0.0.1', user='root', password='password', database='learningsql', charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor ) try: with connection.cursor() as cursor: sql = "INSERT INTO products (name, price) VALUES (%s, %s)" cursor.execute(sql, (name, price)) connection.commit() print(f"Inserted data: Name={name}, Price={price}") finally: connection.close() if __name__ == "__main__": insert_single_data_pymysql("SampleProduct", 99.99) ``` 此代码使用 `PyMySQL` 手动构建 SQL 查询并插入数据[^4]。 #### 异步插入（可选） 如果需要异步操作，可以使用 `aiomysql`。 ```python import asyncio import aiomysql async def insert_single_data_aiomysql(name, price): pool = await aiomysql.create_pool( host='127.0.0.1', user='root', password='password', db='learningsql', charset='utf8mb4', minsize=1, maxsize=10 ) async with pool.acquire() as conn: async with conn.cursor() as cursor: await cursor.execute("INSERT INTO products (name, price) VALUES (%s, %s)", (name, price)) await conn.commit() print(f"Inserted data: Name={name}, Price={price}") pool.close() await pool.wait_closed() if __name__ == "__main__": asyncio.run(insert_single_data_aiomysql("SampleProduct", 99.99)) ``` 此代码展示了如何通过 `aiomysql` 实现异步插入操作[^3]。 ### 注意事项 - 确保数据库连接字符串中的用户名、密码和数据库名称正确无误。 - 如果使用 `SQLAlchemy`，需要先定义模型并确保表结构已存在。 - 对于敏感信息如密码，建议使用环境变量或配置文件存储，而不是硬编码在代码中。