Python检查MySQL数据库表存在并存储/删除库表

需要实现的功能

• 新建数据库之前，检查是否已存在
• 新建数据表之前，检查是否已存在
• 测试用的数据库及数据表，在测试完后删除

---

以下为具体实现

判断数据库存在并创建

使用SqlAlchemy连接MySQL数据库的步骤：
1、创建连接用的url
2、判断该url是否为一个已存在的数据库。若不存在，则应首先创建该数据库。
3、新建连接engine
4、连接数据库

创建连接用的 url

db_url = ‘mysql+pymysql://username:password@localhost:port/db_name

检查数据库是否存在

需要从`sqlalchemy_utils`库中导入`database_exists`和`create_database`
from sqlalchemy_utils import database_exists, create_database`
# 检查数据库是否存在，并创建
if not database_exists(db_url):
  create_database(db_url, encoding='utf8')

# 新建连接engine
engine = create_engine(db_url)

# 连接数据库
conn = engine.connect()

判断数据表存在并创建

通过engine的table_names()方法返回当前数据库里所有表名称，从而判断某个数据表table_name是否存在
由于是通过pandas来存储和读取数据表，因此即便表不存在，可以直接用pandas.to_sql保存，不必提前创建表结构，省去了写sql语句的麻烦。

if table_name not in engine.table_names():
  df.to_sql(table_name, con=conn, index=False) # 如果表不存在，则直接保存

若表存在，则最好先检查待存DataFrame的列标签是否与当前表的表头一致。此处需要用到sqlalchemy.inspect，具体为：

from sqlalchemy import inspect

inspector = inspect(engine)
# 返回某个数据表里的列名
columns_dict_list = inspector(table_name) # 返回的是一个字典的列表，列表中每个元素的’name’键对应该数据表的一个列名
columns = [item[‘name’] for item in columns_dict_list]

比较两组数据表的列名，无误后即可继续使用pandas.to_sql向已有的数据表添加数据，其中参数if_exists的值为append：

if df_columns == columns: # df_columns是待存储DataFrame的列名列表
  df.to_sql(table_name, con=conn, if_exists=‘append’) 

删除测试时添加的数据库表

笔者目前没有在sqlalchemy中找到删除数据库表的方法，所以只能使用sqlalchemy_utils库的drop_database来删除数据库，使用pymysql库及sql语句来删除数据表。

删除数据表

import pymysql
# 使用pymysql建立与数据库的连接：
pymysql_conn = pymysql.connect(‘localhost, username, password, db_name, charset=‘utf8’)
# 删除数据表的sql语句：
drop_table_sql = ‘drop table %s’ %table_name
cursor = pymysql_conn.cursor()
cursor.execute(drop_table_sql)
pymysql_conn.close()

删除数据库

from sqlalchemy_utils import drop_database
drop_database(db_url)  # db_url是通过sqlalchemy连接数据库所使用的url