当前位置：首页 > news >正文

多种方式访问mysql的对比分析

news 2025/1/5 15:47:32

对数据库的操作可谓是每个项目必备的功能，技能上也没什么难度，今天风云梳理了一下python对MySql的访问，作了一个详尽的分析，大家可以根据项目的实际情况，按需要采用，python 访问 MySQL 数据库的方式有多种，主要包括以下几种方式：

1、使用 pymysql

pymysql 是一个纯 python 的 MySQL 客户端库，用于与 MySQL 数据库交互。

优点

轻量级，安装简单。
支持 MySQL 的大部分功能。
支持手动执行 SQL 语句，便于学习和调试。

缺点

缺乏高层次的抽象，需要手动构建 SQL 语句，容易出错。
批量插入和复杂查询的效率较低。

代码示例

import pymysqlconn = pymysql.connect(host='localhost', user='root', password='password', database='testdb')cursor = conn.cursor()
cursor.execute("SELECT * FROM users")rows = cursor.fetchall()for row in rows:print(row)cursor.close()
conn.close()

2、使用 mysql-connector-python

mysql-connector-python 是 MySQL 官方提供的纯 python 驱动程序。

优点

官方支持，更新及时，兼容性好。
内置 MySQL 功能支持，例如 SSL、连接池等。

缺点

性能略逊于第三方库（如 pymysql）。
社区生态和第三方文档相对较少。

代码示例

import mysql.connectorconn = mysql.connector.connect(host='localhost', user='root', password='password', database='testdb')cursor = conn.cursor()
cursor.execute("SELECT * FROM users")rows = cursor.fetchall()for row in rows:print(row)cursor.close()
conn.close()

3、使用 SQLAlchemy

SQLAlchemy 是一个功能强大的数据库工具，它支持关系数据库的对象关系映射（ORM）和手动执行 SQL。

优点

支持 ORM 模型，能够将表映射为类，简化复杂查询。
可移植性强，支持多种数据库（MySQL、PostgreSQL、SQLite 等）。
提供连接池功能，适用于高并发场景。

缺点

学习曲线较陡，初学者需要掌握 ORM 和底层 SQLAlchemy Core。
对简单项目可能显得过于复杂。

代码示例

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmakerBase = declarative_base()class User(Base):__tablename__ = 'users'id = Column(Integer, primary_key=True)name = Column(String(50))age = Column(Integer)engine = create_engine('mysql+pymysql://root:password@localhost/testdb')Session = sessionmaker(bind=engine)session = Session()users = session.query(User).all()for user in users:print(user.name, user.age)

4、使用 Tortoise ORM

Tortoise ORM 是一种异步 ORM，主要用于与异步框架（如 FastAPI、Sanic）配合。

优点

支持异步操作，性能优于传统 ORM。
易于集成到现代异步 Web 框架中。

缺点

异步编程有一定学习成本。
支持的功能尚不如 SQLAlchemy 丰富。

代码示例

from tortoise.models import Model
from tortoise import Tortoise, fields, run_asyncclass User(Model):id = fields.IntField(pk=True)name = fields.CharField(max_length=50)age = fields.IntField()async def run():await Tortoise.init(db_url='mysql://root:password@localhost/testdb', modules={'models': ['__main__']})await Tortoise.generate_schemas()users = await User.all()for user in users:print(user.name, user.age)run_async(run())

6、优缺点对比总结

方式	优点	缺点	适用场景
pymysql	轻量级，简单易用	手动 SQL 构造容易出错，适合简单项目	适合初学者和快速开发
mysql-connector-python	官方支持，功能全面	社区生态较弱，性能一般	需要使用官方功能或稳定支持的项目
SQLAlchemy	功能强大，支持 ORM 和连接池	学习曲线陡峭，配置复杂	数据模型复杂且需要多数据库支持
Tortoise ORM	支持异步，适合现代异步框架	功能尚不完善，适合异步框架	异步 Web 项目（FastAPI、Sanic 等）

