• 自学Python第二十二天- Django框架(四) iframe、其他组件和工具

    Django官方文档

    使用 Celery 异步操作

    Celery是由Python开发、简单、灵活、可靠的分布式任务队列，是一个处理异步任务的框架，其本质是生产者消费者模型，生产者发送任务到消息队列，消费者负责处理任务。Celery侧重于实时操作，但对调度支持也很好，其每天可以处理数以百万计的任务。

    celery 官方英文文档
    非官方中文文档

    由于 django 框架请求/响应的过程是同步的，框架本身无法实现异步响应。所以异步执行前端一般会使用 Ajax，后端则使用 Celery。另外 django-celery 插件已经有一段时间没有更新了，对于新版本的 python 和 django 会有适配问题，所以使用 celery

    一些参考文档

    celery + redis

    celery 使用步骤

    django-celery 使用步骤：

    1. 安装、配置
    2. 设置 celery 主体，创建应用对象
    3. 根据需要创建任务
    4. 数据库迁移，生成 celery 需要的数据表
    5. 调用任务，将其添加至消息队列
    6. 启动 celery，监听队列并执行任务
    7. 查看执行结果，对结果进行处理

    安装及环境配置

    首先安装 django-celery，因为 celery 本身不实现中间件 Broker 的功能，所以还需要使用中间件。django 和 redis 配合的不错，这里就使用 django-redis

    pip install django-celery django-redis

    这里需要注意的是，windows 在 celery4.0 之后不支持多进程方式，而是更换成了协程方式，所以要使用 eventlet 或 gevent。

    pip install eventlet

    然后在 settings.py 中添加配置信息

    ##### celery 配置 #####
from urllib.parse import  quote
PASSWORD = quote('123456')  #使用有特殊字符密码，带有特殊字符需要进行转换才能识别

# Broker配置，使用Redis作为消息中间件
BROKER_URL = f'redis://mast:{PASSWORD}@127.0.0.1:6379/8'.format(PASSWORD)		# 此格式为连接需要验证的 redis
# 后台结果，如果没有此参数则使用 orm 的数据库
CELERY_RESULT_BACKEND = f'redis://mast:{PASSWORD}@127.0.0.1:6379/9'.format(PASSWORD)	
# 结果序列化方案
CELERY_RESULT_SERIALIZER = 'json'
# 任务结果过期时间，秒
CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24
# 指定导入的任务模块，可以指定多个
CELERY_IMPORTS = ('app.tasks',)		# 参数为 tasks.py 文件路径(即 应用.tasks)
CELERY_TIMEZONE = ’Asia/Shanghai'	# 设置时区

CELERYD_LOG_FILE = BASE_DIR + "/logs/celery/celery.log"         # log路径
CELERYBEAT_LOG_FILE = BASE_DIR + "/logs/celery/beat.log"     # beat log路径

# 设置定时器策略
from datetime import timedelta

CELERYBEAT_SCHEDULE = {
    # 定时任务一
    u'邮件发送': {	# 任务名称
        'task': 'app.tasks.print_now',  # 需要执行的任务函数
        # 'schedule': crontab(minute='*/2'),	# 延迟
        'schedule': timedelta(seconds=5),  # 间隔5秒
        'args': ('现在时间',),  # 参数
    },
}
# schedule 参数是执行频率，可以是整型(秒)、timedelta对象、crontab对象
# 还可以设置 kwargs 字典型的关键字参数
# 对于 crontab 函数使用举例
# crontab(hour='*/24')	每隔24小时
# crontab(minute=30, hour=0)	每天的凌晨 00:30
# crontab(hour=6, minute=0, day_of_month='1')	每月1号的 6:00

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37

    celery 主体

    在主工程目录(settings.py 所在目录)添加 celery.py，主体代码就写在这里

    import os
from celery import Celery
from django.conf import settings

# 设置项目运行的环境变量 DJANGO_SETTINGS_MODULE
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'DAdmin.settings')	# 将 DAdmin.settings 添加到环境变量，根据工程名称改变

# 创建 celery 应用对象
app = Celery('AdminCelery')

# 加载配置
app.config_from_object('django.conf:settings')

