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

Python筑基之旅-MySQL数据库(三)

news 2025/7/27 21:04:04

一、数据库操作

1、创建

1-1、用mysql-connector-python库

1-2、用PyMySQL库

1-3、用PeeWee库

1-4、用SQLAlchemy库

2、删除

2-1、用mysql-connector-python库

2-2、用PyMySQL库

2-3、用PeeWee库

2-4、用SQLAlchemy库

二、数据表操作

1、创建

1-1、用mysql-connector-python库

1-2、用PyMySQL库

1-3、用PeeWee库

1-4、用SQLAlchemy库

2、删除

2-1、用mysql-connector-python库

2-2、用PyMySQL库

2-3、用PeeWee库

2-4、用SQLAlchemy库

三、推荐阅读

1、Python函数之旅

2、Python算法之旅

3、博客个人主页

一、数据库操作

1、创建

现需在MySQL服务器上新建数据库test_database；同时，在新建的数据test_database中创建数据表test_table。不管是数据库，还是数据表，若不存在，则新建。

1-1、用mysql-connector-python库

import mysql.connector
# 配置数据库连接信息
config = {'user': 'root','password': '123456',  # 在此输入你的MySQL密码，请自行修改'host': '127.0.0.1','database': 'mysql'    # 如果要指定数据库
}
# 连接到MySQL服务器
try:cnx = mysql.connector.connect(**config)cursor = cnx.cursor()# 创建新数据库test_database（如果不存在）create_db_query = "CREATE DATABASE IF NOT EXISTS test_database"cursor.execute(create_db_query)# 切换到新数据库use_db_query = "USE test_database"cursor.execute(use_db_query)# 创建新表test_table（如果不存在）# 假设我们有一个简单的表，包含 id（整数，主键，自增）、name（字符串）和 age（整数）create_table_query = """    CREATE TABLE IF NOT EXISTS test_table (    id INT AUTO_INCREMENT PRIMARY KEY,    name VARCHAR(255) NOT NULL,    age INT    )    """cursor.execute(create_table_query)# 提交事务cnx.commit()
except mysql.connector.Error as err:print(f"Error: {err}")if cnx.is_connected():cnx.rollback()  # 如果需要回滚事务的话
finally:# 关闭游标和连接if cursor:cursor.close()if cnx.is_connected():cnx.close()

1-2、用PyMySQL库

import pymysql
# 配置数据库连接信息
config = {'user': 'root','password': '123456',  # 在此输入你的MySQL密码，请自行修改'host': '127.0.0.1','database': 'mysql'  # 如果要指定数据库，但这里我们稍后会创建新数据库
}
# 连接到MySQL服务器
try:# 注意：PyMySQL没有直接使用**config的方式，需要分别传入参数cnx = pymysql.connect(host=config['host'],user=config['user'],password=config['password'],charset='utf8mb4',  # 可选，添加字符集支持cursorclass=pymysql.cursors.DictCursor)  # 使用字典游标# 创建新数据库test_database（如果不存在）with cnx.cursor() as cursor:create_db_query = "CREATE DATABASE IF NOT EXISTS test_database"cursor.execute(create_db_query)# 切换到新数据库# 注意：PyMySQL没有直接的USE语句，我们需要断开连接并重新连接到新数据库cnx.close()config['database'] = 'test_database'cnx = pymysql.connect(**config, charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor)# 创建新表test_table（如果不存在）# 假设我们有一个简单的表，包含 id（整数，主键，自增）、name（字符串）和 age（整数）with cnx.cursor() as cursor:create_table_query = """  CREATE TABLE IF NOT EXISTS test_table (  id INT AUTO_INCREMENT PRIMARY KEY,  name VARCHAR(255) NOT NULL,  age INT  )  """cursor.execute(create_table_query)# 提交事务cnx.commit()
except pymysql.MySQLError as err:print(f"Error: {err}")if cnx:cnx.rollback()  # 如果需要回滚事务的话（但对于DDL语句，通常没有回滚的必要）
finally:# 关闭游标和连接if cnx:cnx.close()

1-3、用PeeWee库

