python mvc架构的电影信息管理系统

""" 对pymysql的封装，提供更实用的功能 """ from pymysql import Connection class MySqlHelper: def __init__(self, database="", host="localhost", port=3306, user="root", password="123456"): """ 创建一个数据库连接 :param database:str类型，数据库名称 :param host:str类型，数据库地址 :param port:int类型，端口号 :param user:str类型，用户名 :param password:str类型，密码 """ self.conn = Connection( host=host, port=port, user=user, password=password, database=database, autocommit=True ) def fetch_one(self, sql, *params): """ 执行查询语句,返回结果中的第一行 :param sql:str类型，需要执行的SQL语句 :param params:SQL语句需要的参数 :return:tuple类型,第一行 """ try: with self.conn.cursor() as cursor: cursor.execute(sql, params) return cursor.fetchone() except Exception as e: print(f"发生错误: {e}") def fetch_all(self, sql, *params): """ 执行查询语句,返回所有结果 :param sql:str类型，需要执行的SQL语句 :param params:SQL语句需要的参数 :return:tuple类型,所有查询结果 """ try: with self.conn.cursor() as cursor: cursor.execute(sql, params) return cursor.fetchall() except Exception as e: print(f"发生错误: {e}") def fetch_yield(self, sql, *params): """ 执行查询语句,通过生成器返回所有结果 :param sql:str类型，需要执行的SQL语句 :param params:SQL语句需要的参数 :return:生成器类型,推算每行数据 """ try: with self.conn.cursor() as cursor: cursor.execute(sql, params) while True: data = cursor.fetchone() if data is None: break yield data except Exception as e: print(f"发生错误: {e}") def execute(self, sql, *params): """ 执行单条增删改语句 :param sql:str类型，需要执行的SQL语句 :param params:SQL语句需要的参数 :return:tuple类型,(受影响行数,插入数据自增长主键) """ try: with self.conn.cursor() as cursor: return cursor.execute(sql, params), cursor.lastrowid except Exception as e: print(f"发生错误: {e}") return () def execute_transaction(self, *sql_params): """ 执行多条SQL语句，同时成功或同时失败 :param sql_params:将多个SQL与参数定义在多个元组中 :return:int类型,成功执行的语句数 """ try: self.conn.begin() with self.conn.cursor() as cursor: count = 0 for sql, params in sql_params: count += cursor.execute(sql, params) self.conn.commit() return count except Exception as e: self.conn.rollback() print(f"发生错误: {e}") return -1 def execute_many(self, sql, *params): """ 多次执行同一条SQL语句，每次使用序列中的参数来替换SQL语句中的占位符 :param sql:str类型，需要多次执行的SQL语句 :param params:tuple类型，每个SQL语句需要的参数也要作为tuple类型 :return:int类型,成功执行的语句数 """ try: with self.conn.cursor() as cursor: count = cursor.executemany(sql, params) return count except Exception as e: print(f"发生错误: {e}") return -1 def execute_many_sql(self, *sql_params): """ 执行多条SQL语句，返回失败的SQL语句 :param sql:str类型，需要多次执行的SQL语句 :param params:tuple类型，SQL语句需要的参数 :return:int类型,成功执行的语句数 """ failed_sqls = [] with self.conn.cursor() as cursor: for sql, params in sql_params: try: cursor.execute(sql, params) except: failed_sqls.append(sql) return failed_sqls def close(self): """ 关闭数据库连接 """ self.conn.close() # 如果使用with语句，则需要实现__enter__和__exit__方法 def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): self.close()