Python数据库测试及调试技巧

数据库连接测试

连接数据库基础测试

在 Python 中，我们常用 pymysql 库来连接 MySQL 数据库。首先确保你已经安装了 pymysql，可以使用 pip install pymysql 进行安装。

下面是一个简单的连接测试代码示例：

import pymysql

try:
    # 建立数据库连接
    connection = pymysql.connect(
        host='localhost',
        user='root',
        password='password',
        database='test_db'
    )
    print("成功连接到数据库")
    # 关闭连接
    connection.close()
except pymysql.Error as e:
    print(f"连接数据库时出错: {e}")

在上述代码中，pymysql.connect() 方法用于建立数据库连接，它接收 host（主机地址）、user（用户名）、password（密码）和 database（数据库名）等参数。通过 try - except 块捕获可能出现的连接错误，并进行相应的提示。

如果连接的是 PostgreSQL 数据库，可以使用 psycopg2 库。同样，先使用 pip install psycopg2 安装该库，以下是连接测试代码：

import psycopg2

try:
    connection = psycopg2.connect(
        host="localhost",
        database="test_db",
        user="user",
        password="password"
    )
    print("成功连接到 PostgreSQL 数据库")
    connection.close()
except psycopg2.Error as e:
    print(f"连接 PostgreSQL 数据库时出错: {e}")

连接参数测试

1. 主机地址：在测试连接时，主机地址可以是 IP 地址，也可以是域名。如果使用本地数据库，通常是 localhost 或 127.0.0.1。但如果是远程数据库，需要填写正确的远程服务器 IP 地址或域名。

   import pymysql
   try:
       connection = pymysql.connect(
           host='192.168.1.100',  # 假设远程数据库 IP
           user='root',
           password='password',
           database='test_db'
       )
       print("成功连接到远程数据库")
       connection.close()
   except pymysql.Error as e:
       print(f"连接远程数据库时出错: {e}")
   

2. 端口号：不同数据库有默认端口号，MySQL 默认端口是 3306，PostgreSQL 默认端口是 5432。在连接时，如果数据库使用默认端口，可以不指定端口号；若使用非默认端口，则需要明确指定。

   import pymysql
   try:
       connection = pymysql.connect(
           host='localhost',
           user='root',
           password='password',
           database='test_db',
           port = 3307  # 假设 MySQL 使用非默认端口 3307
       )
       print("成功连接到使用非默认端口的数据库")
       connection.close()
   except pymysql.Error as e:
       print(f"连接数据库时出错: {e}")
   

3. 字符集：字符集设置对于处理不同语言文字非常重要。在连接数据库时，可通过 charset 参数指定字符集，如 utf8、gbk 等。

   import pymysql
   try:
       connection = pymysql.connect(
           host='localhost',
           user='root',
           password='password',
           database='test_db',
           charset='utf8'
       )
       print("成功连接并设置字符集")
       connection.close()
   except pymysql.Error as e:
       print(f"连接数据库时出错: {e}")
   

SQL 查询测试

简单查询测试

1. 执行 SELECT 查询：以 MySQL 为例，使用 pymysql 库执行简单的 SELECT 查询。假设数据库中有一个 students 表，包含 id、name 和 age 字段。

   import pymysql
   
   try:
       connection = pymysql.connect(
           host='localhost',
           user='root',
           password='password',
           database='test_db'
       )
       cursor = connection.cursor()
       sql = "SELECT id, name, age FROM students"
       cursor.execute(sql)
       results = cursor.fetchall()
       for row in results:
           id = row[0]
           name = row[1]
           age = row[2]
           print(f"ID: {id}, Name: {name}, Age: {age}")
       cursor.close()
       connection.close()
   except pymysql.Error as e:
       print(f"执行查询时出错: {e}")
   

   在上述代码中，先获取数据库连接，然后创建游标对象 cursor。通过游标执行 SQL 查询语句 SELECT，并使用 fetchall() 方法获取所有查询结果。最后遍历结果并打印。
2. 限制查询结果数量：有时候我们只需要获取部分数据，可以使用 LIMIT 关键字。

   import pymysql
   
   try:
       connection = pymysql.connect(
           host='localhost',
           user='root',
           password='password',
           database='test_db'
       )
       cursor = connection.cursor()
       sql = "SELECT id, name, age FROM students LIMIT 5"
       cursor.execute(sql)
       results = cursor.fetchall()
       for row in results:
           id = row[0]
           name = row[1]
           age = row[2]
           print(f"ID: {id}, Name: {name}, Age: {age}")
       cursor.close()
       connection.close()
   except pymysql.Error as e:
       print(f"执行查询时出错: {e}")
   

   这里的 LIMIT 5 表示只获取前 5 条记录。

