数据库查询语句的封装思路

import yaml def yamlread(path): # 打开并读取YAML文件 with open(path, 'r', encoding='utf-8') as file: config = yaml.safe_load(file) return configc

创建一个文件操作方法读取文件信息

class dboperations: def __init__(self, config_path=r'D:\PycharmProjects\PythonProject\config\db.yaml'): self.config = fileoperation.yamlread(config_path) self.connect = None self._auto_commit = True @property def get_connect(self): """获取连接""" dbconfig = self.config['database'] # 修正：正确获取配置参数 self.connect = pymysql.connect( host=dbconfig['host'], user=dbconfig['user'], password=dbconfig['password'], database=dbconfig['database'], charset='utf8mb4', autocommit=self._auto_commit ) return self.connect @contextmanager def get_cursor(self, cursor_type=pymysql.cursors.DictCursor): """上下文管理器，自动处理游标和事务""" conn = self.get_connect cursor = conn.cursor(cursor_type) try: yield cursor if self._auto_commit: conn.commit() except Exception as e: conn.rollback() raise e finally: cursor.close() def query_fields_advanced(self, table_name,fields_config,conditions=None): try: with self.get_cursor() as cursor: sql = f"SELECT {fields_config} FROM {table_name}" if conditions: sql += f" WHERE {conditions}" cursor.execute(sql) results = cursor.fetchall() return results except Exception as e: print(f"查询失败: {e}") return None def close(self): """关闭连接""" if self.connect: self.connect.close() self.connect = None

创建数据库操作类封装操作动作

def debg_quarydata(): # 创建数据库操作对象 db = dboperations() table_name = db.config['testmessage']['tablename'] fieldsconfigs = db.config['testmessage']['fields_config'] #将列表转换为数据模式， [a,b]->a,b select_fields = ','.join(fieldsconfigs) print(select_fields,table_name) results = db.query_fields_advanced(table_name,select_fields) print(results) # 关闭连接 db.close()

结果类中进行调用