产品概述
应用场景
产品架构
实例类型
兼容性说明
使用规范建议
驱动名称 | 类型 | 核心优势 | 适用场景 |
mysql-connector-python | 官方驱动 | 由 Oracle/MySQL 官方维护，兼容性最佳，无额外依赖 | 追求稳定性、无需复杂配置的场景 |
PyMySQL | 纯 Python 实现 | 轻量灵活、源码易读、安装简单、社区活跃 | 快速开发、小型项目、需要自定义扩展的场景 |
mysqlclient | C 语言扩展 | 性能最优（比纯 Python 驱动快20%-50%） | 高并发 OLTP 场景、大数据量读写场景 |
aiomysql | 异步纯 Python | 支持 asyncio 异步 IO，高并发异步场景首选 | FastAPI、aiohttp 等异步 Web 框架项目 |
SQLAlchemy | ORM 框架（依赖底层驱动） | 屏蔽底层驱动差异、支持 SQL 抽象与事务管理 | 中大型项目、需要跨数据库兼容的场景 |
import mysql.connector# 配置数据库连接信息（替换为你的实际配置）db_config = {"host": "10.10.10.10","user": "test","port": 7981,"password": "test123","database": "test"}# 建立连接 + 创建单个游标（复用该游标执行所有SQL）conn = mysql.connector.connect(**db_config)cursor = conn.cursor(buffered=True)# 1. 创建表cursor.execute("""CREATE TABLE IF NOT EXISTS test_table (id INT AUTO_INCREMENT PRIMARY KEY,name VARCHAR(50) NOT NULL,age TINYINT DEFAULT 0) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""")# 2. 插入一行数据并提交事务cursor.execute("INSERT INTO test_table (name, age) VALUES (%s, %s)", ("Tom", 22))conn.commit()# 3. 查询数据并打印cursor.execute("SELECT * FROM test_table WHERE name = %s", ("Tom",))data = cursor.fetchone()print("Query result:", data)# 关闭游标和连接cursor.close()conn.close()
import pymysql# 配置数据库连接信息（替换为你的实际配置）db_config = {"host": "10.10.10.10","user": "test","port": 7981,"password": "test123","database": "test"}# 建立连接 + 创建游标conn = pymysql.connect(**db_config)cursor = conn.cursor()# 1. 创建表cursor.execute("""CREATE TABLE IF NOT EXISTS test_table (id INT AUTO_INCREMENT PRIMARY KEY,name VARCHAR(50) NOT NULL,age TINYINT DEFAULT 0) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""")# 2. 插入一行数据并提交事务cursor.execute("INSERT INTO test_table (name, age) VALUES (%s, %s)", ("Tom", 22))conn.commit()# 3. 查询数据并打印cursor.execute("SELECT * FROM test_table WHERE name = %s", ("Tom",))data = cursor.fetchone()print("Query result:", data)# 关闭游标和连接cursor.close()conn.close()
import MySQLdb# 配置数据库连接信息（替换为你的实际配置）db_config = {"host": "10.10.10.10","user": "test","port": 7981,"password": "test123","database": "test"}# 建立连接 + 创建游标conn = MySQLdb.connect(**db_config)cursor = conn.cursor()# 1. 创建表cursor.execute("""CREATE TABLE IF NOT EXISTS test_table (id INT AUTO_INCREMENT PRIMARY KEY,name VARCHAR(50) NOT NULL,age TINYINT DEFAULT 0) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""")# 2. 插入一行数据并提交事务cursor.execute("INSERT INTO test_table (name, age) VALUES (%s, %s)", ("Tom", 22))conn.commit()# 3. 查询数据并打印cursor.execute("SELECT * FROM test_table WHERE name = %s", ("Tom",))data = cursor.fetchone()print("Query result:", data)# 关闭游标和连接cursor.close()conn.close()
import asyncioimport aiomysql# 配置数据库连接信息（替换为你的实际配置）db_config = {"host": "10.10.10.10","user": "test","port": 7981,"password": "test123","db": "test"}async def main():# 1. 建立异步连接conn = await aiomysql.connect(**db_config)# 2. 创建异步游标cursor = await conn.cursor()# 3. 创建表await cursor.execute("""CREATE TABLE IF NOT EXISTS test_table (id INT AUTO_INCREMENT PRIMARY KEY,name VARCHAR(50) NOT NULL,age TINYINT DEFAULT 0) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""")# 4. 插入一行数据并提交事务await cursor.execute("INSERT INTO test_table (name, age) VALUES (%s, %s)", ("Tom", 22))await conn.commit()# 5. 查询数据并打印await cursor.execute("SELECT * FROM test_table WHERE name = %s", ("Tom",))data = await cursor.fetchone()print("Query result:", data)# 6. 关闭游标和连接await cursor.close()conn.close()# Python 3.6 兼容的异步启动方式（替换 asyncio.run()）if __name__ == "__main__":# 1. 获取默认事件循环loop = asyncio.get_event_loop()# 2. 运行异步主函数loop.run_until_complete(main())# 3. 关闭事件循环loop.close()
from sqlalchemy import create_engine, Column, Integer, Stringfrom sqlalchemy.orm import sessionmakerfrom sqlalchemy.ext.declarative import declarative_base# 1. 基础配置：创建基类 + 数据库连接引擎Base = declarative_base()# 连接字符串格式：mysql+底层驱动://账号:密码@地址:端口/数据库名?字符集engine = create_engine("mysql+mysqldb://test:test123@10.10.10.10:7981/test?charset=utf8mb4")# 2. 定义 ORM 模型（对应 test_table 表）class TestTable(Base):__tablename__ = "test_table" # 表名id = Column(Integer, primary_key=True, autoincrement=True) # 自增主键name = Column(String(50), nullable=False) # 用户名age = Column(Integer, default=0) # 年龄# 3. 创建会话（用于数据库操作）Session = sessionmaker(bind=engine)session = Session()# 4. 创建表（若不存在）Base.metadata.create_all(engine)# 5. 插入一行数据new_data = TestTable(name="Tom", age=22)session.add(new_data)session.commit() # 提交事务# 6. 查询数据并打印query_data = session.query(TestTable).filter(TestTable.name == "Tom").first()print("Query result:", (query_data.id, query_data.name, query_data.age))# 7. 关闭会话session.close()
文档反馈