条件查询测试

1. 使用 WHERE 子句：通过 WHERE 子句可以筛选符合特定条件的数据。例如，查询年龄大于 18 岁的学生。

   import pymysql
   
   try:
       connection = pymysql.connect(
           host='localhost',
           user='root',
           password='password',
           database='test_db'
       )
       cursor = connection.cursor()
       sql = "SELECT id, name, age FROM students WHERE age > 18"
       cursor.execute(sql)
       results = cursor.fetchall()
       for row in results:
           id = row[0]
           name = row[1]
           age = row[2]
           print(f"ID: {id}, Name: {name}, Age: {age}")
       cursor.close()
       connection.close()
   except pymysql.Error as e:
       print(f"执行查询时出错: {e}")
   

2. 多条件查询：可以使用 AND 和 OR 关键字连接多个条件。例如，查询年龄大于 18 岁且名字以 'J' 开头的学生。

   import pymysql
   
   try:
       connection = pymysql.connect(
           host='localhost',
           user='root',
           password='password',
           database='test_db'
       )
       cursor = connection.cursor()
       sql = "SELECT id, name, age FROM students WHERE age > 18 AND name LIKE 'J%'"
       cursor.execute(sql)
       results = cursor.fetchall()
       for row in results:
           id = row[0]
           name = row[1]
           age = row[2]
           print(f"ID: {id}, Name: {name}, Age: {age}")
       cursor.close()
       connection.close()
   except pymysql.Error as e:
       print(f"执行查询时出错: {e}")
   

   这里的 LIKE 'J%' 表示名字以 'J' 开头。

数据插入测试

单条数据插入测试

1. 使用 VALUES 插入：以 MySQL 为例，向 students 表插入一条新记录。

   import pymysql
   
   try:
       connection = pymysql.connect(
           host='localhost',
           user='root',
           password='password',
           database='test_db'
       )
       cursor = connection.cursor()
       sql = "INSERT INTO students (name, age) VALUES ('Tom', 20)"
       cursor.execute(sql)
       connection.commit()
       print("成功插入一条记录")
       cursor.close()
       connection.close()
   except pymysql.Error as e:
       print(f"插入数据时出错: {e}")
   

   在上述代码中，使用 INSERT INTO...VALUES 语句插入数据，然后通过 connection.commit() 提交事务，使插入操作生效。
2. 使用变量插入：在实际应用中，通常会使用变量来动态插入数据。

   import pymysql
   
   name = 'Jerry'
   age = 22
   
   try:
       connection = pymysql.connect(
           host='localhost',
           user='root',
           password='password',
           database='test_db'
       )
       cursor = connection.cursor()
       sql = "INSERT INTO students (name, age) VALUES (%s, %s)"
       cursor.execute(sql, (name, age))
       connection.commit()
       print("成功插入一条记录")
       cursor.close()
       connection.close()
   except pymysql.Error as e:
       print(f"插入数据时出错: {e}")
   

   这里使用 %s 作为占位符，通过 execute() 方法的第二个参数传递变量值，这样可以有效防止 SQL 注入攻击。

多条数据插入测试

1. 使用列表插入：如果要插入多条记录，可以使用列表形式传递数据。

   import pymysql
   
   data_list = [
       ('Alice', 21),
       ('Bob', 23)
   ]
   
   try:
       connection = pymysql.connect(
           host='localhost',
           user='root',
           password='password',
           database='test_db'
       )
       cursor = connection.cursor()
       sql = "INSERT INTO students (name, age) VALUES (%s, %s)"
       cursor.executemany(sql, data_list)
       connection.commit()
       print("成功插入多条记录")
       cursor.close()
       connection.close()
   except pymysql.Error as e:
       print(f"插入数据时出错: {e}")
   

   使用 executemany() 方法可以一次性执行多条插入语句，提高插入效率。
2. 从文件读取数据插入：假设数据存储在一个文本文件中，每行包含姓名和年龄，以逗号分隔。

   import pymysql
   
   try:
       connection = pymysql.connect(
           host='localhost',
           user='root',
           password='password',
           database='test_db'
       )
       cursor = connection.cursor()
       sql = "INSERT INTO students (name, age) VALUES (%s, %s)"
       with open('students_data.txt', 'r') as file:
           for line in file:
               name, age = line.strip().split(',')
               cursor.execute(sql, (name, int(age)))
       connection.commit()
       print("成功从文件插入数据")
       cursor.close()
       connection.close()
   except pymysql.Error as e:
       print(f"插入数据时出错: {e}")
   

   上述代码逐行读取文件内容，解析出姓名和年龄后插入数据库。

