在 Python 中实现数据库操作的三层架构一般包括以下三层:
数据访问层(Data Access Layer):负责与数据库进行交互,执行 SQL 查询和更新操作。
业务逻辑层(Business Logic Layer):处理业务逻辑,对数据进行处理和加工,调用数据访问层提供的接口完成数据的增删改查操作。
表现层(Presentation Layer):与用户交互,接收用户输入并展示数据,将用户请求传递给业务逻辑层进行处理,并将处理结果展示给用户。
下面是一个简单的示例,展示如何使用三层架构实现 Python 中的增删改查操作:
# 数据访问层 class Database:def __init__(self, connection):self.connection = connectiondef execute_query(self, query, params=None):cursor = self.connection.cursor()if params:cursor.execute(query, params)else:cursor.execute(query)return cursor.fetchall()def execute_update(self, query, params=None):cursor = self.connection.cursor()if params:cursor.execute(query, params)else:cursor.execute(query)self.connection.commit()return cursor.rowcount# 业务逻辑层 class DataAccessLayer:def __init__(self, database):self.db = databasedef create_record(self, data):query = "INSERT INTO table_name (column1, column2, ...) VALUES (%s, %s, ...)"return self.db.execute_update(query, data)def update_record(self, id, data):query = "UPDATE table_name SET column1 = %s, column2 = %s WHERE id = %s"params = (data['column1'], data['column2'], id)return self.db.execute_update(query, params)def delete_record(self, id):query = "DELETE FROM table_name WHERE id = %s"return self.db.execute_update(query, (id,))def get_record(self, id):query = "SELECT * FROM table_name WHERE id = %s"result = self.db.execute_query(query, (id,))return result[0] if result else None# 表现层 def main():# 建立数据库连接connection = create_database_connection()# 初始化数据访问层db = Database(connection)dal = DataAccessLayer(db)# 业务逻辑操作示例# 创建记录data_to_insert = {'column1': value1, 'column2': value2, ...}inserted_rows = dal.create_record(data_to_insert)# 更新记录record_id_to_update = 1data_to_update = {'column1': new_value1, 'column2': new_value2, ...}updated_rows = dal.update_record(record_id_to_update, data_to_update)# 删除记录record_id_to_delete = 1deleted_rows = dal.delete_record(record_id_to_delete)# 查询记录record_id_to_get = 1record = dal.get_record(record_id_to_get)if record:print("Record found:", record)else:print("Record not found.")# 关闭数据库连接connection.close()if __name__ == "__main__":main()