from peewee import *
# 配置数据库连接信息，这里尝试连接到名为'mysql'的数据库
db = MySQLDatabase('mysql', user='root', password='123456', host='127.0.0.1')
# 尝试连接数据库
try:db.connect()# 创建新数据库test_database（如果不存在），这里通过执行原始SQL语句来实现db.execute_sql("CREATE DATABASE IF NOT EXISTS test_database")# 更改数据库连接对象以连接到新创建的数据库（或已存在的）db = MySQLDatabase('test_database', user='root', password='123456', host='127.0.0.1')db.connect()# 定义模型（对应于数据库表）class TestTable(Model):id = AutoField()  # 自增主键name = CharField(null=False)  # 字符字段，不能为空age = IntegerField(null=True)  # 整数字段，可以为空class Meta:database = db  # 指定该模型使用的数据库连接# 自定义的创建表方法，这里实际上是覆盖了基类Model的create_table方法# 但实际上这里不需要重新定义，因为Peewee会自动在调用create_table时处理提交事务def create_table(cls, fail_silently=False, **options):super(TestTable, cls).create_table(fail_silently, **options)db.commit()  # 这里提交事务其实是不必要的，因为Peewee默认会处理# 调用自定义的create_table方法来创建表（如果不存在）# 但通常我们会直接调用 TestTable.create_table() 而无需修改它TestTable.create_table()
except Exception as e:print(f"Error: {e}")# 如果连接已关闭，这里的关闭操作是多余的，因为db.close()在连接关闭时不会引发错误if db.is_closed():db.close()  # 实际上，这里的检查是多余的，因为db在异常发生前应该是打开的else:db.rollback()  # 如果有事务在执行并且需要回滚的话（但在这个场景中，通常没有显式开启的事务）
finally:# 无论是否发生异常，都要确保数据库连接被关闭if not db.is_closed():db.close()

1-4、用SQLAlchemy库

略，该库本身不支持直接创建数据库，需要借助其他第三方库，如PyMySQL

2、删除

现需在MySQL服务器上删除已有数据库test_database，以下为借助第三方库实现：

2-1、用mysql-connector-python库

import mysql.connector
# MySQL服务器连接参数
config = {'user': 'root','password': '123456',  # 在此输入你的MySQL密码，请自行修改'host': '127.0.0.1','database': 'mysql'    # 如果要指定数据库
}
# 连接到MySQL服务器
try:connection = mysql.connector.connect(**config)cursor = connection.cursor()# 执行SQL语句来删除test_database数据库sql = "DROP DATABASE IF EXISTS test_database"cursor.execute(sql)# 提交事务connection.commit()print("Database test_database deleted successfully.")
except mysql.connector.Error as error:print(f"Error: '{error}'")
finally:# 关闭游标和连接if connection.is_connected():cursor.close()connection.close()

2-2、用PyMySQL库

import pymysql
# MySQL服务器连接参数
config = {'user': 'root',  # 替换为你的MySQL用户名'password': '123456',  # 替换为你的MySQL密码'host': '127.0.0.1',  # 替换为你的MySQL服务器地址'port': 3306,  # 如果你的MySQL不是默认端口，需要指定'charset': 'utf8mb4',  # 设置字符集'cursorclass': pymysql.cursors.DictCursor  # 使用字典游标，这不是必需的
}
# 连接到MySQL服务器（注意这里我们没有指定database参数）
try:# 因为我们要删除数据库，所以不需要连接到特定的数据库connection = pymysql.connect(**{k: v for k, v in config.items() if k != 'database'})with connection.cursor() as cursor:# 执行SQL语句来删除test_database数据库sql = "DROP DATABASE IF EXISTS test_database"cursor.execute(sql)# 提交事务，可省略connection.commit()print("Database test_database deleted successfully.")
except pymysql.MySQLError as error:print(f"Error: '{error}'")
finally:# 关闭连接if connection:connection.close()

2-3、用PeeWee库

略，该库本身不支持直接删除数据库，需要借助其他第三方库，如PyMySQL

2-4、用SQLAlchemy库

略，该库本身不支持直接删除数据库，需要借助其他第三方库，如PyMySQL

二、数据表操作