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

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

news 2026/3/26 17:19:51

文章目录

在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 connection
def 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 connection
def 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 connections
def 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 connection
def update_data():
with connection.cursor() as cursor:
# INSERT
cursor.execute(
"INSERT INTO myapp_person (name, age) VALUES (%s, %s)",
['John', 30]
)
# UPDATE
cursor.execute(
"UPDATE myapp_person SET age = %s WHERE name = %s",
[31, 'John']
)
# DELETE
cursor.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 connection
def 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)