python3 django 使用原生sql及分頁

django 使用原生sql及分頁,類似于django rest framework 的分頁

1、sql_connections.py sql封裝文件

from django.db import connections


def exec_sql(sql, params=None, db='default'):
    """
    執(zhí)行sql，例如insert和update
    :param sql: sql語句
    :param params: sql語句參數(shù)
    :param db: Django數(shù)據(jù)庫名
    """ 
    cursor = connections[db].cursor()
    cursor.execute(sql, params)
    cursor.close()
    cursor.last_insert_id()
    return True


def fetchone_sql(sql, params=None, db='default', flat=False):
    """
    返回一行數(shù)據(jù)
    :param sql: sql語句
    :param params: sql語句參數(shù)
    :param db: Django數(shù)據(jù)庫名
    :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'):
    """
    返回一行數(shù)據(jù)
    :param sql: sql語句
    :param params: sql語句參數(shù)
    :param db: Django數(shù)據(jù)庫名
    :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):
    """
    返回全部數(shù)據(jù)
    :param sql: sql語句
    :param params: sql語句參數(shù)
    :param db: Django數(shù)據(jù)庫名
    :param flat: 如果為True，只返回每行數(shù)據(jù)第一個字段值的元組，例如：(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'):
    """
    返回全部數(shù)據(jù)
    :param sql: sql語句
    :param params: sql語句參數(shù)
    :param db: Django數(shù)據(jù)庫名
    :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


def get_s_sql(table, keys, conditions, isdistinct=0):
    '''
    生成select的sql語句
    @table，查詢記錄的表名
    @key，需要查詢的字段
    @conditions,插入的數(shù)據(jù)，字典
    @isdistinct,查詢的數(shù)據(jù)是否不重復(fù)
    '''
    if isdistinct:
        sql = 'select distinct %s ' % ",".join(keys)
    else:
        sql = 'select  %s ' % ",".join(keys)
    sql += ' from %s ' % table
    if conditions:
        sql += ' where %s ' % dict_2_str_and(conditions)
    return sql


def dict_2_str(dictin):
    '''
    將字典變成，key='value',key='value' 的形式
    '''
    tmplist = []
    for k, v in dictin.items():
        tmp = "{k} = %({v})s".format(k=str(k), v=str(v))
        tmplist.append(' ' + tmp + ' ')
    return ','.join(tmplist)


def dict_2_str_and(dictin):
    '''
    將字典變成，key='value' and key='value'的形式
    '''
    tmplist = []
    for k, v in dictin.items():
        tmp = "{k} = %({v})s".format(k=str(k), v=str(v))
        tmplist.append(' ' + tmp + ' ')
    return ' and '.join(tmplist)

2、sql_paginator.py 分頁文件

from urllib import parse

from django.core.paginator import Paginator
from django.utils.encoding import force_str

from .sql_connections import fetchone_sql, fetchall_to_dict

page_query_param = "page"


def replace_query_param(url, key, val):
    """
    Given a URL and a key/val pair, set or replace an item in the query
    parameters of the URL, and return the new URL.
    """
    (scheme, netloc, path, query, fragment) = parse.urlsplit(force_str(url))
    query_dict = parse.parse_qs(query, keep_blank_values=True)
    query_dict[force_str(key)] = [force_str(val)]
    query = parse.urlencode(sorted(list(query_dict.items())), doseq=True)
    return parse.urlunsplit((scheme, netloc, path, query, fragment))


def remove_query_param(url, key):
    """
    Given a URL and a key/val pair, remove an item in the query
    parameters of the URL, and return the new URL.
    """
    (scheme, netloc, path, query, fragment) = parse.urlsplit(force_str(url))
    query_dict = parse.parse_qs(query, keep_blank_values=True)
    query_dict.pop(key, None)
    query = parse.urlencode(sorted(list(query_dict.items())), doseq=True)
    return parse.urlunsplit((scheme, netloc, path, query, fragment))


def get_next_link(request, page):
    if not page.has_next():
        return None
    url = request.build_absolute_uri()
    page_number = page.next_page_number()
    return replace_query_param(url, page_query_param, page_number)


def get_previous_link(request, page):
    if not page.has_previous():
        return None
    url = request.build_absolute_uri()
    page_number = page.previous_page_number()
    if page_number == 1:
        return remove_query_param(url, page_query_param)
    return replace_query_param(url, page_query_param, page_number)


def paginator(request, data_list, page_size, page):
    """
    封裝Django分頁
    :param data_list: sql語句
    :param page_size: 每頁顯示多少條數(shù)據(jù)
    :param page: 當(dāng)前第幾頁
    :return
    """
    page = int(page)
    page_size = int(page_size)
    pages = Paginator(data_list, page_size)
    """封裝Django分頁"""
    # 防止超出頁數(shù)
    if not page > 0:
        page = 1
    if page > pages.num_pages:
        page = pages.num_pages
    p = pages.page(page)  # 獲取本頁數(shù)據(jù)

    return dict(
        [
            ("page", page),
            ("total_page", pages.num_pages),
            ("count", pages.count),
            ("next", get_next_link(request, p)),
            ("previous", get_previous_link(request, p)),
            ("items", p.object_list),
        ]
    )


class QueryWrapper(object):
    """查詢集包裝器。實現(xiàn)django Paginator需要的必要方法，實現(xiàn)和query一樣使用Paginator分頁"""

    def __init__(self, sql, count_word=None, params=None, db="default"):
        """
        :param sql: sql語句
        :param count_word: 計算總數(shù)的字段,select count(count_word) from table,命名方式：_count.字段名
        :param params: sql語句的params參數(shù)
        :param db: 數(shù)據(jù)庫名稱（Django配置）
        """
        self.db = db
        self.sql = sql
        self.count_word = count_word
        self.params = params

    def count(self):
        """計算總頁數(shù)"""
        if self.count_word:
            sql = """select count(%s) from (%s) _count""" % (self.count_word, self.sql)
        else:
            sql = """select count(*) from (%s) _count""" % (self.sql)
        return fetchone_sql(sql, self.params, db=self.db, flat=True)  # 返回總頁數(shù)

    def __getitem__(self, index):
        """ index: slice(0, 3, None)"""
        sql = self.sql + ' LIMIT {start}, {num}'.format(start=index.start, num=index.stop - index.start)
        return fetchall_to_dict(sql, self.params, db=self.db)  # 字典列表形式返回

3、使用案例

# 在view.py 視圖中引用
from raw_sql.sql_paginator import QueryWrapper, paginator

def city(request):
    """查詢opm用戶列表"""
    page = request.GET.get('page', 1)
    page_size = request.GET.get('page_size', 10)
    sql = """
    SELECT DISTINCT
    a.province_code AS province_code,
    a.province_name AS province_name
    FROM
    city a
    LEFT JOIN city_area b ON a.province_code = b.code  
    WHERE b.area = %(area)s
    """
    params = {"area": area}
    count_word = "_count.province_code"
    query = QueryWrapper(sql, count_word, params, "opm")
    result = paginator(request, query, page_size, page)
    return HttpResponse(json.dumps(result, ensure_ascii=False), content_type='application/json')

# 輸出
result = {
    "page": 1,
    "total_page": 1,
    "count": 3,
    "next": null,
    "previous": null,
    "items": [
        {
            "province_code": "35",
            "province_name": "福建"
        },
        {
            "province_code": "27",
            "province_name": "湖北"
        },
        {
            "province_code": "89",
            "province_name": "海南"
        }
    ]
}