# 如果在工程的应用中创建了tasks.py模块，那么celery应用会自动去添加任务
# 比如添加了一个任务，在 django 中会实时地检索出来
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16

    然后需要在项目中加载 celery 的 app，在项目主工程目录的 init.py 中，添加 all 属性

    from .celery import app as celery_app	#引入 celery.py 里的 app 对象

# 向项目模块中增加 celery_app 对象
__all__ = ['celry_app']

    • 1
    • 2
    • 3
    • 4

    创建任务 tasks

    每个任务本质上就是一个函数，写在 tasks.py 中。

    import datetime
from celery import shared_task
import logging
logger = logging.getLogger(__name__)

# 定时任务，在 settings.py 的 CELERYBEAT_SCHEDULE 中注册
# 参数 info 是在注册时传入
# 调用时也可以触发
@shared_task
def print_now(info):		
    print(info, datetime.datetime.now())
    # 也可以输出日志
    logger.info(info, datetime.datetime.now())

# 没有在 CELERYBEAT_SCHEDULE 中注册，只能通过调用触发
@shared_task
def add(x, y):
	time.sleep(5)
	return x + y

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19

    迁移数据库

    celery 的迁移和 django 其他的一样

    python manage.py makemigrations
    python manage.py migrate

    启动 celery

    接下来就是启动 celery 服务 worker，用来获取消息和执行任务

    celery -A django_test worker -l info
    mac、linux 系统使用此命令，-A 指定工程项目，-l 指定日志等级为 info

    celery -A django_test worker -l info -P eventlet
    windows 使用此命令，-A 指定工程项目，-P 指定执行单元使用 eventlet 实现后台异步操作，-l 指定日志等级为 info

    然后是 beat，启动后就能够执行定时任务了

    celery -A django_test beat -l info

    在视图函数中调用任务和获取结果

    非定时任务必须进行调用

    from .tasks import add		# 引入写好的任务

def add(request):		# 视图函数
	...
	task = add.delay(100, 200)	# 执行任务，返回任务异步结果对象
	return HttpResponse(json.dumps({'status': 'ok','task_id': task.task_id}),'application/json')

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6

    当调用时立刻能获得异步结果对象，此时对象的状态为 PENDING ，结果为 None。

    该对象的一些常用的属性和方法：

    • state 任务状态
    • task_id 任务id
    • result 任务结果
    • ready() 判断任务是否结束
    • wait(t) 等待t秒后获取结果，若任务执行完毕，则不等待直接获取结果，若任务在执行中，则wait期间一直阻塞，直到超时报错
    • successful() 判断任务是否成功，成功为True，否则为False

    通常获取该对象的 task_id 然后可以在另一函数中使用 task_id 获取任务状态和结果

    from celery import result

def get_result_by_taskid(request):
	task_id = request.GET.get('task_id')
	# 获取异步结果对象
	ar = result.AsyncResult(task_id)

    if ar.ready():
        return JsonResponse({'status': ar.state, 'result': ar.result})
    else:
        return JsonResponse({'status': ar.state, 'result': ''})

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11

    关于结果的存储

    celery 的结果可以使用 django-celery-results 包来方便的保存至数据库

    任务绑定

    Celery可通过task绑定到实例获取到task的上下文，这样我们可以在task运行时候获取到task的状态，记录相关日志等

    from celery import shared_task
import logging  
logger = logging.getLogger(__name__)
 
 
# 任务绑定
@shared_task(bind=True)		# bind=True 设置任务绑定
def add(self,x, y):			# 第一个参数 self 能获取任务实例对象
    try:
        logger.info('add__-----'*10)
        logger.info('name:',self.name)
        logger.info('dir(self)',dir(self))
        raise Exception
    except Exception as e:
        # 出错每4秒尝试一次，总共尝试4次
        self.retry(exc=e, countdown=4, max_retries=4)    
    return x + y

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17

    任务钩子

    Celery在执行任务时，提供了钩子方法用于在任务执行完成时候进行对应的操作，在Task源码中提供了很多状态钩子函数如：on_success(成功后执行)、on_failure(失败时候执行)、on_retry(任务重试时候执行)、after_return(任务返回时候执行)

    from celery import Task
 
