本文介绍如何针对 Django 应用程序中的SQL Server执行原始 SQL 查询。 原始 SQL 对于未通过 Django ORM 公开的操作非常有用，例如复杂的 Transact-SQL（T-SQL）、空间查询或性能关键操作。
使用 connection.cursor（）
直接通过 Django connection 的对象访问数据库游标：
from django.db import connection
def get_server_version():
with connection.cursor() as cursor:
cursor.execute("SELECT @@VERSION;")
row = cursor.fetchone()
return row[0]
该 with 语句确保游标在使用后正确关闭。
参数化查询
始终使用参数化查询来防止 SQL 注入。 将参数作为列表传递：
Note
以下示例使用 Django 的默认表命名约定<app_label>_<model_name>（例如）。 myapp_product 如果在模型中db_table重写Meta，请替换该名称。 还可以通过 Product._meta.db_table 在运行时读取解析后的名称。
from django.db import connection
def get_products_by_price(min_price, max_price):
with connection.cursor() as cursor:
cursor.execute(
"SELECT id, name, price FROM myapp_product WHERE price BETWEEN %s AND %s;",
[min_price, max_price],
)
results = cursor.fetchall()
return results
Important
切勿使用字符串格式或 f 字符串在 SQL 查询中嵌入值。 始终使用参数化查询（%s 带有参数列表的占位符），以防止 SQL 注入。
获取结果
Django 的游标提供了几种用于检索结果的方法：
from django.db import connection
def demonstrate_fetch_methods():
with connection.cursor() as cursor:
cursor.execute("SELECT id, name FROM myapp_product;")
# Fetch one row
row = cursor.fetchone()
# Fetch the next 10 rows (continues from where fetchone stopped)
rows = cursor.fetchmany(10)
# Fetch all remaining rows (continues from where fetchmany stopped)
all_rows = cursor.fetchall()
以字典的形式返回结果
使用 cursor.description 将行转换为字典：
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 get_all_products():
with connection.cursor() as cursor:
cursor.execute("SELECT id, name, price FROM myapp_product;")
return dictfetchall(cursor)
将 Manager.raw（） 用于模型查询
如果想要原始 SQL 但仍需要 Django 模型实例，请使用 Manager.raw()：
from myapp.models import Product
products = Product.objects.raw(
"SELECT id, name, price FROM myapp_product WHERE price > %s",
[10.00],
)
for product in products:
print(f"{product.name}: ${product.price}")
查询必须返回模型主键中定义的所有字段。 其他字段加载延迟。
执行 DDL 语句
对于 Django 不直接支持的架构操作，请使用原始 SQL：
from django.db import connection
def create_index():
with connection.cursor() as cursor:
cursor.execute(
"CREATE INDEX IX_product_name ON myapp_product (name) "
"INCLUDE (price);"
)
多个数据库连接
如果使用多个数据库，请指定要使用的连接：
from django.db import connections
def query_reporting_db():
with connections["reporting"].cursor() as cursor:
cursor.execute("SELECT COUNT(*) FROM myapp_product;")
return cursor.fetchone()[0]