7、完整功能封装

这里风云帮大家封装了一个完整的MySQL数据库操作的类，采用pymysql库，包括连接、增、删、改、查，以及事务的提交与回滚等功能。引入此类，大家可以轻松操作MySQL。

代码实现

import pymysqlclass MySQLDatabase:"""MySQLDatabase: 封装对 MySQL 数据库的常用操作。"""def __init__(self, host, user, password, database, charset='utf8mb4', autocommit=True):"""初始化数据库连接。:param host: 数据库主机地址:param user: 数据库用户名:param password: 数据库密码:param database: 数据库名称:param charset: 字符集:param autocommit: 是否自动提交事务"""self.host = hostself.user = userself.password = passwordself.database = databaseself.charset = charsetself.autocommit = autocommitself.connection = Noneself.cursor = Nonedef connect(self):"""连接到数据库。"""try:self.connection = pymysql.connect(host=self.host,user=self.user,password=self.password,database=self.database,charset=self.charset,autocommit=self.autocommit,)self.cursor = self.connection.cursor()print("数据库连接成功！")except pymysql.MySQLError as e:print(f"数据库连接失败：{e}")raisedef close(self):"""关闭数据库连接。"""if self.cursor:self.cursor.close()if self.connection:self.connection.close()print("数据库连接已关闭。")def execute(self, query, params=None):"""执行单条 SQL 语句。:param query: SQL 查询语句:param params: 查询参数:return: 受影响的行数"""try:self.cursor.execute(query, params)return self.cursor.rowcountexcept pymysql.MySQLError as e:print(f"SQL 执行失败：{e}")raisedef fetchall(self, query, params=None):"""执行查询并返回所有结果。:param query: SQL 查询语句:param params: 查询参数:return: 查询结果"""try:self.cursor.execute(query, params)return self.cursor.fetchall()except pymysql.MySQLError as e:print(f"查询失败：{e}")raisedef fetchone(self, query, params=None):"""执行查询并返回单条结果。:param query: SQL 查询语句:param params: 查询参数:return: 查询结果"""try:self.cursor.execute(query, params)return self.cursor.fetchone()except pymysql.MySQLError as e:print(f"查询失败：{e}")raisedef commit(self):"""提交事务。"""try:self.connection.commit()print("事务已提交。")except pymysql.MySQLError as e:print(f"事务提交失败：{e}")raisedef rollback(self):"""回滚事务。"""try:self.connection.rollback()print("事务已回滚。")except pymysql.MySQLError as e:print(f"事务回滚失败：{e}")raise# 使用示例if __name__ == "__main__":# 初始化数据库实例db = MySQLDatabase(host="localhost",user="root",password="yourpassword",database="testdb",autocommit=False,  # 使用手动事务)try:# 连接到数据库db.connect()# 创建表create_table_query = """CREATE TABLE IF NOT EXISTS users (id INT AUTO_INCREMENT PRIMARY KEY,name VARCHAR(50) NOT NULL,age INT NOT NULL,email VARCHAR(100),created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)"""db.execute(create_table_query)db.commit()# 插入数据insert_query = "INSERT INTO users (name, age, email) VALUES (%s, %s, %s)"db.execute(insert_query, ('Alice', 25, 'alice@example.com'))db.execute(insert_query, ('Bob', 30, 'bob@example.com'))db.commit()# 查询数据select_query = "SELECT * FROM users"results = db.fetchall(select_query)print("查询结果：")for row in results:print(row)# 更新数据update_query = "UPDATE users SET age = %s WHERE name = %s"db.execute(update_query, (26, 'Alice'))db.commit()# 删除数据delete_query = "DELETE FROM users WHERE name = %s"db.execute(delete_query, ('Bob',))db.commit()except Exception as e:print(f"发生异常：{e}")db.rollback()finally:db.close()

运行结果

运行上述代码后，程序会：