使用python操作数据库

一、问题背景

在日常开发过程中，随着项目进展和业务功能的迭代，我们需要对数据库的表结构进行修改，向部分表中追加字段，并对追加后的字段进行数据填充。但是如果需要追加字段的表比较多，并且追加字段后，还可能需要对数据库的脚本进行维护，此时手动操作就过于耗时，所以借助python来操作数据库，进行统一修改。

二、安装python

Python3 环境搭建 | 菜鸟教程

三、代码示例

在使用之前需要先安装mysql.connector包，命令如下：

pip3 install mysql.connector

python代码如下：

import mysql.connector
from mysql.connector import errorcode# 数据库连接配置
config = {'host': '你的地址','port': '你的端口','user': '你的用户','password': '你的密码','database': '你的数据库'
}# 要追加的表和字段，允许指定多个表和多个字段，格式为 {'表名': [('字段名', '注释'), ...]}
tables_to_check = {'sys_user': [('org_code', '组织机构代码')],
}def connect_to_database(config):"""连接数据库:param config: 数据库连接配置:return: 数据库连接"""try:cnx = mysql.connector.connect(**config)# 启用事务cnx.autocommit = Falsereturn cnxexcept mysql.connector.Error as err:if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:print("用户名或密码错误")elif err.errno == errorcode.ER_BAD_DB_ERROR:print("指定的数据库不存在")else:print(err)return Nonedef check_and_add_fields(cursor, table, fields):"""检查并添加字段:param cursor: 游标:param table: 表:param fields: 字段列表:return: alter 语句"""cursor.execute(f"DESCRIBE {table}")existing_fields = {row[0] for row in cursor.fetchall()}alter_statements = []for field, comment in fields:if field not in existing_fields:alter_statement = f"ALTER TABLE {table} ADD COLUMN {field} VARCHAR(50) COMMENT '{comment}';"alter_statements.append(alter_statement)cursor.execute(alter_statement)return alter_statementsdef update_org_code(cursor, table):"""更新 org_code 字段（此方法用于对追加的字段填充数据，可根据具体需求进行改造）:param cursor: 游标:param table: 表:return: 空"""update_statement = f"""UPDATE {table} tJOIN sys_dept d ON t.dept_id = d.dept_idSET t.org_code = d.org_code;"""cursor.execute(update_statement)def print_summary(alter_statements_summary, tables_updated, no_alter_needed):"""输出更新和未更新表的信息:param alter_statements_summary: 修改语句列表:param tables_updated: 更新了的表:param no_alter_needed: 无需修改的表列表:return: 空"""print("\n追加字段修改语句：")for table, alter_statements in alter_statements_summary:print(f"{table} 修改语句：")for stmt in alter_statements:print(stmt)print("\n更新追加字段的表：")for table in tables_updated:print(table)print("\n不需要追加字段的表：")for table in no_alter_needed:print(table)def main():"""主函数:return: 空"""print("连接数据库...")cnx = connect_to_database(config)if not cnx:returnprint("开始修改...")try:cursor = cnx.cursor()alter_statements_summary = []tables_updated = []no_alter_needed = []for table, fields in tables_to_check.items():alter_statements = check_and_add_fields(cursor, table, fields)if alter_statements:alter_statements_summary.append((table, alter_statements))tables_updated.append(table)update_org_code(cursor, table)else:no_alter_needed.append(table)# 提交事务cnx.commit()print("所有修改提交成功")print_summary(alter_statements_summary, tables_updated, no_alter_needed)except mysql.connector.Error as err:# 回滚事务cnx.rollback()print(f"发生异常，错误信息: {err}")print("所有修改全部回滚")finally:cursor.close()cnx.close()if __name__ == "__main__":main()

执行结果：

连接数据库...
开始修改...
所有修改提交成功追加字段修改语句：
sys_user 修改语句：
ALTER TABLE sys_user ADD COLUMN org_code VARCHAR(50) COMMENT '组织机构代码';更新追加字段的表：
sys_user不需要追加字段的表：

四、总结

以上python代码处理的事情，只是我们日常开发中遇到问题的一个缩影，大家可以根据自己的具体需求，进行修改。最重要的是要清楚自己想解决的问题是什么，如何借助工具更好地去解决问题，从而提升自己的工作效率。