Django原生sql使用Paginator分页

雨点打透心脏的1/2处 2022-04-03 01:58 364阅读 0赞

使用Django肯定经常使用Paginator分页，很便捷。但是他可接受的分页对象必须是django orm的查询集或者list、tuple。当需要使用原生sql查询数据，分页就无法使用Paginator。

Paginator其实只需要实现两个方法`count`和`__getitem__`就可以自定义一个让Paginator分页器支持的对象，然后就可以使用Paginator分页了，不需要单独对原生sql写分页逻辑

from functools import my_custom_sql
    
    
    class QueryWrapper:
        """查询集包装器。实现django Paginator需要的必要方法，实现和query一样使用Paginator分页"""
    
        def __init__(self, sql, model_class=None):
            """
            :param sql: sql语句
            """
            self.sql = sql.replace(';', '')
            self.model_class = model_class
    
        def count(self):
            """计算总页数"""
            sql = """select count(*) as count from (%s) _count""" % self.sql
            data = my_custom_sql(sql)
            if data:
                return data[0]['count']
            return 0
    
        def __getitem__(self, k):
            x, y = k.start, k.stop
            sql = self.sql + ' LIMIT {start}, {num}'.format(start=x, num=y - x)
            result = my_custom_sql(sql)
            if self.model_class:
                result = [self.model_class(**item) for item in result]
            return result
    
        def all(self):
            return my_custom_sql(self.sql)

#### 关于`__getitem__`的说明: ####

Python 2.0以后不建议使用`__getslice__(self, i, j)`,Python 3.0以上已废弃了`__getslice__`方法, 可以使用slice对象作为`__getitem__()`的参数来实现切片。

##### `__getitem__(self, key)` #####

求表达式self\[key\]的值，对于序列类型，key应该是整数或slice对象。对于切片，key是一个slice对象。  
slice对象: slice(start, stop, step) 用于表示切片参数，未提供的参数将为None

#### 使用： ####

from utils import QueryWrapper
    
    class User(models.Model):
        username = models.CharField(max_length=100)
        first_name = models.CharField(max_length=100)
    
    sql = 'select id, username, first_name from auth_user'
    queryset = QueryWrapper(sql, User)
    
    count = queryset.count() 
    data = queryset.all()
    
    # 在Django中使用
    from django.core.paginator import Paginator
    pages = Paginator(queryset, per_page=10)
    page = pages.page(page_no) # 获取某页数据
    
    
    # 在django rest framework中使用
    page = self.paginate_queryset(queryset)
    results = self.get_paginated_response(page).data
    print(results)
    
    >>>
    {
    	"count": 25,
    	"next": "http://127.0.0.1:8888/test/?page=2",
    	"previous": null,
    	"results": [{
    		"id": 11349230,
    		"username": "张三",
    		"phone": "1440182340944",
    	
    	}, {
    		"id": 11344204,
    		"username": "李四",
    		"phone": "1440182333431",
    	},..
    }

关于pymysql默认游标（pymysql.cursors.Cursor）获取的数据是元组类型，如果想要字典类型的数据，需要指定cursor为`pymysql.cursors.DictCursor`

import pymysql
    #连接数据库
    conn = pymysql.connect(host='192.168.1.152',port= 3306,user = 'root',passwd='123123',db='test') #db：库名
    #设置游标类型，默认游标类型为元祖形式
    #将游标类型设置为字典形式
    cur = conn.cursor(cursor=pymysql.cursors.DictCursor)
    cur.execute("select * from lcj")  #逼表中所有的操作都可以再此进行操作
    #将lcj表中所有数据以字典形式输出
    ret = cur.fetchall()
    print(ret)   #[{'age': 18, 'tel': '13520617734', 'name': 'xiaoluo', 'id': 1, 'sex': '?'},

MySQLdb和pymysql类似，默认游标为`MySQLdb.cursors.BaseCursor`，获取的数据是元组类型，如果想要字典类型的数据，就要设置cursorclass参数为`MySQLdb.cursors.DictCursor`类。

conn = MySQLdb.connect(host='localhost', user='root', passwd='123456',db='test' cursorclass=MySQLdb.cursors.DictCursor)
    cur = conn.cursor()
    
    或者
    cur = conn.cursor(cursorclass=MySQLdb.cursors.DictCursor)
    cur.close()

附django原生sql查询封装常用方法

def fetchone_sql(sql, params=None, db='default', flat=False):
        """
        返回一行数据
        :param sql: sql语句
        :param params: sql语句参数
        :param db: Django数据库名
        :param flat: 如果为True，只返回第一个字段值，例如：id
        :return: 例如：(id, 'username', 'first_name')
        """
        cursor = connections[db].cursor()
        cursor.execute(sql, params)
        fetchone = cursor.fetchone()
        cursor.close()
        if fetchone:
            fetchone = fetchone[0] if flat else fetchone
        return fetchone
    
    
    def fetchone_to_dict(sql, params=None, db='default'):
        """
        返回一行数据
        :param sql: sql语句
        :param params: sql语句参数
        :param db: Django数据库名
        :return: 例如：{"id": id, "username": 'username', "first_name": 'first_name'}
        """
        cursor = connections[db].cursor()
        cursor.execute(sql, params)
        desc = cursor.description
        row = dict(zip([col[0] for col in desc], cursor.fetchone()))
        cursor.close()
        return row
    
    
    def fetchall_sql(sql, params=None, db='default', flat=False):
        """
        返回全部数据
        :param sql: sql语句
        :param params: sql语句参数
        :param db: Django数据库名
        :param flat: 如果为True，只返回每行数据第一个字段值的元组，例如：(id1, id2, id3)
        :return: 例如：[(id, 'username', 'first_name')]
        """
        cursor = connections[db].cursor()
        cursor.execute(sql, params)
        fetchall = cursor.fetchall()
        cursor.close()
        if fetchall:
            fetchall = tuple([o[0] for o in fetchall]) if flat else fetchall
        return fetchall
    
    
    def fetchall_to_dict(sql, params=None, db='default'):
        """
        返回全部数据
        :param sql: sql语句
        :param params: sql语句参数
        :param db: Django数据库名
        :return: 例如：[{"id": id, "username": 'username', "first_name": 'first_name'}]
        """
        cursor = connections[db].cursor()
        cursor.execute(sql, params)
        desc = cursor.description
        object_list = [
            dict(zip([col[0] for col in desc], row))
            for row in cursor.fetchall()
        ]
        cursor.close()
        return object_list