赞同 1
分享

优雅的使用Django

简介:Django项目的一大特点:“开机即用”Django是PythonWEB框架中名声最大的框架,它诞生于2003年堪萨斯(Kansas)州 Lawrence Journal-World 报纸的新闻站点。
  2021.01.31
  Bug Man
  1
  137
  3.237.27.159
  中国.上海
 
 

1.面对数据库Lost connection to MySQL server during query的报错提示:

在我们写一些中间件(rabbit、Kafka)消费者端的时候,通常都是以脚本的方式独立运行的,为了更加优雅的时候数据库我们需要去加载Django项目的上下文环境(就是manage.py中的DJANGO_SETTINGS_MODULE)。

这样我们就可以使用ORM对象来操作绝大部分的数据库操作,但这个同时等待时间过长带来的连接丢失的问题就凸显出来了,如果是用mysql驱动包的形式连接数据库,每一次操作之前重新连接就不会有这种问题了。

如果我们把连接时长(CONN_MAX_AGE)改得足够大,是完全没有任何作用的。那针对连接丢失的问题Django已经为我们准备好断开旧的连接这样的一个方法,我们只需要捕捉操作数据库时报出连接丢失的错误,然后使用这个方法,我们就可以重新操作数据库了。

import django

try:
    # 业务逻辑
    pass
except Exception as e:
    django.db.close_old_connections()
    # 重复为操作完的业务逻辑

2.Django使用SQL语句来查询数据库:

在日常开发中大部分的数据库操作使用ORM对象是可以解决的,如果对于比较复杂的联表可能用SQL语句能够更高效的解决。通常情况下我们的项目中都会有一个连接数据库的公共方法/类,直接用公共的方法去访问项目的数据库是完全可以的,Django给我们提供了一个更好、更直接、更准确的方法去访问项目数据库。

相比之下Django提供给我们的执行原始SQL的方法,不用去配置一堆数据库信息(还容易出错),使用起来和直接取调用驱动包的方法是一样的,这里在官网的示例中也提供了SQL注入的写法(execute方法第一个参数后面是[params])。如果为了方便可以将这个使用方法封装成一个专门用来执行项目数据库SQL语句执行类,这样代码健壮性更好。

from django.db import connection

sqlstr = """SQL Syntax"""
cursor = connection.cursor()
cursor.execute(sqlstr)
rows = cursor.fetchall()
cursor.close()
3.Django项目上下文环境的加载:

在日常开发中只要不是项目接口内的视图需要使用项目内的方法类、ORM对象等项目数据,想要以项目内的使用方法一样时就需要用到项目上下文加载了,比如你再编写一个异步任务时需要用到ORM和一些常量时。

这个应该大部分使用Django的后端都知道的,但也有可能有没有见过这种写法的,这种写法主要从manage.py文件中汲取,设置环境变量然后加载Django达到使用项目内的一些方法类(需要写在脚本文件最前面)。

import os
import django

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'project_dir.settings')
django.setup()

4.Django日志打印

我目前工作中的web项目里面日志打印其实特别不好,任何异常在日志文件中或者是对应的日志数据表中看到的都是错误概述,并没有详细的堆栈追踪信息。

这种情况下我认为可以设计一种较好的试图函数编写风格,做到所有的试图函数的错误都可以记下概述和详细堆栈追踪信息。这里主要是改变记录错误日志的编写风格,那我们在日常开发中应该怎么记录日志的堆栈信息呢?

在Django中我们在打印错误日志的时候可以写成logger.error(e, exc_info=True)这样的风格,在error方法写上打印错误信息的配置,这样一来就可以在日志文件中看到堆栈追踪信息(就是DEBUG模式下抛出的错误)。

也可以直接获取堆栈信息traceback.format_exc(),这种方式是直接获取到变量,然后开发者想往哪个下游端口发就往哪发。还有一点需要注意的是,尽可能的在所有捕捉错误的地方打印堆栈追踪信息或是传递错误堆栈信息,如果不这样做可能就会存在包装多层错误捕获时内层并没有把错误堆栈信息抛出,只是随意抛出一个错误概述信息(这个对解决错误起不到很大的作用)。

