目录
一、数据库操作
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库
略，该库本身不支持直接创建数据库，需要借助其他第三方库，如PyMySQL2、删除
现需在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库
略，该库本身不支持直接删除数据库，需要借助其他第三方库，如PyMySQL2-4、用SQLAlchemy库
略，该库本身不支持直接删除数据库，需要借助其他第三方库，如PyMySQL二、数据表操作
1、创建
在MySQL服务器上已有数据库test_database，需在该数据库中创建数据表myelsa_table，以下为借助第三方库实现：
1-1、用mysql-connector-python库
import mysql.connector
# 数据库连接配置
config = {
'user': 'root', # 替换为你的MySQL用户名
'password': '123456', # 替换为你的MySQL密码
'host': '127.0.0.1', # 如果数据库在远程服务器上，请替换为相应的主机名或IP地址
'database': 'test_database', # 数据库名
'raise_on_warnings': True
}
# 连接到数据库
cnx = mysql.connector.connect(**config)
cursor = cnx.cursor()
# 创建数据表的SQL语句
create_table_query = """
CREATE TABLE IF NOT EXISTS myelsa_table (
name VARCHAR(255) NOT NULL,
ID_Card VARCHAR(255) NOT NULL,
age INT NOT NULL,
city VARCHAR(255) NOT NULL,
PRIMARY KEY (ID_Card) -- 如果ID_Card字段是唯一的，可以设为主键，否则可以移除这行
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
"""
# 执行SQL语句
cursor.execute(create_table_query)
# 提交更改（如果使用了事务）
cnx.commit()
# 关闭连接
cursor.close()
cnx.close()
print("Table myelsa_table created successfully!")1-2、用PyMySQL库
import pymysql
# 数据库连接配置
config = {
'user': 'root', # 替换为你的MySQL用户名
'password': '123456', # 替换为你的MySQL密码
'host': '127.0.0.1', # 如果数据库在远程服务器上，请替换为相应的主机名或IP地址
'database': 'test_database', # 数据库名
'charset': 'utf8mb4', # 字符集设置
'cursorclass': pymysql.cursors.DictCursor # 使用字典游标
}
# 连接到数据库
connection = pymysql.connect(**config)
try:
with connection.cursor() as cursor:
# 创建数据表的SQL语句
create_table_query = """
CREATE TABLE IF NOT EXISTS myelsa_table (
name VARCHAR(255) NOT NULL,
ID_Card VARCHAR(255) NOT NULL,
age INT NOT NULL,
city VARCHAR(255) NOT NULL,
PRIMARY KEY (ID_Card) -- 如果ID_Card字段是唯一的，可以设为主键
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
"""
# 执行SQL语句
cursor.execute(create_table_query)
# 提交更改（对于自动提交模式，这一步可能不是必须的）
connection.commit()
finally:
connection.close()
print("Table myelsa_table created successfully!")1-3、用PeeWee库
from peewee import *
# 数据库连接配置
db = MySQLDatabase('test_database', user='root', password='123456', host='127.0.0.1')
# 定义模型
class MyElsaTable(Model):
name = CharField(max_length=255, null=False)
ID_Card = CharField(max_length=255, unique=True) # 假设ID_Card是唯一的
age = IntegerField(null=False)
city = CharField(max_length=255, null=False)
class Meta:
database = db
# 连接到数据库
db.connect()
# 创建表（如果表不存在）
db.create_tables([MyElsaTable])
# 关闭数据库连接（在 Peewee 中，通常不需要显式关闭连接，因为它会自动管理连接池）
db.close()
print("Table myelsa_table created successfully!")1-4、用SQLAlchemy库
from sqlalchemy import create_engine, Column, Integer, String, MetaData
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 数据库连接配置（使用 SQLAlchemy 的 URI 格式）
DATABASE_URI = 'mysql+pymysql://root:[email protected]/test_database?charset=utf8mb4'
# 创建一个引擎实例
engine = create_engine(DATABASE_URI, echo=True) # echo=True 用于显示生成的 SQL 语句，调试时可以打开
# 创建基类
Base = declarative_base()
# 定义模型类
class MyElsaTable(Base):
__tablename__ = 'myelsa_table'
name = Column(String(255), nullable=False)
ID_Card = Column(String(255), primary_key=True) # 设置为主键
age = Column(Integer, nullable=False)
city = Column(String(255), nullable=False)
# 创建表（如果表不存在）
Base.metadata.create_all(engine)
# 如果你想要使用 ORM 来进行操作，可以创建一个 session 类
Session = sessionmaker(bind=engine)
session = Session()
# 这里不需要执行 SQL 语句或提交更改，因为 create_all 方法会自动处理
# 关闭 session（如果需要的话，但在这种情况下我们并没有进行任何 ORM 操作）
# session.close()
print("Table myelsa_table created successfully!")2、删除
在MySQL服务器上已有数据库test_database，且该数据库下已有数据表myelsa_table，现需删除数据表myelsa_table，以下为借助第三方库实现：
2-1、用mysql-connector-python库
import mysql.connector
# 数据库连接配置
config = {
'user': 'root', # 替换为你的MySQL用户名
'password': '123456', # 替换为你的MySQL密码
'host': '127.0.0.1', # 如果数据库在远程服务器上，请替换为相应的主机名或IP地址
'database': 'test_database', # 数据库名
}
try:
# 连接到数据库
cnx = mysql.connector.connect(**config)
cursor = cnx.cursor()
# 执行SQL语句来删除表
drop_table_query = "DROP TABLE IF EXISTS myelsa_table;"
cursor.execute(drop_table_query)
# 提交更改（在这个例子中，删除表不需要显式提交，因为不是事务的一部分）
# 但如果你之前做了其他更改，这里可以调用
cnx.commit()
print("Table myelsa_table deleted successfully!")
except mysql.connector.Error as err:
print(f"Error: '{err}'")
finally:
# 关闭游标和连接
if cursor:
cursor.close()
if cnx.is_connected():
cnx.close()2-2、用PyMySQL库
import pymysql
# 数据库连接配置
config = {
'user': 'root', # 替换为你的MySQL用户名
'password': '123456', # 替换为你的MySQL密码
'host': '127.0.0.1', # 如果数据库在远程服务器上，请替换为相应的主机名或IP地址
'database': 'test_database', # 数据库名
'charset': 'utf8mb4', # 设置字符集，可选
'cursorclass': pymysql.cursors.DictCursor # 使用字典游标，可选
}
try:
# 连接到数据库
connection = pymysql.connect(**config)
with connection.cursor() as cursor:
# 执行SQL语句来删除表
drop_table_query = "DROP TABLE IF EXISTS myelsa_table;"
cursor.execute(drop_table_query)
# 提交更改（在这个例子中，删除表不需要显式提交，因为不是事务的一部分）
# 但如果你之前做了其他更改，并且它们是在一个事务中，这里可以调用 connection.commit()
print("Table myelsa_table deleted successfully!")
except pymysql.MySQLError as err:
print(f"Error: '{err}'")
finally:
# 关闭连接
if connection.open:
connection.close()2-3、用PeeWee库
from peewee import *
# 数据库连接配置
db = MySQLDatabase('test_database', user='root', password='123456', host='127.0.0.1')
# 尝试连接到数据库
try:
db.connect()
# 定义一个模型，即使我们不使用它来进行数据操作，也需要它来引用表
class MyElsaTable(Model):
class Meta:
database = db
table_name = 'myelsa_table' # 如果表名与模型名不同，可以指定
# 执行SQL语句来删除表
# 注意：Peewee没有直接提供删除表的API，但我们可以通过执行原始SQL来实现
db.execute_sql("DROP TABLE IF EXISTS `myelsa_table`;")
print("Table myelsa_table deleted successfully!")
except Exception as e:
print(f"Error: {e}")
finally:
# 关闭数据库连接（Peewee会在析构时自动关闭连接，但显式关闭是个好习惯）
db.close()2-4、用SQLAlchemy库
from sqlalchemy import create_engine, MetaData, Table
# 数据库连接配置
config = {
'user': 'root', # 替换为你的MySQL用户名
'password': '123456', # 替换为你的MySQL密码
'host': '127.0.0.1', # 如果数据库在远程服务器上，请替换为相应的主机名或IP地址
'database': 'test_database', # 数据库名
'port': 3306, # 如果你的MySQL服务不在默认端口上，请指定端口
}
# 创建数据库引擎
engine = create_engine(f'mysql+pymysql://{config["user"]}:{config["password"]}'
f'@{config["host"]}:{config["port"]}/{config["database"]}')
# 连接到数据库（SQLAlchemy会自动处理连接）
# 定义元数据对象
metadata = MetaData()
# 定义要删除的表（不需要定义列，因为我们只是删除它）
myelsa_table = Table('myelsa_table', metadata, autoload_with=engine)
try:
# 删除表（如果存在）
myelsa_table.drop(engine)
print("Table myelsa_table deleted successfully!")
except Exception as e:
print(f"Error: {e}")
# SQLAlchemy会自动管理连接，你通常不需要显式关闭它
# 但如果你需要确保所有资源都被释放，可以调用dispose()方法
engine.dispose()