Python筑基之旅-MySQL數據庫(三)

一、數據庫操作

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

二、數據表操作