Python判断MySQL表是否存在的两种方法

pip install mysql-connector-python

import mysql.connector

def check_and_create_table():
    # 数据库连接配置
    config = {
        'user': 'your_username',
        'password': 'your_password',
        'host': 'localhost',
        'database': 'your_database',
        'raise_on_warnings': True
    }
    
    try:
        # 建立数据库连接
        conn = mysql.connector.connect(**config)
        cursor = conn.cursor()
        
        # 表名
        table_name = 'your_table'
        
        # 检查表是否存在
        cursor.execute(f"SHOW TABLES LIKE '{table_name}'")
        result = cursor.fetchone()
        
        if not result:
            print(f"表 {table_name} 不存在，正在创建...")
            # 创建表的SQL语句
            create_table_sql = """
            CREATE TABLE your_table (
                id INT AUTO_INCREMENT PRIMARY KEY,
                name VARCHAR(255) NOT NULL,
                age INT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
            """
            cursor.execute(create_table_sql)
            print(f"表 {table_name} 创建成功")
        else:
            print(f"表 {table_name} 已存在")
            
    except mysql.connector.Error as err:
        print(f"数据库错误: {err}")
    finally:
        if 'conn' in locals() and conn.is_connected():
            cursor.close()
            conn.close()

# 调用函数
check_and_create_table()

import mysql.connector

def check_and_create_table_v2():
    config = {
        'user': 'your_username',
        'password': 'your_password',
        'host': 'localhost',
        'database': 'your_database',
        'raise_on_warnings': True
    }
    
    try:
        conn = mysql.connector.connect(**config)
        cursor = conn.cursor()
        
        table_name = 'your_table'
        
        # 使用information_schema查询表是否存在
        query = """
        SELECT COUNT(*)
        FROM information_schema.tables
        WHERE table_schema = %s
        AND table_name = %s
        """
        cursor.execute(query, (config['database'], table_name))
        count = cursor.fetchone()[0]
        
        if count == 0:
            print(f"表 {table_name} 不存在，正在创建...")
            create_table_sql = """
            CREATE TABLE your_table (
                id INT AUTO_INCREMENT PRIMARY KEY,
                name VARCHAR(255) NOT NULL,
                age INT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
            """
            cursor.execute(create_table_sql)
            print(f"表 {table_name} 创建成功")
        else:
            print(f"表 {table_name} 已存在")
            
    except mysql.connector.Error as err:
        print(f"数据库错误: {err}")
    finally:
        if 'conn' in locals() and conn.is_connected():
            cursor.close()
            conn.close()

check_and_create_table_v2()

import mysql.connector
from mysql.connector import Error

class MySQLTableManager:
    def __init__(self, host, user, password, database):
        self.host = host
        self.user = user
        self.password = password
        self.database = database
        self.connection = None
        
    def connect(self):
        try:
            self.connection = mysql.connector.connect(
                host=self.host,
                user=self.user,
                password=self.password,
                database=self.database
            )
            return True
        except Error as e:
            print(f"连接数据库失败: {e}")
            return False
            
    def disconnect(self):
        if self.connection and self.connection.is_connected():
            self.connection.close()
            
    def table_exists(self, table_name):
        if not self.connection or not self.connection.is_connected():
            if not self.connect():
                return False
                
        try:
            cursor = self.connection.cursor()
            query = """
            SELECT COUNT(*)
            FROM information_schema.tables
            WHERE table_schema = %s
            AND table_name = %s
            """
            cursor.execute(query, (self.database, table_name))
            return cursor.fetchone()[0] > 0
        except Error as e:
            print(f"查询表存在性失败: {e}")
            return False
        finally:
            if 'cursor' in locals():
                cursor.close()
                
    def create_table(self, table_name, create_sql):
        if not self.connection or not self.connection.is_connected():
            if not self.connect():
                return False
                
        try:
            cursor = self.connection.cursor()
            cursor.execute(create_sql)
            self.connection.commit()
            print(f"表 {table_name} 创建成功")
            return True
        except Error as e:
            print(f"创建表失败: {e}")
            self.connection.rollback()
            return False
        finally:
            if 'cursor' in locals():
                cursor.close()
                
    def ensure_table_exists(self, table_name, create_sql):
        if not self.table_exists(table_name):
            print(f"表 {table_name} 不存在，正在创建...")
            return self.create_table(table_name, create_sql)
        else:
            print(f"表 {table_name} 已存在")
            return True

# 使用示例
if __name__ == "__main__":
    manager = MySQLTableManager(
        host='localhost',
        user='your_username',
        password='your_password',
        database='your_database'
    )
    
    table_name = 'employees'
    create_table_sql = """
    CREATE TABLE employees (
        id INT AUTO_INCREMENT PRIMARY KEY,
        name VARCHAR(100) NOT NULL,
        position VARCHAR(100),
        salary DECIMAL(10,2),
        hire_date DATE,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
    """
    
    manager.ensure_table_exists(table_name, create_table_sql)
    manager.disconnect()