PyMySQL 概述
- 纯 Python 实现:不依赖于 C 扩展,易于安装和使用。
- 兼容性好:与 MySQL 数据库的多种版本兼容。
- 功能丰富:支持大多数 MySQL 数据库特性,包括事务、游标、连接池等。
- 支持 Python 3:自始至终为 Python 3 进行优化。
安装 PyMySQL
可以使用 pip
来安装 PyMySQL
。在终端或命令提示符中输入以下命令:
pip install PyMySQL
基本用法
连接到 MySQL 数据库
使用 pymysql
库创建与 MySQL 的连接:
import pymysql
# 连接到数据库
connection = pymysql.connect(
host='localhost', # MySQL 服务器地址
user='your_username', # 用户名
password='your_password', # 密码
database='your_database', # 数据库名
charset='utf8mb4', # 字符集
cursorclass=pymysql.cursors.DictCursor # 返回字典类型,默认为元祖类型
)
创建游标
通过连接对象创建游标,用于执行 SQL 查询:
cursor = connection.cursor()
增删改查操作
1. 创建表
可以通过 CREATE TABLE
SQL 语句创建表:
create_table_query = """
CREATE TABLE IF NOT EXISTS users (
id INT AUTO_INCREMENT PRIMARY KEY,
username VARCHAR(100) NOT NULL,
email VARCHAR(100) NOT NULL
)
"""
cursor.execute(create_table_query)
2. 插入数据
使用 INSERT INTO
语句添加数据:
insert_query = "INSERT INTO users (username, email) VALUES (%s, %s)"
cursor.execute(insert_query, ('Alice', 'alice@example.com'))
connection.commit() # 提交事务
代码分析
insert_query = "INSERT INTO users (username, email) VALUES (%s, %s)"
VALUES 后面的部分是占位符。%s 是一个占位符,它们会在后续的 execute 方法中被实际的值所替代。
使用占位符的好处是可以安全地插入数据,而不必担心 SQL 注入。数据库会处理值的转义和类型安全。
接下来,定义了插入的 SQL 语句并将其赋值给 insert_query
变量。
cursor.execute(insert_query, ('Alice', 'alice@example.com'))
执行插入操作
通过 connection.cursor()
创建的游标对象,允许我们通过cursor.execute()与数据库交互。
('Alice', 'alice@example.com')
:第二个参数是一个元组,其中包含了实际插入的值。这是实际要替换占位符 %s
的地方:'Alice'
会被插入到 username
列。'alice@example.com'
会被插入到 email
列。
提交事务
connection.commit():
在执行插入、更新或删除等操作后,这一行代码是非常重要的。它会将所有未提交的事务记录到数据库中,确保数据持久化。
在数据库中,执行 DML 操作(如 INSERT
)时,默认情况下是在一个事务中。如果不调用 commit()
,这条操作不会被永久保存到数据库中。
commit()
确保了从当前事务中所有已做的更改都会被写入数据库。如果在 commit()
之前发生了错误或异常,您可以调用 rollback()
来撤销目前为止所做的所有操作。
3. 查询数据
使用 SELECT
语句从表中获取数据:
select_query = "SELECT * FROM users"
cursor.execute(select_query)
# 获取查询结果
results = cursor.fetchall() # 获取所有结果
for row in results:
print(row) # 每一行以字典形式输出
除了 fetchall()
方法之外,pymysql
还提供了其他几种方法来获取查询结果。以下是 pymysql
中常用的获取结果的方法:
1. fetchall()
- 描述:获取查询结果中的所有行,并返回一个包含每一行数据的列表。
- 示例:
results = cursor.fetchall()
2. fetchone()
- 描述:获取查询结果中的一行数据。每次调用该方法,都会返回结果集中下一行的数据。如果没有更多的行可用,则返回
None
。 - 示例:
row = cursor.fetchone()
while row:
print(row) # 处理当前行
row = cursor.fetchone() # 获取下一行
3. fetchmany(size)
- 描述:获取指定数量(
size
)的行数据,并返回一个包含这些行的列表。如果没有足够的行可用,返回可能少于size
的结果列表。 - 示例:
results = cursor.fetchmany(5) # 获取前 5 行
for row in results:
print(row)
4. 更新数据
使用 UPDATE
语句更新已有数据:
update_query = "UPDATE users SET email = %s WHERE username = %s"
cursor.execute(update_query, ('alice_new@example.com', 'Alice'))
connection.commit() # 提交事务
5. 删除数据
使用 DELETE
语句删除数据:
delete_query = "DELETE FROM users WHERE username = %s"
cursor.execute(delete_query, ('Alice',))
connection.commit() # 提交事务
关闭连接
完成操作后,记得关闭游标和连接:
cursor.close()
connection.close()
异常处理
在执行数据库操作时,应考虑异常处理:
try:
connection = pymysql.connect(...)
cursor = connection.cursor()
cursor.execute("...") # 你的操作
except pymysql.MySQLError as e:
print(f"Error: {e}")
finally:
if connection:
connection.close() # 确保连接被关闭
连接池
为了更高效地处理连接,可以使用连接池。PyMySQL
本身没有提供连接池,但可以与其他库(如 sqlalchemy
或 DBUtils
)结合使用。
完整 demo
import pymysql
def main():
# 连接到数据库
connection = pymysql.connect(
host='localhost', # MySQL 服务器地址
user='your_username', # 替换为你的用户名
password='your_password', # 替换为你的密码
database='your_database', # 替换为你的数据库名
charset='utf8mb4', # 字符集
cursorclass=pymysql.cursors.DictCursor # 返回字典类型
)
try:
cursor = connection.cursor()
# 1. 创建表
create_table_query = """
CREATE TABLE IF NOT EXISTS users (
id INT AUTO_INCREMENT PRIMARY KEY,
username VARCHAR(100) NOT NULL,
email VARCHAR(100) NOT NULL
)
"""
cursor.execute(create_table_query)
print("Table created successfully.")
# 2. 插入数据
insert_query = "INSERT INTO users (username, email) VALUES (%s, %s)"
cursor.execute(insert_query, ('Alice', 'alice@example.com'))
cursor.execute(insert_query, ('Bob', 'bob@example.com'))
connection.commit() # 提交事务
print("Inserted 2 users successfully.")
# 3. 查询数据
select_query = "SELECT * FROM users"
cursor.execute(select_query)
results = cursor.fetchall() # 获取所有结果
print("Users in the database:")
for row in results:
print(row) # 输出每一行以字典形式
# 4. 更新数据
update_query = "UPDATE users SET email = %s WHERE username = %s"
cursor.execute(update_query, ('alice_new@example.com', 'Alice'))
connection.commit() # 提交事务
print("Updated Alice's email successfully.")
# 查询以验证更新
cursor.execute(select_query)
results = cursor.fetchall()
print("Updated users in the database:")
for row in results: # 此时 results 是一个列表,row 是列表下的每一个字典(SQL 中体现为每一行数据)
print(row)
# 5. 删除数据
delete_query = "DELETE FROM users WHERE username = %s"
cursor.execute(delete_query, ('Bob',))
connection.commit() # 提交事务
print("Deleted user Bob successfully.")
except pymysql.MySQLError as e:
print(f"Error: {e}")
finally:
# 关闭游标和连接
cursor.close()
connection.close()
print("Connection closed.")
if __name__ == "__main__":
main()

发布者:LJH,转发请注明出处:https://www.ljh.cool/42927.html