django的數據庫原生操作sql

from django.db import connection
from django.db import transaction
from django.db.utils import (IntegrityError,OperationalError,ProgrammingError,DataError
)
from django.utils import timezoneclass Db(object):"""數據庫操作工具類，封裝了CRUD、批量操作、事務和軟刪除功能數據表必須要有：updated_at-更新時間created_at-創建時間is_deleted-軟刪除標志 0為刪除，1已刪除deleted_at-刪除時間"""# ------------------------------# 核心查詢方法# ------------------------------def execute_query(self, sql, params=None, fetchone=False, return_last_id=False):"""執行SQL查詢:param sql: SQL語句:param params: 查詢參數（元組）:param fetchone: 是否只返回第一條記錄（SELECT專用）:param return_last_id: 是否返回最后插入的ID（INSERT/UPDATE/DELETE專用）:return: SELECT返回字典列表/單條字典，其他語句返回受影響行數或最后插入ID"""try:with connection.cursor() as cursor:cursor.execute(sql, params or ())# 打印sql# print(f"{sql}====\n===={params}")if sql.strip().lower().startswith(('select', 'show')):columns = [col[0] for col in cursor.description]results = [dict(zip(columns, row)) for row in cursor.fetchall()]return results[0] if fetchone and results else resultselse:if return_last_id:try:return cursor.lastrowidexcept AttributeError:# 如果數據庫后端不支持lastrowid，則返回受影響行數return cursor.rowcountelse:return cursor.rowcountexcept IntegrityError as e:error_msg = f"數據完整性錯誤（可能原因：主鍵沖突、唯一約束重復）：{str(e)}"print(error_msg)raise ValueError(error_msg) from eexcept OperationalError as e:error_msg = f"數據庫連接/操作失敗（可能原因：數據庫未啟動、網絡中斷）：{str(e)}"print(error_msg)raise ConnectionError(error_msg) from eexcept ProgrammingError as e:error_msg = f"SQL語法/表結構錯誤（可能原因：表/字段不存在、SQL拼寫錯誤）：{str(e)}"print(error_msg)raise SyntaxError(error_msg) from eexcept DataError as e:error_msg = f"數據格式錯誤（可能原因：類型不匹配、長度超限）：{str(e)}"print(error_msg)raise TypeError(error_msg) from eexcept Exception as e:error_msg = f"意外錯誤：{str(e)}"print(error_msg)raise# ------------------------------# 插入操作# ------------------------------def insert_data(self, table, data, auto_time=True, return_last_id=True):"""插入單條數據:param table: 表名字符串:param data: 數據字典，格式 {字段名: 值}:param auto_time: 是否自動添加時間字段（默認True）:param return_last_id: 是否返回最后插入的ID:return: 最后插入的ID或受影響的行數"""fields = list(data.keys())values = list(data.values())if auto_time:now = timezone.now()fields.extend(['created_at', 'updated_at'])values.extend([now, now])placeholders = ', '.join(['%s'] * len(fields))sql = f"INSERT INTO {table} ({', '.join(fields)}) VALUES ({placeholders})"return self.execute_query(sql, values, return_last_id=return_last_id)def batch_insert(self, table, data_list, auto_time=True, batch_size=1000):"""批量插入數據（自動分批次）:param table: 表名字符串:param data_list: 數據字典列表，格式 [{字段名: 值}, ...]:param auto_time: 是否自動添加時間字段（默認True）:param batch_size: 每批次處理的記錄數（默認1000）:return: 總共插入的行數"""if not data_list:return 0total_rows = 0total_batches = (len(data_list) + batch_size - 1) // batch_sizefor i in range(total_batches):start = i * batch_sizeend = start + batch_sizebatch_data = data_list[start:end]first_item = batch_data[0].copy()fields = list(first_item.keys())if auto_time:now = timezone.now()fields.extend(['created_at', 'updated_at'])for item in batch_data:item['created_at'] = nowitem['updated_at'] = nowplaceholders = ', '.join(['%s'] * len(fields))value_groups = [tuple(item.values()) for item in batch_data]flat_values = [val for group in value_groups for val in group]sql = f"INSERT INTO {table} ({', '.join(fields)}) VALUES "sql += ', '.join([f"({placeholders})"] * len(batch_data))rows = self.execute_query(sql, flat_values)total_rows += rowsreturn total_rows# ------------------------------# 更新操作# ------------------------------def update_data(self, table, data, where, params=None, auto_time=True, return_last_id=False):"""更新數據:param table: 表名字符串:param data: 要更新的數據字典，格式 {字段名: 值}:param where: WHERE條件（不帶WHERE關鍵字），例如 "id = %s":param params: WHERE條件的參數列表:param auto_time: 是否自動更新updated_at字段（默認True）:param return_last_id: 是否返回最后插入的ID（在支持的數據庫中可能返回更新的行ID）:return: 受影響的行數或最后插入ID"""set_items = []values = list(data.values())if auto_time:data['updated_at'] = timezone.now()set_items = [f"{field} = %s" for field in data.keys()]values = list(data.values())else:set_items = [f"{field} = %s" for field in data.keys()]set_clause = ', '.join(set_items)sql = f"UPDATE {table} SET {set_clause} WHERE {where}"if params:values.extend(params)return self.execute_query(sql, values, return_last_id=return_last_id)def batch_update(self, table, data_list, where_field='id', auto_time=True, batch_size=1000):"""批量更新數據（使用CASE WHEN優化）:param table: 表名字符串:param data_list: 數據字典列表，每個字典必須包含where_field字段:param where_field: 用于匹配記錄的字段（默認id）:param auto_time: 是否自動更新updated_at字段（默認True）:param batch_size: 每批次處理的記錄數（默認1000）:return: 總共更新的行數"""if not data_list:return 0total_rows = 0total_batches = (len(data_list) + batch_size - 1) // batch_sizefor i in range(total_batches):start = i * batch_sizeend = start + batch_sizebatch_data = data_list[start:end]case_clauses = []values = []update_fields = {k for item in batch_data for k in item.keys() if k != where_field}for field in update_fields:case_sql = f"{field} = CASE {where_field} "for item in batch_data:case_sql += f"WHEN %s THEN %s "values.extend([item[where_field], item[field]])case_sql += "END"case_clauses.append(case_sql)where_ids = [item[where_field] for item in batch_data]values.extend(where_ids)if auto_time:now = timezone.now()case_clauses.append(f"updated_at = %s")values.append(now)set_clause = ', '.join(case_clauses)where_placeholders = ', '.join(['%s'] * len(where_ids))sql = f"UPDATE {table} SET {set_clause} WHERE {where_field} IN ({where_placeholders})"rows = self.execute_query(sql, values)total_rows += rowsreturn total_rows# ------------------------------# 刪除/軟刪除操作# ------------------------------def soft_delete(self, table, where, params=None, return_last_id=False):"""軟刪除數據（標記is_deleted=1）:param table: 表名字符串:param where: WHERE條件（不帶WHERE關鍵字），例如 "id = %s":param params: WHERE條件的參數列表:param return_last_id: 是否返回最后插入的ID（在支持的數據庫中可能返回刪除的行ID）:return: 受影響的行數或最后插入ID"""data = {'is_deleted': 1, 'deleted_at': timezone.now()}return self.update_data(table, data, where, params, auto_time=False, return_last_id=return_last_id)def delete_data(self, table, where, params=None, hard_delete=False, return_last_id=False):"""刪除數據（默認軟刪除）:param table: 表名字符串:param where: WHERE條件（不帶WHERE關鍵字），例如 "id = %s":param params: WHERE條件的參數列表:param hard_delete: 是否執行硬刪除（物理刪除）:param return_last_id: 是否返回最后插入的ID（在支持的數據庫中可能返回刪除的行ID）:return: 受影響的行數或最后插入ID"""if hard_delete:sql = f"DELETE FROM {table} WHERE {where}"return self.execute_query(sql, params, return_last_id=return_last_id)else:return self.soft_delete(table, where, params, return_last_id=return_last_id)# ------------------------------# 查詢操作# ------------------------------def get_list(self, table, where=None, params=None, order_by=None, limit=None, offset=None,with_deleted=False, only_deleted=False, fields=None):"""查詢列表數據:param table: 表名字符串:param where: WHERE條件（不帶WHERE關鍵字）:param params: WHERE條件的參數列表:param order_by: 排序字段，例如 "created_at DESC":param limit: 返回記錄數限制:param offset: 偏移量（用于分頁）:param with_deleted: 是否包含已刪除數據:param only_deleted: 是否只返回已刪除數據:param fields: 要返回的字段列表，默認返回所有字段:return: 符合條件的記錄列表"""if fields:select_str = ', '.join(fields)else:select_str = '*'conditions = []query_params = []if where:conditions.append(where)if params:query_params.extend(params)if not with_deleted:if only_deleted:conditions.append("is_deleted = 1")else:conditions.append("is_deleted = 0")order_str = f"ORDER BY {order_by}" if order_by else ""limit_str = f"LIMIT {limit}" if limit is not None else ""offset_str = f"OFFSET {offset}" if offset is not None else ""where_clause = " AND ".join(conditions) if conditions else ""where_str = f"WHERE {where_clause}" if where_clause else ""sql = f"SELECT {select_str} FROM {table} {where_str} {order_str} {limit_str} {offset_str}"return self.execute_query(sql, query_params)def get_one(self, table, where, params=None, with_deleted=False, only_deleted=False, fields=None):"""查詢單條數據:param table: 表名字符串:param where: WHERE條件（不帶WHERE關鍵字）:param params: WHERE條件的參數列表:param with_deleted: 是否包含已刪除數據:param only_deleted: 是否只返回已刪除數據:param fields: 要返回的字段列表，默認返回所有字段:return: 符合條件的單條記錄或None"""return self.get_list(table, where, params,with_deleted=with_deleted,only_deleted=only_deleted,fields=fields,limit=1)def get_count(self, table, where=None, params=None, with_deleted=False, only_deleted=False):"""查詢記錄總數:param table: 表名字符串:param where: WHERE條件（不帶WHERE關鍵字）:param params: WHERE條件的參數列表:param with_deleted: 是否包含已刪除數據:param only_deleted: 是否只返回已刪除數據:return: 記錄總數"""conditions = []query_params = []if where:conditions.append(where)if params:query_params.extend(params)if not with_deleted:if only_deleted:conditions.append("is_deleted = 1")else:conditions.append("is_deleted = 0")where_clause = " AND ".join(conditions) if conditions else ""where_str = f"WHERE {where_clause}" if where_clause else ""sql = f"SELECT COUNT(*) AS count FROM {table} {where_str}"result = self.execute_query(sql, query_params, fetchone=True)return result['count'] if result else 0# ------------------------------# 事務操作# ------------------------------def execute_transaction(self, operations):"""執行一組SQL操作作為事務:param operations: SQL操作列表，格式 [(SQL語句, 參數元組), ...]:return: 每個操作的返回結果列表:raises: 任何操作失敗時拋出異常，所有操作回滾"""with transaction.atomic():results = []for idx, (sql, params) in enumerate(operations):try:result = self.execute_query(sql, params)results.append(result)except Exception as e:raise RuntimeError(f"事務中第{idx + 1}條SQL執行失敗：{str(e)}") from ereturn results# ------------------------------# 數據恢復# ------------------------------def restore_data(self, table, where, params=None):"""恢復軟刪除的數據（is_deleted=0）:param table: 表名字符串:param where: WHERE條件（不帶WHERE關鍵字），例如 "id = %s":param params: WHERE條件的參數列表:return: 受影響的行數"""data = {'is_deleted': 0, 'deleted_at': None}return self.update_data(table, data, where, params, auto_time=True)

    def get_count(self, table, where=None, params=None, with_deleted=False, only_deleted=False):"""查詢記錄總數:param table: 表名字符串:param where: WHERE條件（不帶WHERE關鍵字）:param params: WHERE條件的參數列表:param with_deleted: 是否包含已刪除數據:param only_deleted: 是否只返回已刪除數據:return: 記錄總數"""conditions = []query_params = []if where:conditions.append(where)if params:query_params.extend(params)# if not with_deleted:#     if only_deleted:#         conditions.append("is_deleted = 1")#     else:#         conditions.append("is_deleted = 0")where_clause = " AND ".join(conditions) if conditions else ""where_str = f"WHERE {where_clause}" if where_clause else ""sql = f"SELECT COUNT(*) AS count FROM {table} {where_str}"result = self.execute_query(sql, query_params, fetchone=True)return result['count'] if result else 0

在這里插入圖片描述