数据更新测试

简单更新测试

1. 无条件更新：不使用 WHERE 子句的更新操作会影响表中的所有记录。例如，将所有学生的年龄加 1。

   import pymysql
   
   try:
       connection = pymysql.connect(
           host='localhost',
           user='root',
           password='password',
           database='test_db'
       )
       cursor = connection.cursor()
       sql = "UPDATE students SET age = age + 1"
       cursor.execute(sql)
       connection.commit()
       print("成功更新所有学生年龄")
       cursor.close()
       connection.close()
   except pymysql.Error as e:
       print(f"更新数据时出错: {e}")
   

   这种操作要谨慎使用，因为它会改变整个表的数据。
2. 有条件更新：通过 WHERE 子句指定更新条件，如将名字为 'Tom' 的学生年龄加 1。

   import pymysql
   
   try:
       connection = pymysql.connect(
           host='localhost',
           user='root',
           password='password',
           database='test_db'
       )
       cursor = connection.cursor()
       sql = "UPDATE students SET age = age + 1 WHERE name = 'Tom'"
       cursor.execute(sql)
       connection.commit()
       print("成功更新指定学生年龄")
       cursor.close()
       connection.close()
   except pymysql.Error as e:
       print(f"更新数据时出错: {e}")
   

复杂更新测试

1. 多字段更新：可以同时更新多个字段。例如，将名字为 'Alice' 的学生年龄加 1 并将名字改为 'Alice Smith'。

   import pymysql
   
   try:
       connection = pymysql.connect(
           host='localhost',
           user='root',
           password='password',
           database='test_db'
       )
       cursor = connection.cursor()
       sql = "UPDATE students SET age = age + 1, name = 'Alice Smith' WHERE name = 'Alice'"
       cursor.execute(sql)
       connection.commit()
       print("成功更新多个字段")
       cursor.close()
       connection.close()
   except pymysql.Error as e:
       print(f"更新数据时出错: {e}")
   

2. 基于子查询更新：子查询可以在更新语句中用于获取更复杂的条件数据。假设存在另一个表 students_bonus，记录了每个学生的年龄奖励值，我们要根据这个表更新 students 表中的年龄。

   import pymysql
   
   try:
       connection = pymysql.connect(
           host='localhost',
           user='root',
           password='password',
           database='test_db'
       )
       cursor = connection.cursor()
       sql = """
           UPDATE students
           SET age = students.age + (
               SELECT bonus_age
               FROM students_bonus
               WHERE students.id = students_bonus.student_id
           )
       """
       cursor.execute(sql)
       connection.commit()
       print("成功基于子查询更新数据")
       cursor.close()
       connection.close()
   except pymysql.Error as e:
       print(f"更新数据时出错: {e}")
   

   上述代码通过子查询从 students_bonus 表获取每个学生对应的年龄奖励值，并更新 students 表中的年龄。

数据删除测试

简单删除测试

1. 无条件删除：不使用 WHERE 子句的删除操作会删除表中的所有记录。例如，删除 students 表中的所有学生记录。

   import pymysql
   
   try:
       connection = pymysql.connect(
           host='localhost',
           user='root',
           password='password',
           database='test_db'
       )
       cursor = connection.cursor()
       sql = "DELETE FROM students"
       cursor.execute(sql)
       connection.commit()
       print("成功删除所有学生记录")
       cursor.close()
       connection.close()
   except pymysql.Error as e:
       print(f"删除数据时出错: {e}")
   

   这种操作极其危险，执行前一定要确认是否真的需要删除所有数据。
2. 有条件删除：通过 WHERE 子句指定删除条件，如删除年龄大于 30 岁的学生记录。

   import pymysql
   
   try:
       connection = pymysql.connect(
           host='localhost',
           user='root',
           password='password',
           database='test_db'
       )
       cursor = connection.cursor()
       sql = "DELETE FROM students WHERE age > 30"
       cursor.execute(sql)
       connection.commit()
       print("成功删除指定学生记录")
       cursor.close()
       connection.close()
   except pymysql.Error as e:
       print(f"删除数据时出错: {e}")
   

关联删除测试