class MyHookTask(Task):
     def on_success(self, retval, task_id, args, kwargs):
        logger.info(f'task id:{task_id} , arg:{args} , successful !')
 
    def on_failure(self, exc, task_id, args, kwargs, einfo):
        logger.info(f'task id:{task_id} , arg:{args} , failed ! erros: {exc}')
 
    def on_retry(self, exc, task_id, args, kwargs, einfo):
        logger.info(f'task id:{task_id} , arg:{args} , retry !  erros: {exc}')
 
# 在对应的task函数的装饰器中，通过 base=MyHookTask 指定
@shared_task(base=MyHookTask, bind=True)
def add(self,x, y):
    logger.info('add__-----'*10)
    logger.info('name:',self.name)
    logger.info('dir(self)',dir(self))
    return x + y

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19

    flower

    flower 是 celery 的一个图形化管理界面

    富文本方案

    django-mdeditor

    DjangoUeditor

    django2集成DjangoUeditor富文本编辑器

    django-RESTful

    REST 即表述性状态传递（Representational State Transfer），它是一种针对网络应用的设计和开发方式，可以降低开发的复杂性，提高系统的可伸缩性。REST通常基于使用HTTP，URI，和XML以及HTML这些现有的广泛流行的协议和标准。

    RESTful API 设计规范
    官方网站
    中文文档

    RESTful API四大基本原则：

    • 为每个资源设置URI
    • 通过XML / JSON进行数据传递
    • 无状态连接，服务器端不应保存过多上下文状态，即每个请求都是独立的
    • 使用HTTP动词：GET POST PUT DELETE

    安装和环境配置

    django-RESTful 需要以下包支持(除了主插件程序包外，其他的包为可选项)

    • DjangoRESTframework - 主插件程序包
    • PyYAML, uritemplate (5.1+, 3.0.0+) - Schema生成支持。
    • Markdown (3.0.0+) - 为browsable API 提供Markdown支持。
    • Pygments (2.4.0+) - 为Markdown处理提供语法高亮。
    • django-filter (1.0.1+) - Filtering支持。
    • django-guardian (1.1.1+) - 对象级别的权限支持。

    可以根据需要安装相应的包

    pip install djangorestframework
