本文說明如何從 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）。 如果你在模型的Meta中覆寫db_table，請改用該名稱。 你也可以在執行時用 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 陳述式
使用 raw SQL 來處理 Django 不直接支援的結構操作：
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]