pymysql sqlite 数据库封装类
import pymysql import sqlite3 class Database: def __init__(self, dbtype, dbname): """ 初始化数据库连接 :param dbtype: 数据库类型，可以是'mysql'或'sqlite' :param dbname: 数据库名称 """ self.dbtype = dbtype self.dbname = dbname self.conn = None self.cursor = None if dbtype == 'mysql': self.conn = pymysql.connect(host='localhost', user='root', password='', db=dbname, charset='utf8mb4') self.cursor = self.conn.cursor() elif dbtype == 'sqlite': self.conn = sqlite3.connect(dbname) self.cursor = self.conn.cursor() def execute(self, query, params=None): """ 执行SQL查询 :param query: SQL查询语句 :param params: 查询参数，可以是一个元组或字典 :return: 查询结果 """ if self.dbtype == 'mysql': self.cursor.execute(query, params) self.conn.commit() return self.cursor.fetchall() elif self.dbtype == 'sqlite': if params is None: self.cursor.execute(query) else: self.cursor.execute(query, params) self.conn.commit() return self.cursor.fetchall() def executemany(self, query, params): """ 执行多个SQL查询 :param query: SQL查询语句 :param params: 查询参数，必须是一个列表，其中每个元素都是一个元组或字典 :return: 查询结果 """ if self.dbtype == 'mysql': self.cursor.executemany(query, params) self.conn.commit() return self.cursor.fetchall() elif self.dbtype == 'sqlite': self.cursor.executemany(query, params) self.conn.commit() return self.cursor.fetchall() def close(self): """ 关闭数据库连接 """ self.cursor.close() self.conn.close()
使用示例：
# 初始化MySQL数据库连接
db = Database('mysql', 'testdb')# 执行SQL查询
result = db.execute('SELECT * FROM users')print(result)
# 执行多个SQL查询
params = [('John', 'Doe'), ('Jane', 'Doe')]
result = db.executemany('INSERT INTO users (first_name, last_name) VALUES (%s, %s)', params)
print(result)# 关闭数据库连接
db.close()
# 初始化SQLite数据库连接
db = Database('sqlite', 'testdb.db')# 执行SQL查询
result = db.execute('SELECT * FROM users')
print(result)# 执行多个SQL查询
params = [('John', 'Doe'), ('Jane', 'Doe')]
result = db.executemany('INSERT INTO users (first_name, last_name) VALUES (?, ?)', params)
print(result)# 关闭数据库连接
db.close()
该封装类包含以下方法：
该封装类包含以下方法：
__init__(self, dbtype, dbname)：初始化数据库连接。接受两个参数，分别是数据库类型和数据库名称。根据数据库类型，使用相应的库连接到数据库，并创建游标对象。
execute(self, query, params=None)：执行SQL查询。接受两个参数，分别是SQL查询语句和查询参数。如果查询参数是一个元组或字典，则使用游标对象的execute()方法执行查询，并返回查询结果。如果查询参数是None，则直接执行查询语句。
executemany(self, query, params)：执行多个SQL查询。接受两个参数，分别是SQL查询语句和查询参数。查询参数必须是一个列表，其中每个元素都是一个元组或字典。使用游标对象的executemany()方法执行查询，并返回查询结果。
close(self)：关闭数据库连接。使用游标对象的close()方法和连接对象的close()方法关闭数据库连接。
使用示例：
初始化MySQL数据库连接：
db = Database('mysql', 'testdb')执行SQL查询：
result = db.execute('SELECT * FROM users')执行多个SQL查询：
params = [('John', 'Doe'), ('Jane', 'Doe')]
result = db.executemany('INSERT INTO users (first_name, last_name) VALUES (%s, %s)', params)关闭数据库连接：
db.close()
初始化SQLite数据库连接：
db = Database('sqlite', 'testdb.db')执行SQL查询：
result = db.execute('SELECT * FROM users')执行多个SQL查询：
params = [('John', 'Doe'), ('Jane', 'Doe')]
result = db.executemany('INSERT INTO users (first_name, last_name) VALUES (?, ?)', params)关闭数据库连接：
db.close()