Django REST Framework中自定义原生SQL查询的代码组织方案

核心原则：别把SQL逻辑放serializers.py #

serializers.py的核心职责是数据的序列化（将模型/原始数据转为响应格式）和反序列化（将请求数据转为可处理的格式），把SQL查询逻辑放在这里会违背单一职责原则，导致代码耦合、难以维护和复用。

推荐的组织方案 #

1. 单独创建数据访问层文件 #

在你的Django应用下新建queries.py、repositories.py或sql_utils.py这类文件，将原生SQL查询封装成独立的函数/类，让视图层只负责调用这些方法，实现逻辑分离。

示例（repositories.py）：

from django.db import connection

def get_all_users():
    with connection.cursor() as cursor:
        cursor.execute("SELECT * FROM users")
        return cursor.fetchall()

def create_user(username, bio):
    with connection.cursor() as cursor:
        cursor.execute(
            "INSERT INTO users (username, bio) VALUES (%s, %s)",
            [username, bio]
        )

对应修改views.py：

from rest_framework.views import APIView
from rest_framework.response import Response
from .repositories import get_all_users, create_user

class UsersView(APIView):
    def get(self, request):
        users_data = get_all_users()
        return Response(users_data)

    def post(self, request):
        data = request.data
        create_user(data.get('username'), data.get('bio'))
        return Response("Ok")

2. 封装到模型管理器（Model Manager）中 #

如果原生SQL是和特定模型绑定的，推荐自定义模型管理器，把查询逻辑放在管理器里，符合Django的模型层设计思路。

示例（models.py）：

from django.db import models, connection

class UserManager(models.Manager):
    def get_all_users(self):
        with connection.cursor() as cursor:
            cursor.execute("SELECT * FROM users")
            return cursor.fetchall()
    
    def create_user(self, username, bio):
        with connection.cursor() as cursor:
            cursor.execute(
                "INSERT INTO users (username, bio) VALUES (%s, %s)",
                [username, bio]
            )

class User(models.Model):
    username = models.CharField(max_length=100)
    bio = models.TextField()
    
    objects = UserManager()

对应修改views.py：

from rest_framework.views import APIView
from rest_framework.response import Response
from .models import User

class UsersView(APIView):
    def get(self, request):
        users_data = User.objects.get_all_users()
        return Response(users_data)

    def post(self, request):
        data = request.data
        User.objects.create_user(data.get('username'), data.get('bio'))
        return Response("Ok")

额外注意事项 #

• 封装的SQL方法要保持单一职责，一个方法对应一个查询操作，避免大而全的函数。
• 始终使用参数化查询（如你代码中的%s占位符），杜绝直接拼接SQL字符串，防止SQL注入风险。
• 如果查询结果需要转为模型实例或结构化字典，可以在封装方法内完成转换，避免视图层处理额外逻辑。

内容的提问来源于stack exchange，提问作者Ivan