1. 基于表关联删除：假设有两个表 students 和 students_courses，students_courses 表记录了学生选修的课程，其中包含 student_id 字段关联到 students 表的 id 字段。现在要删除所有选修了某门课程（假设课程编号为 101）的学生记录。

   import pymysql
   
   try:
       connection = pymysql.connect(
           host='localhost',
           user='root',
           password='password',
           database='test_db'
       )
       cursor = connection.cursor()
       sql = """
           DELETE FROM students
           WHERE id IN (
               SELECT student_id
               FROM students_courses
               WHERE course_id = 101
           )
       """
       cursor.execute(sql)
       connection.commit()
       print("成功基于表关联删除数据")
       cursor.close()
       connection.close()
   except pymysql.Error as e:
       print(f"删除数据时出错: {e}")
   

   上述代码通过子查询获取选修了课程编号为 101 的学生 id，然后在 students 表中删除这些学生记录。
2. 级联删除：在数据库设计时，如果设置了外键关系并启用级联删除，当删除主表中的记录时，相关联的从表记录也会自动删除。例如，在创建 students_courses 表时设置外键关系如下：

   CREATE TABLE students_courses (
       id INT AUTO_INCREMENT PRIMARY KEY,
       student_id INT,
       course_id INT,
       FOREIGN KEY (student_id) REFERENCES students(id) ON DELETE CASCADE
   );
   

   这里的 ON DELETE CASCADE 表示当删除 students 表中的学生记录时，students_courses 表中与之关联的记录也会自动删除。在 Python 中执行删除操作时，只需删除 students 表中的记录，相关联的 students_courses 表记录会自动处理。

   import pymysql
   
   try:
       connection = pymysql.connect(
           host='localhost',
           user='root',
           password='password',
           database='test_db'
       )
       cursor = connection.cursor()
       sql = "DELETE FROM students WHERE id = 1"
       cursor.execute(sql)
       connection.commit()
       print("成功删除学生记录及相关联课程记录")
       cursor.close()
       connection.close()
   except pymysql.Error as e:
       print(f"删除数据时出错: {e}")
   

数据库调试技巧

错误信息分析

1. 连接错误：当连接数据库失败时，pymysql.Error 或 psycopg2.Error 会提供详细的错误信息。例如，“Access denied for user 'root'@'localhost' (using password: YES)” 表示用户名或密码错误。此时需要检查用户名和密码是否正确，以及数据库用户权限设置。
2. SQL 语法错误：如果执行 SQL 语句时出现语法错误，错误信息会指出大致的错误位置。例如，“You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near 'xxx' at line 1”，这里的 'xxx' 附近就是可能出错的位置。仔细检查 SQL 语句的语法，包括关键字拼写、标点符号等。
3. 数据类型不匹配错误：当插入或更新数据时，如果数据类型与表结构定义不匹配，会出现错误。例如，将字符串类型的数据插入到整数类型的字段中。错误信息可能类似于 “Incorrect integer value: 'abc' for column 'age' at row 1”，此时需要检查数据类型并进行相应的转换。

使用日志记录

1. 数据库操作日志：可以使用 Python 的 logging 模块记录数据库操作日志。在执行数据库操作前后记录关键信息，如 SQL 语句、执行时间等。

   import pymysql
   import logging
   
   logging.basicConfig(level=logging.INFO)
   
   try:
       connection = pymysql.connect(
           host='localhost',
           user='root',
           password='password',
           database='test_db'
       )
       cursor = connection.cursor()
       sql = "SELECT id, name, age FROM students"
       logging.info(f"即将执行 SQL: {sql}")
       start_time = time.time()
       cursor.execute(sql)
       results = cursor.fetchall()
       end_time = time.time()
       logging.info(f"查询完成，耗时: {end_time - start_time} 秒")
       for row in results:
           id = row[0]
           name = row[1]
           age = row[2]
           print(f"ID: {id}, Name: {name}, Age: {age}")
       cursor.close()
       connection.close()
   except pymysql.Error as e:
       logging.error(f"执行查询时出错: {e}")
   

   上述代码通过 logging 模块记录了 SQL 语句的执行情况和可能出现的错误，方便调试和性能分析。
2. 连接日志：对于数据库连接相关的操作，也可以记录日志。例如，记录连接的建立和关闭时间，以及连接过程中出现的错误。

   import pymysql
   import logging
   import time
   
   logging.basicConfig(level=logging.INFO)
   
   try:
       start_time = time.time()
       connection = pymysql.connect(
           host='localhost',
           user='root',
           password='password',
           database='test_db'
       )
       end_time = time.time()
       logging.info(f"成功建立数据库连接，耗时: {end_time - start_time} 秒")
       connection.close()
       logging.info("数据库连接已关闭")
   except pymysql.Error as e:
       logging.error(f"连接数据库时出错: {e}")
   

