Python单例模式封装pymysql，包括数据库创建、表创建和增删改查方法

在Python中，结合单例模式与pymysql进行数据库操作的封装，可以保证全局只有一个数据库连接实例，避免频繁创建和销毁连接带来的性能开销。以下是基于单例模式封装的DatabaseSingleton类，包含了数据库连接、数据库和表的创建以及基本的增删改查操作：

import pymysql
from typing import List, Tuple
from singleton_decorator import singleton

@singleton
class DatabaseSingleton:
    def __init__(self, host: str, user: str, password: str, default_db: str, charset='utf8mb4'):
        self.conn = None
        self.cursor = None
        self.host = host
        self.user = user
        self.password = password
        self.default_db = default_db
        self.charset = charset
        self._initialize()

    def _initialize(self):
        try:
            self.conn = pymysql.connect(
                host=self.host,
                user=self.user,
                password=self.password,
                charset=self.charset,
            )
            self.conn.select_db(self.default_db)
            self.cursor = self.conn.cursor()
        except pymysql.MySQLError as e:
            print(f"初始化数据库连接时发生错误: {e}")

    def create_database(self, db_name: str):
        sql = f"CREATE DATABASE IF NOT EXISTS `{db_name}` CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci;"
        try:
            self.cursor.execute(sql)
            self.conn.commit()
            print(f"数据库'{db_name}'创建成功")
        except pymysql.MySQLError as e:
            print(f"创建数据库时发生错误: {e}")

    def use_database(self, db_name: str):
        try:
            self.conn.select_db(db_name)
        except pymysql.ProgrammingError as e:
            print(f"切换至数据库'{db_name}'时发生错误: {e}")

    def create_table(self, table_schema: str):
        try:
            self.cursor.execute(table_schema)
            self.conn.commit()
            print("表创建成功")
        except pymysql.MySQLError as e:
            print(f"创建表时发生错误: {e}")

    def insert(self, table_name: str, values: Tuple, columns: Tuple[str, ...] = None):
        placeholders = ', '.join(['%s'] * len(values))
        if columns:
            columns_str = ', '.join(columns)
            sql = f"INSERT INTO `{table_name}` ({columns_str}) VALUES ({placeholders})"
        else:
            sql = f"INSERT INTO `{table_name}` VALUES ({placeholders})"
        try:
            self.cursor.execute(sql, values)
            self.conn.commit()
            return self.cursor.lastrowid
        except pymysql.MySQLError as e:
            print(f"插入数据时发生错误: {e}")

    def update(self, table_name: str, set_clause: str, where_clause: str, values: Tuple):
        sql = f"UPDATE `{table_name}` SET {set_clause} WHERE {where_clause}"
        try:
            self.cursor.execute(sql, values)
            self.conn.commit()
            return self.cursor.rowcount
        except pymysql.MySQLError as e:
            print(f"更新数据时发生错误: {e}")

    def delete(self, table_name: str, where_clause: str, values: Tuple):
        sql = f"DELETE FROM `{table_name}` WHERE {where_clause}"
        try:
            self.cursor.execute(sql, values)
            self.conn.commit()
            return self.cursor.rowcount
        except pymysql.MySQLError as e:
            print(f"删除数据时发生错误: {e}")

    def select(self, query: str, params: Tuple = None):
        try:
            self.cursor.execute(query, params)
            return self.cursor.fetchall()
        except pymysql.MySQLError as e:
            print(f"执行查询时发生错误: {e}")

# 使用示例：
ds = DatabaseSingleton('127.0.0.1', 'username', 'password', 'default_db')

# 创建数据库
ds.create_database('new_database')

# 切换到新数据库
ds.use_database('new_database')

# 创建表
table_schema = """
CREATE TABLE IF NOT EXISTS test_table (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(100),
    email VARCHAR(100),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
"""
ds.create_table(table_schema)

# 插入数据
ds.insert('test_table', ('John Doe', 'john.doe@example.com'))

# 更新数据
set_clause = "name = %s, email = %s"
where_clause = "id = %s"
values = ('Jane Doe', 'jane.doe@example.com', 1)
ds.update('test_table', set_clause, where_clause, values)

# 删除数据
where_clause = "id = %s"
values = (1,)
ds.delete('test_table', where_clause, values)

# 查询数据
results = ds.select("SELECT * FROM test_table")
for row in results:
    print(row)

注意：这里的@singleton装饰器需要先导入一个支持单例模式的装饰器模块，如singleton_decorator，如果你没有这样的模块，可以通过自定义单例模式类实现。上述示例假设了存在这样一个装饰器，实际上你可能需要自行实现或安装相应的库。

自定义单例实现

import pymysql

class Singleton(type):
    _instances = {}

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
        return cls._instances[cls]

class DatabaseSingleton(metaclass=Singleton):
    def __init__(self, host: str, user: str, password: str, default_db: str, charset='utf8mb4'):
        self.conn = None
        self.cursor = None
        self.host = host
        self.user = user
        self.password = password
        self.default_db = default_db
        self.charset = charset
        self._initialize()

    def _initialize(self):
        try:
            self.conn = pymysql.connect(
                host=self.host,
                user=self.user,
                password=self.password,
                charset=self.charset,
            )
            self.conn.select_db(self.default_db)
            self.cursor = self.conn.cursor()
        except pymysql.MySQLError as e:
            print(f"初始化数据库连接时发生错误: {e}")

    # ... 其余方法与上一回答相同 ...

此外，为了简化问题，上述示例中并未处理所有的异常情况，在实际应用中，你应该更加细致地处理可能出现的异常，确保资源的正确关闭和释放。