python dbUtil

本文主要是介绍python dbUtil，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

# -*- coding: utf-8 -*-    import pyodbc
import cx_Oracleclass Database:def __init__(self, driver, server, database, username, password):"""链接数据库:param driver:  数据库驱动:param server: 地址:端口:param database: 数据库名称:param username: 用户名:param password: 密码"""self.driver = driverself.database = databaseif self.database.upper() == '数据库':self.server = "地址:端口"self.username = "用户名"self.password = "密码"else:self.server = serverself.database = databaseself.username = usernameself.password = passwordself.connection = Noneself.cursor = Nonedef prepare(self, query):"""准备查询语句"""try:self.cursor.prepare(query)except Exception as e:print(f"准备查询语句时发生错误: {e}")def connect(self):"""连接数据库"""try:if self.driver == "ODBC Driver 17 for SQL Server":self.connection = pyodbc.connect(f"DRIVER={{{self.driver}}};SERVER={self.server};DATABASE={self.database};UID={self.username};PWD={self.password}")elif self.driver == "Oracle":self.connection = cx_Oracle.connect(f"{self.username}/{self.password}@{self.server}/{self.database}")self.cursor = self.connection.cursor()except Exception as e:print(f"连接数据库时出错: {e}")def execute_query(self, query):"""执行查询语句"""try:self.cursor.execute(query)return self.cursor.fetchall()except Exception as e:print(f"执行查询语句时出错: {e}")def execute_non_query(self, query):"""执行非查询语句"""try:self.cursor.prepare(query)self.cursor.execute(query)self.connection.commit()except Exception as e:print(f"执行非查询语句时出错: {e} | Error line: [{e.__traceback__.tb_lineno}]")def get_table_count(self, table_name, where_clause=None):"""获取表总数"""try:if where_clause:query = f"SELECT COUNT(*) FROM {table_name} where {where_clause}"else:query = f"SELECT COUNT(*) FROM {table_name}"self.cursor.execute(query)return self.cursor.fetchone()[0]except Exception as e:print(f"获取表总数时出错: {e}")def execute_batch_insert(self, table_name, data):"""执行批量插入语句"""try:query = f"INSERT INTO {table_name} VALUES "for row in data:query += f"({','.join([str(val) for val in row])}),"query = query[:-1]  # 去掉最后一个逗号self.cursor.execute(query)self.connection.commit()except Exception as e:print(f"执行批量插入语句时出错: {e}")def execute_many(self, query, params):"""执行多个参数的查询语句"""try:# print(query, params)self.cursor.executemany(query, params)self.connection.commit()except Exception as e:print(f"执行多个参数的sql语句时出错: {e}")# def call_stored_procedure(self, procedure_name, params):#     """#     调用存储过程#     """#     try:#         self.cursor.callproc(procedure_name, params)#         self.connection.commit()#     except Exception as e:#         print(f"调用存储过程时出错: {e}")def call_stored_procedure(self, procedure_name, params=None):"""调用存储过程"""try:if params:if self.driver == "ODBC Driver 17 for SQL Server":self.cursor.execute(f"EXEC {procedure_name} {','.join(['?' for _ in params])}", params)elif self.driver == "Oracle":self.cursor.callproc(procedure_name, params)else:if self.driver == "ODBC Driver 17 for SQL Server":self.cursor.execute(f"EXEC {procedure_name}")elif self.driver == "Oracle":self.cursor.callproc(procedure_name)self.connection.commit()except Exception as e:print(f"调用存储过程时出错: {e}")def __del__(self):"""关闭连接"""try:if self.cursor:self.cursor.close()if self.connection:self.connection.close()except Exception as e:self.cursor.close()self.connection.close()print(f"关闭连接时出错: {e}")def read_sql_file(file_path):"""读取sql文件D:\Desktop\待办sql.sql文件格式为文本文件，包含SQL查询语句，每条语句以分号结尾:param file_path::return:"""try:with open(file_path, 'r') as f:sql = f.read()# print(sql)return sqlexcept Exception as e:print(f"读取SQL文件时出错: {e}")

这篇关于python dbUtil的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---