pip install markdown       # 为browsable API 提供Markdown支持。
pip install django-filter  # Filtering支持。

    • 1
    • 2
    • 3

    使用 django-RESTful 需要注册应用到 settings.py 的 INSTALLED_APPS 中

    INSTALLED_APPS = [
    ...
    'rest_framework',
]

    • 1
    • 2
    • 3
    • 4

    REST framework API的所有全局设定都会放在一个叫REST_FRAMEWORK的配置词典里，并添加到 settings.py 中：

    REST_FRAMEWORK = {
	# 在这里配置访问许可
    'DEFAULT_PERMISSION_CLASSES': [
        # 'rest_framework.permissions.DjangoModelPermissionsOrAnonReadOnly'   # 匿名只读，登录用户可用
        'rest_framework.permissions.IsAdminUser',       # 只能管理员使用
    ],
    # 配置分页器
    'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.PageNumberPagination',
    'PAGE_SIZE': 10,
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10

    如果打算用browsable API(一个可视化API测试工具)，可能也会用REST framework的登录注销视图。可以添加路由到根目录的urls.py文件

    urlpatterns = [
    ...
    path('api-auth/', include('rest_framework.urls'))	# 路由路径可以更改
]

    • 1
    • 2
    • 3
    • 4

    测试安装和配置

    以一个实例进行测试：创建一个读写API来访问项目的用户信息。

    在根目录的 urls.py 中创建API

    # urls.py 

from django.urls import path, include
from django.contrib.auth.models import User
from rest_framework import routers, serializers, viewsets

# 序列化器是用来定义API的表示形式。
class UserSerializer(serializers.HyperlinkedModelSerializer):
    class Meta:
        model = User
        fields = ['url', 'username', 'email', 'is_staff']

# ViewSets定义视图的行为。
class UserViewSet(viewsets.ModelViewSet):
    queryset = User.objects.all()
    serializer_class = UserSerializer

# 路由器提供一个简单自动的方法来决定URL的配置。
router = routers.DefaultRouter()
router.register(r'users', UserViewSet)	# 注册路由

# 通过URL自动路由来给我们的API布局。
# 此外，我们还要把登录的URL包含进来。
urlpatterns = [
    path('', include(router.urls)),		# 注册 REST 路由到主路由
    path('api-auth/', include('rest_framework.urls', namespace='rest_framework'))
]

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27

    现在可以在浏览器中打开(默认页，即第一条路由)http://127.0.0.1:8000/，查看 ‘user’ API了。如果使用了右上角的登录控制，还可以在系统中添加、创建并删除用户。这样就是安装和配置成功了。

    基本用法

    在实际项目中，django-RESTful 的使用分为三个步骤：

    1. 创建序列化器：及定义API的表现形式
    2. 创建视图：用来处理api请求并返回响应
    3. 注册路由：将视图注册到路由中

    创建序列化器

    定义序列化程序，可以创建一个新的文件，这里使用 serializers.py

    from django.contrib.auth.models import User
from rest_framework import serializers

# 定义序列化处理器，继承自 serializers.HyperlinkedModelSerializer，使用超链接关系
class UserSerializer(serializers.ModelSerializer):
    class Meta:
    	# model 定义使用的数模型
        model = User
        # fields 定义使用数据模型的哪些字段
        fields = ('url', 'username', 'email', 'groups')

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10

    django-RESTful 提供了3个序列化类的模板: Serializer、ModelSerializer、HyperlinkedModelSerializer

    • Serializer : 标准序列化类。需要完全自定义声明相关字段，且需实现 create(**validate_data) 和 update(instance, **validate_data) 两个函数
    • ModelSerializer : Serialize 的子类，使用数据模型的序列化类。通过声明 Meta 类来指定相关属性(使用的模型类和需要序列化的字段)
    • HyperlinkedModelSerializer : ModelSerializer 的子类，

    创建视图

    django-RESTful 提供了一些预设的视图处理类

    from rest_framework import viewsets
from django.contrib.auth.models import User
from .serializers import UserSerializer

# 视图处理类
class UserViewSet(viewsets.ModelViewSet):
	# queryset 为查询的结果集合
    queryset = User.objects.all()
    # serializer_classs 使用的序列化器
    serializer_class = UserSerializer

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10

    创建路由

    可以在应用中创建路由，并注册到主路由中

    # 应用中的 urls.py

from django.urls import path, include
from rest_framework import routers

# 路由器提供一个简单自动的方法来决定URL的配置。
# 创建路由器
router = routers.DefaultRouter()
# 在路由器中注册路由及处理器(FBV或CBV)
router.register(r'users', UserViewSet)

# 通过URL自动路由来给我们的API布局。
urlpatterns = [
    path('', include(router.urls)),		# 将 REST 路由器中的路由添加到 app 的路由中
    path('api-auth/', include('rest_framework.urls')),	# 添加 browsable API 的登录路由
]

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16

    自定义 API 相关

    根据实际情况自定义序列化器或者使用自定义的视图函数，有一些需要使用到的

    序列化与反序列化

    在自定义的视图中，需要将数据通过序列化器序列化后发出响应，接收到的请求也需要通过反序列化获取具体数据

    from .serializers import UserSerializer
from django.contrib.auth.models import User

s1 = UserSerializer(User.objects.get(pk=1))		# 序列化单条数据
s2 = UserSerializer(User.objects.all(),	many=True)		# 序列化多条数据

# 渲染成Json字符串
from rest_framework.renderers import JSONRenderer
content = JSONRenderer().render(s1.data)
# 对已经渲染的Json字符串反序列化为Json对象
data = JSONParser().parse(io.BytesIO(content))
# 或直接解析 request 对象(POST)
# from rest_framework.parsers import JSONParser
# data = JSONParser().parse(request)
# serializer = UserSerializer(data=data)	# 根据提交的参数获取序列化对象
# if serizlizer.is_valid():		# 通过验证
#     serializer.save()		# 保存数据

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17

    关联关系的序列化

    在查看和测试API的过程中，会发现某些数据表的外键指向的是关联表的数据链接，而不是数据内容。如果要使用数据内容，就需要将关联关系序列化。在定义序列化器时，将外键字段声明成为字符串关系字段即可。

    from django.contrib.auth.models import User
from rest_framework import serializers

# 定义序列化处理器，继承自 serializers.HyperlinkedModelSerializer，使用超链接关系
class UserSerializer(serializers.HyperlinkedModelSerializer):
	# 在此定义字段
	# 定义关系字段，将数据信息字符串化
	groups = serializers.StringRelatedField()	# 如果是对多关系(对方表是多端)则需要参数 many=True
	# 也可以将关系对象整体序列化(即返回的 json 中，此字段是给包含了所有数据的 json 对象)
	# groups = GroupSerializer()	# 需先定义 GroupSerializer 序列化器，且如果对方是多端也需要 many=True
    class Meta:
    	# model 定义使用的数模型
        model = User
        # fields 定义使用数据模型的哪些字段
        fields = ('url', 'username', 'email', 'groups')

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15

    这个方法在一对一、一对多、多对多关系都适用。如果对方是多端(即此表的该外键字段或反查字段有多个数据)，添加参数 many=True 即可。其中一些关联字段类型为

    • StringRelatedField : 获取关联模型对象字符串化(使用 str() 函数返回的字符串)
    • PrimarykeyRelatedField : 获取关联模型对象的主键
    • SlugRelatedField : 获取关联模型对象指定字段(slug_field 参数值)
    • HyperlinkedRelatedField : 获取关联模型对象的查询api链接，此种方式为默认方式

    自定义视图处理器

    使用的封装好的django-RESTful视图处理类能够方便的获取请求返回响应，但是如果需要使用自定义的视图处理器，则可使用 @api_view 装饰器(FBV)或 APIView 基类(CBV)。

    FBV 方式举例，需注意的是注册路由使用 path 方法直接到 urlpatterns

    from rest_framework.decorators import api_view
from .serializers import UserSerializer
from rest_framework.response import Response
from rest_framework.parsers import JSONParser
from django.contrib.auth.models import User
from django.http import JsonResponse

@api_view(['GET', 'POST'])
def user_func(request, pk=None):
    if request.method == 'GET':
        if not pk:
            queryset = User.objects.all()
            serializer = UserSerializer(queryset, many=True, context={'request': request})
        else:
            queryset = User.objects.get(pk=pk)
            serializer = UserSerializer(queryset, context={'request': request})
        # django-RESTful 重新封装了 Response，可以直接序列化
        # 可以根据请求类型返回数据，例如浏览器直接请求会返回管理页面，请求 json 格式则返回 json 数据
        # 也可以在请求地址后添加参数 ?format=json 指定获取 json 数据或 api 页面
        return Response(serializer.data)
        # return JsonResponse(serializer.data)		# 返回的是序列化后的json字符串

    elif request.method == 'POST':
    	# 接收的 request 是 django-RESTful 重新封装后的 request 对象
    	# 可以直接将其内容反序列化，然后通过序列化器从数据库中查询相关数据，再进行序列化
        data = JSONParser().parse(request)
        serializer = UserSerializer(data, context={'request': request})
        # 序列化器对象可以进行验证
        if serializer.is_valid():
            serializer.save()	# 保存至数据库
            return Response(serializer.data, status=201)
        else:
            return Response(serializer.errors, status=400)

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33

    CBV 方式举例，需注意的是注册路由使用 path 方法直接到 urlpatterns

    from rest_framework.views import APIView
from .serializers import UserSerializer
from django.contrib.auth.models import User
from rest_framework.response import Response
from rest_framework.parsers import JSONParser

class UserClass(APIView):
    def get(self, request, pk=None):
        if not pk:
            queryset = User.objects.all()
            serializer = UserSerializer(queryset, many=True, context={'request': request})
        else:
            queryset = User.objects.get(pk=pk)
            serializer = UserSerializer(queryset, context={'request': request})
        return Response(serializer.data)
        
    def post(self,request):
        data = JSONParser().parse(request)
        serializer = UserSerializer(data, context={'request': request})
        # 序列化器对象可以进行验证
        if serializer.is_valid():
            serializer.save()	# 保存至数据库
            return Response(serializer.data, status=201)
        else:
            return Response(serializer.errors, status=400)

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25

    获取 request 的数据

    rest_framework 的 request 和 django 的 request 不太一样，获取数据是使用 request.data ，类似于 django 的 request.body 但是不是字节码，而是字符串。

    name = request.data.get('name', None)

    • 1

    关于授权认证

    默认情况下， APIView 中的相关接口方法不验证权限(授权)，对资源并不安全，所以需要增加验证。

    首先在 settings.py 中的 REST_FRAMEWORK 字段配置权限访问许可

    # settings.py

REST_FRAMEWORK = {
	'DEFAULT_AUTHENTICATION_CLASSES': [			# 默认使用的授权认证
		'rest_framework.authentication.BasicAuthentication',	# 基本授权认证
		'rest_framework.authentication.SessionAuthentication',		# 基于session的授权认证
	]
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8

    然后可以在视图中指定验证方式和许可类型

    class EcampleView(APIView):
	authentication_classes = (SessionAuthentication, BasicAuthentication)	# 验证方式
	permission_classes = (IsAuthenticated,)		# 许可类型

	def get(self, request):
		pass

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6

    但是这种授权是基于 session 登录的，即 auth 模块的认证。在 api 接口中，通常会使用 token 进行认证。

    django-RESTful 使用的 token 验证

    TokenAuthentication 提供了简单的基于 Token 的HTTP认证方案，适用于客户端 - 服务器设置，如本地桌面和移动客户端。要在 django-RESTful 中使用 TokenAuthentication，需要配置认证类包含 TokenAuthentication，另外需要注册 rest_framework.authtoken 这个 app。需注意的是，确保在修改设置后运行一下 manage.py migrate ，因为 rest_framework.authtoken 会提交一些数据库迁移操作。

    # settings.py
INSTALLED_APPS = [
	...
	'rest_framework.authtoken',
]
REST_FRAMEWORK = {
	'DEFAULT_AUTHENTICATION_CLASSES': [			# 默认使用的授权认证
		'rest_framework.authentication.TokenAuthentication',	# token授权认证
	]
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10

    创建令牌

    from rest_framework.authtoken.models import Token
from django.contrib.auth.models import User

user = User.objects.get(pk=1)		# 获取用户
token = Token.objects.create(user=user)		# 根据用户创建 token 实例
print(token.key)		# token.key 就是需要验证的字段

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6

    通常会在每个用户创建时，创建对应的 token，可以捕捉用户的 post_save 信号

    from django.conf import settings
from django.db.models.signals import post_save
from django.dispatch import receiver
from rest_framework.authtoken.models import Token

@receiver(post_save, sender=settings.AUTH_USER_MODEL)
def create_auth_token(sender, instance=None, created=False, **kwargs):
    if created:
        Token.objects.create(user=instance)

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9

    也可以为现有用户生成令牌

    from django.contrib.auth.models import User
from rest_framework.authtoken.models import Token

for user in User.objects.all():
    Token.objects.get_or_create(user=user)
    # 此方法可以返回2个值，分别为 token 对象和 created

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6

    验证令牌

    对客户端进行身份验证，token需要包含在名为 Authorization 的HTTP头中。密钥应该是以字符串"Token"为前缀，以空格分割的两个字符串。例如：

    function ajax_get() {
	fetch('/api/user/, {
		headers: {
			'Authorization': 'Token 9944b09199c62bcf9418ad846dd0e4bbdfc6ee4b'
		}
	}).then(response=>response.json())
		.then(data=>{
			console.log(data)
		})
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10

    如果认证成功，TokenAuthentication 提供以下认证信息：

    • request.user 将是一个Django User 实例。
    • request.auth 将是一个rest_framework.authtoken.models.Token 实例。

    那些被拒绝的未经身份验证的请求会返回使用适当WWW-Authenticate标头的HTTP 401 Unauthorized响应。例如：

    WWW-Authenticate: Token

    • 1

    通过暴露 api 端点获取令牌

    当使用TokenAuthentication时，可能希望为客户端提供一个获取给定用户名和密码的令牌的机制。 REST framework 提供了一个内置的视图来提供这个功能。要使用它，需要将 obtain_auth_token 视图添加到你的URLconf：

    from rest_framework.authtoken import views
urlpatterns += [
    path('api-token-auth/', views.obtain_auth_token)	# url 路由可以自定义
]

    • 1
    • 2
    • 3
    • 4

    当使用form表单或JSON将有效的username和password字段POST提交到视图时，obtain_auth_token 视图将返回JSON响应：

    { 'token' : '9944b09199c62bcf9418ad846dd0e4bbdfc6ee4b' }

    • 1

    请注意，默认的obtain_auth_token视图显式使用JSON请求和响应，而不是使用settings中配置的默认渲染器和解析器类。如果需要自定义版本的obtain_auth_token视图，可以通过重写ObtainAuthToken类，并在url conf中使用它来实现。默认情况下，没有权限或限制应用于obtain_auth_token视图。如果你希望应用限制，则需要重写视图类，并使用throttle_classes属性包含它们。

    邮件

    django 有组件支持发送邮件

    配置邮件信息

    发送邮件需要配置邮件服务器信息

    EMAIL_HOST = 'smtp.163.com'
EMAIL_PORT = 25
EMAIL_HOST_USER = 'emailUser@163.com'
EMAIL_HOST_PASSWORD = 'emailPassword'

    • 1
    • 2
    • 3
    • 4

    需注意的是 EMAIL_HOST_PASSWORD 使用的是邮件服务器的授权码而不是登录密码。

    发送邮件的方法

    django 使用 django.core.mail.send_mail() 发送邮件，其用法为

    from django.core.mail import send_mail

send_mail(title, message, from_email, recipient_list, fail_silently, auth_user, auth_password, connection, html_message)

    • 1
    • 2
    • 3
    • title : 邮件标题，字符串
    • message : 邮件正文，字符串，必须有此参数，哪怕是空字符串。
    • from_mail : 发送者，字符串
    • recipient_list : 接收者，字符串列表，可以有多个接收者
    • fail_silently :
    • auth_user :
    • auth_password :
    • connection :
    • html_message : 正文中的 html 文本，可以代替 message 参数，接受 html 内容

    单元测试

    可以使用 unittest 库进行单元测试。创建测试类，继承自 unittest 库中的相应类，然后创建方法。方法可以独立执行，用作测试使用。单元测试的方便之处在于可以直接调用已经写好的代码，或测试写好的代码。注意单元测试的方法名以test为开头，如果不以test开头则不会进行测试。

    from django.test import TestCase
import unittest

class UserTestCase(TestCase):
	def setUp(self):
		print('--执行测试前进行配置--')

	def test_01_add(self):
		print('--add order--')

	def test_02_get_info(self):
		print('--info--')

	def tearDown(self):
		print('--测试后进行释放资源等收尾工作--')

if __name__ == '__main__':
	unittest.main()		# 执行单个测试

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18

    执行顺序为：

    • setUp 用于初始化资源
    • 各自定义方法，不是按照声明顺序，而是按照 ASCII 排序，所以常以 test 加序号加业务名称来声明
    • tearDown 用于回收资源

    单个套件

    除了直接执行测试类，也可以创建套件执行

    from unittest import TestSuite, TextTestRunner

def suite():	# 声明套件类
	suite_ = TestSuite()		# 创建测试套件对象
	suite_.addTest(UserTestCase.test_01_add)		# 添加套件的测试方法
	suite_.addTest(UserTestCase.test_02_get_info)
	
	return suite_

if __name__ == '__main__':
	runner = TextTestRunner()
	runner.run(suite())

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12

    多个套件

    多个套件可以按照排序来进行不同的测试类的测试

    def suite1():		# 套件类1
	suite_ = TestSuite()		# 创建测试套件对象
	suite_.addTest(UserTestCase.test_01_add)		# 添加套件的测试方法
	suite_.addTest(UserTestCase.test_02_get_info)
	
	return suite_

def suite2():		# 套件类2
	suite_ = TestSuite()		# 创建测试套件对象
	suite_.addTest(OrderTestCase.test_01_add)		# 添加套件的测试方法
	suite_.addTest(OrderTestCase.test_02_get_info)
	
if __name__ == '__main__':
	TextTestRunner().run(TestSuite((suite1(), suite2())))		# 按顺序进行测试

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14

    TestCase类的方法

    • assertTrue(boolean condition)
      如果 condition 为 false 则失败；否则通过测试；
    • assertEquals(Object expected, Object actual)
      根据 equals() 方法，如果 expected 和 actual 不相等则失败；否则通过测试；
    • assertEquals(int expected, int actual)
      根据==操作符，如果 expected 和 actual 不相等则失败；否则通过测试。对每一个原始类型：int、float、double、char、byte、long、short和boolean，这个方法都会都一个函数的重载。（参见assertEquals() 的注释）
    • assertSame(Object expected, Object actual)
      如果 expected 和 actual 引用不同的内存对象则失败；如果它们引用相同的内存对象则通过测试。两个对象可能并不是相同的，但是它们可能通过 equals() 方法仍然可以是相等的
    • assertNull(Object object)
      如果对象为null则通过测试，反之看作失败

    测试模型

    django 测试模型不会使用实际数据库， 会为测试创建单独的空白数据库。所以进行模型测试需要先在 setUp() 方法中创建模型数据，进行数据初始化。当测试正常结束后，无论结果如何，会将临时创建的数据库销毁。

    测试接口(视图、单元)

    django 在TestCase对象中定义了Client类用于模拟客户端发送请求，所以可以使用这个工具来测试接口或视图。

    def test_api(self):
	response = self.client.post('/api')
	self.assertEqual(r.status_code, 200)
	content = json.loads(r.content)
	self.assertEqual(content['result'], True)

    • 1
    • 2
    • 3
    • 4
    • 5

    client 可以使用 get 、post、put 等常用发放模拟客户端发送请求，这些方法的格式基本一致：post(path, data, content_type, follow, secure)。其参数含义为：

    • path 发送请求使用url
    • data 发送请求时携带的数据
    • content_type 携带数据的格式
    • secure 客户端将模拟HTTPS请求
    • follow 客户端将遵循任何重定向

    测试 UI (模板)

    线程共用和独立的静态数据

    django 是基于多线程的，每一个请求处理都是一个独立的线程。当定义一个全局的静态数据时(python 没有真正意义上的静态数据，所定义的静态数据也是可以更改的，这里的静态只是相对而言)，所有视图均可以访问此数据。因此静态数据是线程共用的，即不管访问用户和访问请求，均使用同一个数据。

    如果希望有一个数据是线程独立的，例如需要在信号、钩子函数中访问 request 信息，而又不方便显式将 request 作为参数传递的时候。此数据在请求中间件随请求创建，随响应消亡。可以使用 threading 模块的 local() 方法来创建线程数据。

    线程数据

    使用 local() 方法将返回一个本地线程对象，此对象是线程独立的。可以将数据保存在此对象的属性中，在需要时调取。

    # 自定义的中间件文件 middleware.py
from django.utils.deprecation import MiddlewareMixin
from threading import local

_locals = local()

# getter函数
def get_current_request():
	return getattr(_locals,'request',None)

# 处理请求时添加数据
class RequestMiddlewart(MiddlewareMixin):
	@staticmethod
	def process_request(request):
		_locals.request = request

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
  • 相关阅读:
    【IVI】15.1.1 系统稳定性优化篇（LMKD）Android低内存查杀守护进程（Android12）
    反射的作用（ 越过泛型检查 和 可以使用反射保存所有对象的具体信息 ）
    Vue定义全局组件的三种方式
    帮助团队成长是唯一的出路
    docker生成ssl证书（按步骤来即可，真实可用）
    跨考408的C语言需要什么水平?
    每天几道面试题｜Kafka基础概念(一)
    重保主题公开课举办，实战专家分享能源行业安全防护的破局之道
    Spring源码笔记之SpringIOC--（2）从BeanDefinition到Bean实例
    Java8 到 Java17 升级指南(Bug大全)
  • 原文地址：https://blog.csdn.net/runsong911/article/details/127936167