# 打印到日志文件
import logging
logger = logging.getLogger("xxxxx")
logger.error(e, exc_info=True)

# 直接获取堆栈信息
import traceback
traceback.format_exc()

5.Django运行测试定时任务无法导入项目应用方法

不知道你的定时任务是写在项目的哪个位置的,我们的项目中是,哪个应用的定时任务就写在哪个应用的包下,这样会出现一种情况:单独运行该文件时,同项目其他应用下方法导入错误,这是因为作为脚本运行时没有加入项目路径到环境变量。

那么这种情况下一般有哪些方法解决呢?我所知道的就是两种方法,其一:在项目根目录下的tasks.py文件中去导入定时任务执行,这样可以解决找不到其他应用下方法的问题;其二:在文件头部写上项目根目录加入环境变量中。如果在生产中当前文件就是会被当作脚本来运行那么就选第二种;如果生产中是按照项目来定位该文件,比如celery来调用。那这个时候只是调试就可以在根目录task.py文件中来引入定时任务来测试。

import os
import sys

# 打包后添加当前目录到环境变量以便导入项目中的包
sys.path.append(os.path.dirname(os.path.abspath(__file__)))

6.查看ORM执行的sql

如果我们要做接口数据库查询的sql优化,除了可以从查询结果中查看Queryset.query属性来查看转换的sql,还可以用以下的方式来获取执行的sql。

from django.db import connection

print(connection.queries)
# [{'sql': 'SELECT polls_polls.id, polls_polls.question, polls_polls.pub_date FROM polls_polls', 'time': '0.002'}]

7.下载使用内存暂存Excel

在下载Excel表格的场景下我们可能思考过一个问题:能不能让我们需要下载的不存储到本地,而是从数据库中读取之后直接装载到Response对象中。以前我有这种疑问,后来我的同事给我看了一个例子使用到BytesIO容器,将需要保存的文件以这种容器的方式保存就不会保存到本地,我将例子改为适用于项目中的类。需要注意的是,必须限制这种方式下载的excel文件的大小,不可以几个GB的内容往内存中存储,以防内存溢出导致错误。这种方式只是提供一种区别于通常将文件先存储到本地之后再读取返回的方式,本质上没有大区别,只要能完成功能且不会有bug的解决方案都是好的方案。

import typing

from io import BytesIO

import xlwt

from django.http import HttpResponse


class GenerateExcelResponse:
    """ 生成excel响应类 """

    def __init__(self, file_name: str):
        """ 初始化响应头配置 """

        self.file_name = file_name

    @staticmethod
    def write_excel(
            sheet_name: str,
            excel_data: typing.List,
            excel_headers: typing.List = None,
            work_book: xlwt.Workbook = None) -> xlwt.Workbook:
        """ 写入excel数据 """

        work_book = work_book or xlwt.Workbook(encoding="utf-8")
        work_sheet = work_book.add_sheet(sheet_name)
        offset = 0
        # 处理表头
        if excel_headers:
            for index, title in enumerate(excel_headers):
                work_sheet.write(0, index, title)
            offset = 1
        # 处理内容
        for row_num, row_data in enumerate(excel_data):
            for index, cell_data in enumerate(row_data):
                work_sheet.write(row_num + offset, index, cell_data)
        return work_book

    def generate_response(self, work_book: xlwt.Workbook = None) -> HttpResponse:
        """ 生成响应类 """

        http_response = HttpResponse()
        output = BytesIO()
        work_book = work_book or xlwt.Workbook(encoding="utf-8")
        work_book.save(output)
        output.seek(0)
        http_response["Content-Type"] = "application/octet-stream"
        http_response["Content-Disposition"] = f"attachment;filename={self.file_name}.xls"
        http_response.write(output.getvalue())
        return http_response


if __name__ == "__main__":
    g_e_r = GenerateExcelResponse("data")
    work_book = g_e_r.write_excel(sheet_name=key, excel_data=excel_data, excel_headers=header_list, work_book=work_book)
    response = g_e_r.generate_response(work_book=work_book)