当前位置：首页 > news >正文

【python】在Django中，执行原生SQL查询

news 2025/10/21 9:48:04

文章目录

在Django中，执行原生SQL查询
- 1. 使用 `raw()` 方法
- 2. 使用数据库连接直接执行SQL
- - 基本用法：
  - 返回字典格式的结果：
- 3. 使用多个数据库
- 4. 执行写操作（INSERT/UPDATE/DELETE）
- 5. 事务处理
- 6. 实用的封装函数
- 注意事项

在Django中，执行原生SQL查询

1. 使用 `raw()` 方法

raw() 方法允许你执行原生SQL查询并返回模型实例：

from myapp.models import Person# 基本查询
people = Person.objects.raw('SELECT * FROM myapp_person WHERE age > %s', [18])# 遍历结果
for person in people:print(person.name, person.age)# 带参数的查询
people = Person.objects.raw('SELECT * FROM myapp_person WHERE age BETWEEN %s AND %s',[20, 30]
)

2. 使用数据库连接直接执行SQL

基本用法：

from django.db import connectiondef my_custom_sql():with connection.cursor() as cursor:# 执行查询cursor.execute("SELECT * FROM myapp_person WHERE age > %s", [25])# 获取所有结果rows = cursor.fetchall()# 或者逐行获取# row = cursor.fetchone()# rows = cursor.fetchmany(size=10)return rows

返回字典格式的结果：

from django.db import connectiondef dictfetchall(cursor):"""将游标结果转换为字典列表"""columns = [col[0] for col in cursor.description]return [dict(zip(columns, row))for row in cursor.fetchall()]def my_custom_sql():with connection.cursor() as cursor:cursor.execute("SELECT id, name, age FROM myapp_person")results = dictfetchall(cursor)return results

3. 使用多个数据库

如果你的项目配置了多个数据库：

from django.db import connectionsdef query_multiple_dbs():# 使用默认数据库with connections['default'].cursor() as cursor:cursor.execute("SELECT * FROM table1")# 处理结果# 使用其他数据库with connections['other_db'].cursor() as cursor:cursor.execute("SELECT * FROM table2")# 处理结果

4. 执行写操作（INSERT/UPDATE/DELETE）

from django.db import connectiondef update_data():with connection.cursor() as cursor:# INSERTcursor.execute("INSERT INTO myapp_person (name, age) VALUES (%s, %s)",['John', 30])# UPDATEcursor.execute("UPDATE myapp_person SET age = %s WHERE name = %s",[31, 'John'])# DELETEcursor.execute("DELETE FROM myapp_person WHERE age < %s",[18])

5. 事务处理

from django.db import transaction, connection@transaction.atomic
def complex_operation():with connection.cursor() as cursor:try:# 一系列SQL操作cursor.execute("UPDATE accounts SET balance = balance - %s", [100])cursor.execute("UPDATE accounts SET balance = balance + %s", [100])except Exception as e:# 出错时会自动回滚print(f"Error: {e}")

6. 实用的封装函数

from django.db import connectiondef execute_query(query, params=None, return_dict=True):"""执行SQL查询的通用函数"""with connection.cursor() as cursor:cursor.execute(query, params or [])if query.strip().upper().startswith('SELECT'):if return_dict:columns = [col[0] for col in cursor.description]return [dict(zip(columns, row)) for row in cursor.fetchall()]else:return cursor.fetchall()else:# 对于非SELECT查询，返回受影响的行数return cursor.rowcount# 使用示例
results = execute_query("SELECT name, age FROM myapp_person WHERE age > %s",[25]
)

注意事项

SQL注入防护：始终使用参数化查询，不要直接拼接字符串

# ✅ 安全：使用参数化查询
cursor.execute("SELECT * FROM users WHERE username = %s", [username])# ❌ 危险：字符串拼接（SQL注入风险）
cursor.execute(f"SELECT * FROM users WHERE username = '{username}'")# ❌ 危险：直接传递用户输入
cursor.execute("SELECT * FROM users WHERE username = %s" % username)