在Django中如何执行原生SQL查询详解_MsSqlserver

在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)