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

Python判断MySQL表是否存在，不存在则创建

news 2026/3/27 1:34:27

引言

在数据库开发中，经常需要检查某个表是否存在，如果不存在则创建它。这在初始化数据库结构或部署新应用时特别有用。本文将介绍如何使用Python连接MySQL数据库，并实现表存在性检查与创建的功能。

准备工作

首先确保你已经安装了必要的Python库：

mysql-connector-python或PyMySQL（本文以mysql-connector为例）

可以通过pip安装：

pipinstallmysql-connector-python

基本实现方法

方法一：使用`SHOW TABLES`查询

importmysql.connectordefcheck_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()ifnotresult: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}已存在")exceptmysql.connector.Erroraserr:print(f"数据库错误:{err}")finally:if'conn'inlocals()andconn.is_connected():cursor.close()conn.close()# 调用函数check_and_create_table()

方法二：查询information_schema（更推荐）

importmysql.connectordefcheck_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]ifcount==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}已存在")exceptmysql.connector.Erroraserr:print(f"数据库错误:{err}")finally:if'conn'inlocals()andconn.is_connected():cursor.close()conn.close()check_and_create_table_v2()

方法比较

SHOW TABLES方法：
- 简单直观
- 但表名匹配是模糊的（LIKE操作符）
- 在某些MySQL版本中可能有大小写敏感问题
information_schema方法：
- 更标准、更可靠
- 使用参数化查询，防止SQL注入
- 明确指定数据库名和表名
- 推荐在生产环境中使用

完整封装类

下面是一个更完整的封装类，可以重复使用：

importmysql.connectorfrommysql.connectorimportErrorclassMySQLTableManager:def__init__(self,host,user,password,database):self.host=host self.user=user self.password=password self.database=database self.connection=Nonedefconnect(self):try:self.connection=mysql.connector.connect(host=self.host,user=self.user,password=self.password,database=self.database)returnTrueexceptErrorase:print(f"连接数据库失败:{e}")returnFalsedefdisconnect(self):ifself.connectionandself.connection.is_connected():self.connection.close()deftable_exists(self,table_name):ifnotself.connectionornotself.connection.is_connected():ifnotself.connect():returnFalsetry: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))returncursor.fetchone()[0]>0exceptErrorase:print(f"查询表存在性失败:{e}")returnFalsefinally:if'cursor'inlocals():cursor.close()defcreate_table(self,table_name,create_sql):ifnotself.connectionornotself.connection.is_connected():ifnotself.connect():returnFalsetry:cursor=self.connection.cursor()cursor.execute(create_sql)self.connection.commit()print(f"表{table_name}创建成功")returnTrueexceptErrorase:print(f"创建表失败:{e}")self.connection.rollback()returnFalsefinally:if'cursor'inlocals():cursor.close()defensure_table_exists(self,table_name,create_sql):ifnotself.table_exists(table_name):print(f"表{table_name}不存在，正在创建...")returnself.create_table(table_name,create_sql)else:print(f"表{table_name}已存在")returnTrue# 使用示例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()