调试工具使用

1. 使用 pdb 调试：Python 的 pdb 模块是一个内置的调试工具，可以在数据库操作代码中设置断点进行调试。

   import pymysql
   import pdb
   
   try:
       connection = pymysql.connect(
           host='localhost',
           user='root',
           password='password',
           database='test_db'
       )
       cursor = connection.cursor()
       sql = "SELECT id, name, age FROM students"
       pdb.set_trace()  # 设置断点
       cursor.execute(sql)
       results = cursor.fetchall()
       for row in results:
           id = row[0]
           name = row[1]
           age = row[2]
           print(f"ID: {id}, Name: {name}, Age: {age}")
       cursor.close()
       connection.close()
   except pymysql.Error as e:
       print(f"执行查询时出错: {e}")
   

   当程序执行到 pdb.set_trace() 时，会暂停执行，进入调试模式。此时可以查看变量值、单步执行代码等，方便找出问题所在。
2. 数据库客户端工具辅助调试：使用数据库客户端工具，如 MySQL Workbench 或 pgAdmin，手动执行 SQL 语句，查看执行结果和错误信息。这些工具通常具有友好的界面，能更直观地查看数据库结构、数据和执行计划。例如，在 MySQL Workbench 中，可以直接编写和执行 SQL 语句，查看查询结果集，以及分析执行计划，帮助优化 SQL 语句。同时，这些工具还能方便地查看数据库的各种状态信息，如连接数、锁状态等，对于调试数据库性能问题非常有帮助。

事务处理调试

1. 手动回滚测试：在测试事务处理时，可以故意引发错误，观察事务是否正确回滚。例如，在一个事务中插入两条记录，第二条记录故意插入错误数据类型，查看第一条记录是否也不会插入成功。

   import pymysql
   
   try:
       connection = pymysql.connect(
           host='localhost',
           user='root',
           password='password',
           database='test_db'
       )
       cursor = connection.cursor()
       try:
           sql1 = "INSERT INTO students (name, age) VALUES ('Eve', 25)"
           cursor.execute(sql1)
           sql2 = "INSERT INTO students (name, age) VALUES ('Frank', 'twenty')"  # 故意错误的数据类型
           cursor.execute(sql2)
           connection.commit()
       except pymysql.Error as e:
           connection.rollback()
           print(f"事务回滚，错误: {e}")
       cursor.close()
       connection.close()
   except pymysql.Error as e:
       print(f"连接或其他错误: {e}")
   

   上述代码中，由于第二条插入语句数据类型错误，事务会回滚，第一条插入语句也不会生效。
2. 事务嵌套调试：在存在事务嵌套的情况下，调试事务的提交和回滚逻辑。例如，外层事务包含内层事务，内层事务出现错误时，确保外层事务也能正确回滚。

   import pymysql
   
   try:
       connection = pymysql.connect(
           host='localhost',
           user='root',
           password='password',
           database='test_db'
       )
       outer_cursor = connection.cursor()
       try:
           outer_sql1 = "INSERT INTO students (name, age) VALUES ('Grace', 24)"
           outer_cursor.execute(outer_sql1)
           inner_cursor = connection.cursor()
           try:
               inner_sql = "INSERT INTO students (name, age) VALUES ('Hank', 'thirty')"  # 故意错误的数据类型
               inner_cursor.execute(inner_sql)
               connection.commit()
           except pymysql.Error as e:
               connection.rollback()
               print(f"内层事务回滚，错误: {e}")
           inner_cursor.close()
           outer_sql2 = "INSERT INTO students (name, age) VALUES ('Ivy', 26)"
           outer_cursor.execute(outer_sql2)
           connection.commit()
       except pymysql.Error as e:
           connection.rollback()
           print(f"外层事务回滚，错误: {e}")
       outer_cursor.close()
       connection.close()
   except pymysql.Error as e:
       print(f"连接或其他错误: {e}")
   

   在这个例子中，内层事务出现错误回滚后，外层事务也会回滚，确保数据的一致性。

通过上述全面的数据库测试及调试技巧，开发者能够更好地确保 Python 与数据库交互的正确性和稳定性，提高开发效率和应用程序的质量。无论是简单的连接测试，还是复杂的事务处理调试，都需要细致的分析和实践，以应对各种可能出现的问题。