Django第三方功能的使用

news2024/10/6 1:39:15

Django第三方功能的使用

  • Django REST framework
    • 前言
    • 1、Django--Restframework--coreapi版文档
    • BUG:AssertionError: `coreapi` must be installed for schema support.
    • How to run Django with Uvicorn webserver?
    • 2、序列化类 Serializer的使用
    • 模型序列化类 ModelSerializer的使用
    • 序列化的嵌套使用
  • 验证码的生成和使用
  • 站内搜索引擎
    • 步骤(注意:有特殊要求)
    • Celery异步任务和定时任务
      • 异步任务
      • 定时任务

Django REST framework

前言

django-rest-framework官网

PYPI: djangorestframework
Django REST framework API 指南
参考博客

1、Django–Restframework–coreapi版文档

安装包
pip3 install djangorestframework
pip3 install markdown       # Markdown support for the browsable API.
pip3 install django-filter
pip3 install Pygments
pip3 install coreapi
pip3 install PyYAML

项目url下设置文档路由
from rest_framework.documentation import include_docs_urls
urlpatterns = [
    path('admin/', admin.site.urls),
    path('docs/', include_docs_urls(title='My API Title')),
]

项目settings
REST_FRAMEWORK = {
    # Use Django's standard `django.contrib.auth` permissions,
    # or allow read-only access for unauthenticated users.
    # 'DEFAULT_PERMISSION_CLASSES': [
    #     'rest_framework.permissions.DjangoModelPermissionsOrAnonReadOnly'  # 适用于添加身份验证和权限以后。
    # ]
    'DEFAULT_SCHEMA_CLASS': 'rest_framework.schemas.AutoSchema' # (推荐) 因为新版的restframework需要指定默认schema
    # 或者 'DEFAULT_SCHEMA_CLASS': 'rest_framework.schemas.coreapi.AutoSchema'
}

BUG:AssertionError: coreapi must be installed for schema support.

解决办法:主要问题是urllib3的版本,降低版本

pip install urllib3==1.26.15

How to run Django with Uvicorn webserver?

问题:使用Uvicorn 运行的时候显示静态文件丢失
在这里插入图片描述

解决办法:

 settings.py 
STATIC_ROOT = os.path.join(BASE_DIR, 'static', )


项目 urls.py
from django.conf.urls.static import static
from django.conf import settings

urlpatterns = [.
.....] + static(settings.STATIC_URL, document_root=settings.STATIC_ROOT)
Then run below command but static directory must exist

python manage.py collectstatic --noinput   # 收集静态文件
--noinput 参数的作用是执行收集静态文件的命令时不会询问任何输入,一般用于自动化脚本或者不需要交互式输入的场景
启动uvicorn
uvicorn your_project.asgi:application --reload --host 0.0.0.0 --port 8000

2、序列化类 Serializer的使用

serializers.py
import asyncio

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


class UserSerializer(serializers.HyperlinkedModelSerializer):
    class Meta:
        model = User
        fields = ['url', 'username', 'email', 'groups']


class GroupSerializer(serializers.HyperlinkedModelSerializer):
    class Meta:
        model = Group
        fields = ['url', 'name']

from .models import PersonInfo, Vocation

nameList = PersonInfo.objects.values('name').all()
NAME_CHOICES = [item['name'] for item in nameList]

class MySerializer(serializers.Serializer):
    id = serializers.IntegerField(read_only=True)
    job = serializers.CharField(max_length=20)
    title = serializers.CharField(max_length=20)
    payment = serializers.IntegerField()
    name = serializers.PrimaryKeyRelatedField(queryset=NAME_CHOICES)  # models的外键字段

    def create(self, validated_data):
        return Vocation.objects.create(**validated_data)

    def update(self, instance, validated_data):
        return instance.update(**validated_data)
views.py
from django.contrib.auth.models import Group, User
from rest_framework import permissions, viewsets

from .serializers import GroupSerializer, UserSerializer


class UserViewSet(viewsets.ModelViewSet):
    """
    API endpoint that allows users to be viewed or edited.
    """
    queryset = User.objects.all().order_by('-date_joined')
    serializer_class = UserSerializer
    # permission_classes = [permissions.IsAuthenticated]


class GroupViewSet(viewsets.ModelViewSet):
    """
    API endpoint that allows groups to be viewed or edited.
    """
    queryset = Group.objects.all().order_by('name')
    serializer_class = GroupSerializer
    # permission_classes = [permissions.IsAuthenticated]

from channels.db import database_sync_to_async
from .models import PersonInfo, Vocation
from .serializers import MySerializer
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import status
from rest_framework.pagination import PageNumberPagination
from rest_framework.decorators import api_view


@api_view(['GET', 'POST'])
def vocationDef(request):
    if request.method == 'GET':
        q = Vocation.objects.all()
        pg = PageNumberPagination()
        p = pg.paginate_queryset(queryset=q, request=request)
        serializer = MySerializer(instance=p, many=True)
        return Response(serializer.data)
    elif request.method == 'POST':
        data = request.data
        id = data['name']
        data['name'] = PersonInfo.objects.filter(id=id).first()
        instance = Vocation.objects.filter(id=data.get('id', 0))
        if instance:
            MySerializer().update(instance, data)
        else:
            MySerializer().create(data)
        return Response('Done', status=status.HTTP_201_CREATED)


class VocationClass(APIView):
    def get(self, request):
        q = Vocation.objects.all()
        pg = PageNumberPagination()
        p = pg.paginate_queryset(queryset=q, request=request, view=self)
        serializer = MySerializer(instance=p, many=True)
        return Response(serializer.data)

    def post(self, request):
        data = request.data
        id = data['name']
        data['name'] = PersonInfo.objects.filter(id=id).first()
        instance = Vocation.objects.filter(id=data.get('id', 0))
        if instance:
            MySerializer().update(instance, data)
        else:
            MySerializer().create(data)
        return Response('Done', status=status.HTTP_201_CREATED)

模型序列化类 ModelSerializer的使用

serializers.py
from rest_framework import serializers
class VocationSerializer(serializers.ModelSerializer):
    class Meta:
        model = Vocation
        fields = ('id', 'job', 'title', 'payment', 'name')
views.py
from django.contrib.auth.models import Group, User
from rest_framework import permissions, viewsets

from .serializers import GroupSerializer, UserSerializer


class UserViewSet(viewsets.ModelViewSet):
    """
    API endpoint that allows users to be viewed or edited.
    """
    queryset = User.objects.all().order_by('-date_joined')
    serializer_class = UserSerializer
    # permission_classes = [permissions.IsAuthenticated]


class GroupViewSet(viewsets.ModelViewSet):
    """
    API endpoint that allows groups to be viewed or edited.
    """
    queryset = Group.objects.all().order_by('name')
    serializer_class = GroupSerializer
    # permission_classes = [permissions.IsAuthenticated]

from channels.db import database_sync_to_async
from .models import PersonInfo, Vocation
from .serializers import MySerializer, VocationSerializer
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import status
from rest_framework.pagination import PageNumberPagination
from rest_framework.decorators import api_view


@api_view(['GET', 'POST'])
def vocationDef(request):
    if request.method == 'GET':
        q = Vocation.objects.all()
        pg = PageNumberPagination()
        p = pg.paginate_queryset(queryset=q, request=request)
        serializer = VocationSerializer(instance=p, many=True)
        return Response(serializer.data)
    elif request.method == 'POST':
        id = request.data.get('id', 0)
        operation = Vocation.objects.filter(id=id).first()
        serializer = VocationSerializer(data=request.data)
        if serializer.is_valid():
            if operation:
                data = request.data
                id = data['name']
                data['name'] = PersonInfo.objects.filter(id=id).first()
                serializer.update(operation, data)
            else:
                serializer.save()
            return Response(serializer.data)
        return Response(serializer.errors, status=status.HTTP_404_NOT_FOUND)



class VocationClass(APIView):
    def get(self, request):
        q = Vocation.objects.all()
        pg = PageNumberPagination()
        p = pg.paginate_queryset(queryset=q, request=request, view=self)
        serializer = VocationSerializer(instance=p, many=True)
        return Response(serializer.data)

    def post(self, request):
        id = request.data.get('id', 0)
        operation = Vocation.objects.filter(id=id).first()
        serializer = VocationSerializer(data=request.data)
        if serializer.is_valid():
            if operation:
                data = request.data
                id = data['name']
                data['name'] = PersonInfo.objects.filter(id=id).first()
                serializer.update(operation, data)
            else:
                serializer.save()
            return Response(serializer.data)
        return Response(serializer.errors, status=status.HTTP_404_NOT_FOUND)

序列化的嵌套使用

模型之间存在数据关系才能进行数据嵌套

class PersonInfoSerializer(serializers.ModelSerializer):
    class Meta:
        model = PersonInfo
        fields = '__all__'

class VocationSerializer(serializers.ModelSerializer):
    name = PersonInfoSerializer()
    class Meta:
        model = Vocation
        fields = ('id', 'job', 'title', 'payment', 'name')

    def create(self, validated_data):
        print('vad', validated_data)
        name = validated_data.get('name', '')
        id = name.get('id', 0)
        p = PersonInfo.objects.filter(id=id).first()
        if not p:
            p = PersonInfo.objects.create(**name)
        data = validated_data
        data['name'] = p
        v = Vocation.objects.create(**data)
        return v

    def update(self, instance, validated_data):
        print('vad', validated_data)
        name = validated_data.get('name', '')
        id = name.get('id', 0)
        p = PersonInfo.objects.filter(id=id).first()
        if p:
            PersonInfo.objects.filter(id=id).update(**name)
            data = validated_data
            data['name'] = p
            id = validated_data.get('id', '')
            v = Vocation.objects.filter(id=id).update(**data)
            return v
from django.contrib.auth.models import Group, User
from rest_framework import permissions, viewsets

from .serializers import GroupSerializer, UserSerializer


class UserViewSet(viewsets.ModelViewSet):
    """
    API endpoint that allows users to be viewed or edited.
    """
    queryset = User.objects.all().order_by('-date_joined')
    serializer_class = UserSerializer
    # permission_classes = [permissions.IsAuthenticated]


class GroupViewSet(viewsets.ModelViewSet):
    """
    API endpoint that allows groups to be viewed or edited.
    """
    queryset = Group.objects.all().order_by('name')
    serializer_class = GroupSerializer
    # permission_classes = [permissions.IsAuthenticated]

from channels.db import database_sync_to_async
from .models import PersonInfo, Vocation
from .serializers import MySerializer, VocationSerializer
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import status
from rest_framework.pagination import PageNumberPagination
from rest_framework.decorators import api_view


@api_view(['GET', 'POST'])
def vocationDef(request):
    if request.method == 'GET':
        q = Vocation.objects.all()
        pg = PageNumberPagination()
        p = pg.paginate_queryset(queryset=q, request=request)
        serializer = VocationSerializer(instance=p, many=True)
        return Response(serializer.data)
    elif request.method == 'POST':
        id = request.data.get('id', 0)
        operation = Vocation.objects.filter(id=id).first()
        serializer = VocationSerializer(data=request.data)
        if serializer.is_valid():
            if operation:
                serializer.update(operation, request.data)
            else:
                serializer.save()
            return Response(serializer.data)
        return Response(serializer.errors, status=status.HTTP_404_NOT_FOUND)



class VocationClass(APIView):
    def get(self, request):
        q = Vocation.objects.all()
        pg = PageNumberPagination()
        p = pg.paginate_queryset(queryset=q, request=request, view=self)
        serializer = VocationSerializer(instance=p, many=True)
        return Response(serializer.data)

    def post(self, request):
        id = request.data.get('id', 0)
        operation = Vocation.objects.filter(id=id).first()
        serializer = VocationSerializer(data=request.data)
        if serializer.is_valid():
            if operation:
                serializer.update(operation, request.data)
            else:
                serializer.save()
            return Response(serializer.data)
        return Response(serializer.errors, status=status.HTTP_404_NOT_FOUND)

验证码的生成和使用

PYPI:django-simple-captcha

pip3 install django-simple-captcha==0.5.20
settings.py
INSTALLED_APPS = [
   'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',

    'rest_framework',
    # 添加验证码功能
    'captcha',
]

# Django Simple Captcha的基本配置
# 设置验证码的显示顺序
# 一个验证码识别包含文本输入框、隐藏域和验证码图片
# CAPTCHA_OUTPUT_FORMAT是设置三者的显示顺序
CAPTCHA_OUTPUT_FORMAT = '%(text_field)s %(hidden_field)s %(image)s'
# 设置图片噪点
CAPTCHA_NOISE_FUNCTIONS = ( # 设置样式
                            'captcha.helpers.noise_null',
                            # 设置干扰线
                           'captcha.helpers.noise_arcs',
                            # 设置干扰点
                           'captcha.helpers.noise_dots',
                           )
# 图片大小
CAPTCHA_IMAGE_SIZE = (100, 25)
# 设置图片背景颜色
CAPTCHA_BACKGROUND_COLOR = '#ffffff'
# 图片中的文字为随机英文字母
# CAPTCHA_CHALLENGE_FUNCT = 'captcha.helpers.random_char_challenge'
# 图片中的文字为英文单词
# CAPTCHA_CHALLENGE_FUNCT = 'captcha.helpers.word_challenge'
# 图片中的文字为数字表达式
CAPTCHA_CHALLENGE_FUNCT = 'captcha.helpers.math_challenge'
# 设置字符个数
CAPTCHA_LENGTH = 4
# 设置超时(minutes)
CAPTCHA_TIMEOUT = 1
生成数据表
python manage.py migrate
forms.py
from django import forms
from captcha.fields import CaptchaField

class CaptchaTestForm(forms.Form):
    username = forms.CharField(label='用户名')
    password = forms.CharField(label='密码', widget=forms.PasswordInput)
    captcha = CaptchaField()
项目urls
from django.contrib import admin
from django.urls import path, include
from django.conf.urls.static import static
from django.conf import settings
from rest_framework.documentation import include_docs_urls

urlpatterns = [
    path('admin/', admin.site.urls),
    path('docs/', include_docs_urls(title='My API Title')),
    path('api-auth/', include('rest_framework.urls')),
    path('api/', include('apiwx.urls')),
    path('captcha/', include('captcha.urls')),   # 添加验证码路由
] + static(settings.STATIC_URL, document_root=settings.STATIC_ROOT)
应用urls
from django.urls import include, path
from rest_framework import routers

from . import views

router = routers.DefaultRouter()
router.register(r'users', views.UserViewSet)
router.register(r'groups', views.GroupViewSet)

# Wire up our API using automatic URL routing.
# Additionally, we include login URLs for the browsable API.
urlpatterns = [
    path('', include(router.urls)),
    path('func/', views.vocationDef),
    path('class/', views.VocationClass.as_view()),
    path('login/', views.loginView, name='login'),
    path('ajax_val/', views.ajax_val, name='ajax_val'),
]
views.pyfrom django.shortcuts import render
from django.contrib.auth.models import User
from django.contrib.auth import login, authenticate
from .forms import CaptchaTestForm
# 用户登录
def loginView(request):
    if request.method == 'POST':
        form = CaptchaTestForm(request.POST)
        # 验证表单数据
        if form.is_valid():
            u = form.cleaned_data['username']
            p = form.cleaned_data['password']
            if User.objects.filter(username=u):
                user = authenticate(username=u, password=p)
                if user:
                    if user.is_active:
                        login(request, user)
                        tips = '登录成功'

                else:
                    tips = '账号密码错误,请重新输入'
            else:
                tips = '用户不存在,请注册'
    else:
        form = CaptchaTestForm()
    return render(request, 'user.html', locals())

# ajax接口,实现动态验证验证码
from django.http import JsonResponse
from captcha.models import CaptchaStore
def ajax_val(request):
    if request.is_ajax():
        # 用户输入的验证码结果
        r = request.GET['response']
        # 隐藏域的value值
        h = request.GET['hashkey']
        cs = CaptchaStore.objects.filter(response=r, hashkey=h)
        # 若存在cs,则验证成功,否则验证失败
        if cs:
            json_data = {'status':1}
        else:
            json_data = {'status':0}
        return JsonResponse(json_data)
    else:
        json_data = {'status':0}
        return JsonResponse(json_data)
html
<!DOCTYPE html>
<html lang="en">
    <head>
        <meta charset="UTF-8" />
        <title>Django</title>
        <script src="http://apps.bdimg.com/libs/jquery/2.1.1/jquery.min.js"></script>
        <link rel="stylesheet" href="https://unpkg.com/mobi.css/dist/mobi.min.css">
    </head>
    <body>
        <div class="flex-center">
            <div class="container">
            <div class="flex-center">
            <div class="unit-1-2 unit-1-on-mobile">
                <h1>MyDjango Verification</h1>
                    {% if tips %}
                <div>{{ tips }}</div>
                    {% endif %}
                <form class="form" action="" method="post">
                    {% csrf_token %}
                    <div>用户名:{{ form.username }}</div>
                    <div>密 码:{{ form.password }}</div>
                    <div>验证码:{{ form.captcha }}</div>
                    <button type="submit" class="btn btn-primary btn-block">确定</button>
                </form>
            </div>
            </div>
            </div>
        </div>
    <script>
        $(function(){
            $('.captcha').click(function(){
                console.log('click');
                    $.getJSON("/captcha/refresh/",
                function(result){
                    $('.captcha').attr('src', result['image_url']);
                    $('#id_captcha_0').val(result['key'])
            });});
            $('#id_captcha_1').blur(function(){
                json_data={
                    'response':$('#id_captcha_1').val(),
                    'hashkey':$('#id_captcha_0').val()
                }
                $.getJSON('/ajax_val', json_data, function(data){
                    $('#captcha_status').remove()
                    if(data['status']){
                        $('#id_captcha_1').after('<span id="captcha_status">*验证码正确</span>')
                    }else{
                        $('#id_captcha_1').after('<span id="captcha_status">*验证码错误</span>')
                    }
                });
            });
        })
    </script>
    </body>
</html>

站内搜索引擎

django-haystack 是专门提供搜索功能的DJango第三方应用,支持solr,elasticserch,whoosh,xapian多种搜索引擎,配合中文自然语言处理库jieba分词,可以实现全文搜索系统

pip3 install django-haystack
pip3 install whoosh
pip3 install jieba

步骤(注意:有特殊要求)

  • 在项目应用中添加 search_indexes.py和whoosh_cn_backend.py
  • 在项目的根目录创建文件夹 static和templates,static存放CSS样式文件,templates存放search.html和搜索引擎文件product_text.txt(文件的命名方式有具体的要求,下面会说明)
  1. search_indexes.py : 定义模型的索引类,使模型的数据能被搜索引擎搜索
  2. whoosh_cn_backend.py:自定义搜索引擎文件,由于Whoosh不支持中文搜索,重新定义搜索引擎文件,将jieba分词器添加到搜索引擎中,使其具有中文搜索功能
  3. product_text.txt:搜索引擎的索引模板文件,模板文件命名以及路径有固定格式,如:templates/search/indexes/项目应用的名称/模型名称(小写)_text.txt
INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',

    'rest_framework',
    # 添加验证码功能
    'captcha',
    # 配置haystack
    'haystack',
]

# 配置haystack
HAYSTACK_CONNECTIONS = {
    'default': {
        # 设置搜索引擎,文件是apiwx(应用)的whoosh_cn_backend.py
        'ENGINE': 'apiwx.whoosh_cn_backend.WhooshEngine',
        'PATH': str(BASE_DIR / 'whoosh_index'),
        'INCLUDE_SPELLING': True,
    },
}
# 设置每页显示的数据量
HAYSTACK_SEARCH_RESULTS_PER_PAGE = 4
# 当数据库改变时,会自动更新索引,非常方便
HAYSTACK_SIGNAL_PROCESSOR = 'haystack.signals.RealtimeSignalProcessor'
whoosh_cn_backend.py
# encoding: utf-8
# 文件来自haystack包,路径为Python\Lib\site-packages\haystack\backends\whoosh_backend.py
# 导入from jieba.analyse import ChineseAnalyzer包,添加中文搜索功能
# 将schema_fields[field_class.index_fieldname] = TEXT....的内容修改为:schema_fields[field_class.index_fieldname] = TEXT(stored=True, analyzer=ChineseAnalyzer(),field_boost=field_class.boost, sortable=True)


from haystack.backends.whoosh_backend import *
from jieba.analyse import ChineseAnalyzer

class MyWhooshSearchBackend(WhooshSearchBackend):
    def build_schema(self, fields):
        schema_fields = {
            ID: WHOOSH_ID(stored=True, unique=True),
            DJANGO_CT: WHOOSH_ID(stored=True),
            DJANGO_ID: WHOOSH_ID(stored=True),
        }
        # Grab the number of keys that are hard-coded into Haystack.
        # We'll use this to (possibly) fail slightly more gracefully later.
        initial_key_count = len(schema_fields)
        content_field_name = ''

        for field_name, field_class in fields.items():
            if field_class.is_multivalued:
                if field_class.indexed is False:
                    schema_fields[field_class.index_fieldname] = IDLIST(stored=True, field_boost=field_class.boost)
                else:
                    schema_fields[field_class.index_fieldname] = KEYWORD(stored=True, commas=True, scorable=True, field_boost=field_class.boost)
            elif field_class.field_type in ['date', 'datetime']:
                schema_fields[field_class.index_fieldname] = DATETIME(stored=field_class.stored, sortable=True)
            elif field_class.field_type == 'integer':
                schema_fields[field_class.index_fieldname] = NUMERIC(stored=field_class.stored, numtype=int, field_boost=field_class.boost)
            elif field_class.field_type == 'float':
                schema_fields[field_class.index_fieldname] = NUMERIC(stored=field_class.stored, numtype=float, field_boost=field_class.boost)
            elif field_class.field_type == 'boolean':
                # Field boost isn't supported on BOOLEAN as of 1.8.2.
                schema_fields[field_class.index_fieldname] = BOOLEAN(stored=field_class.stored)
            elif field_class.field_type == 'ngram':
                schema_fields[field_class.index_fieldname] = NGRAM(minsize=3, maxsize=15, stored=field_class.stored, field_boost=field_class.boost)
            elif field_class.field_type == 'edge_ngram':
                schema_fields[field_class.index_fieldname] = NGRAMWORDS(minsize=2, maxsize=15, at='start', stored=field_class.stored, field_boost=field_class.boost)
            else:
                schema_fields[field_class.index_fieldname] = TEXT(stored=True, analyzer=ChineseAnalyzer(),field_boost=field_class.boost, sortable=True)

            if field_class.document is True:
                content_field_name = field_class.index_fieldname
                schema_fields[field_class.index_fieldname].spelling = True

        # Fail more gracefully than relying on the backend to die if no fields
        # are found.
        if len(schema_fields) <= initial_key_count:
            raise SearchBackendError("No fields were found in any search_indexes. Please correct this before attempting to search.")

        return (content_field_name, Schema(**schema_fields))

# 重新定义搜索引擎
class WhooshEngine(BaseEngine):
    # 将搜索引擎指向自定义的MyWhooshSearchBackend
    backend = MyWhooshSearchBackend
    query = WhooshSearchQuery
models.py
class Product(models.Model):
    id = models.AutoField('序号', primary_key=True)
    name = models.CharField('名称', max_length=50)
    weight = models.CharField('重量', max_length=20)
    describe = models.CharField('描述', max_length=500)

    # 设置返回值
    def __str__(self):
        return self.name
search_indexes.py
from haystack import indexes
from .models import Product
# 类名必须为模型名+Index
# 比如模型Product,则索引类为ProductIndex
class ProductIndex(indexes.SearchIndex, indexes.Indexable):
    text = indexes.CharField(document=True, use_template=True)
    # 设置模型
    def get_model(self):
        return Product
    # 设置查询范围
    def index_queryset(self, using=None):
        return self.get_model().objects.all()
# 定义索引类的文件名必须是 search_indexes.py, 不得修改文件名
# 模型的索引类的类名格式必须为 “模型名+Index”, 每个模型对应一个索引类,如果模型为Product,则对应的索引类为ProductIndex
# 字段text 设置document=True,表示搜索引擎以此字段的内容作为索引
# use_template=True 表示使用索引模板文件,可以理解为在模板中设置模型的查询字段,如设置Product的name字段,就可以通过name字段检索Product数据
# 类函数get_model是将索引类和模型进行绑定,index_queryset用于设置索引的查询范围
templates/search/indexes/项目应用的名称/模型名称(小写)_text.txt
索引模板文件

{{ object.name }}
{{ object.describe }}

# 对模型的name和describe 字段建立索引,当搜索引擎进行搜索的时候,Django根搜索条件对这两个字段进行全文搜索匹配,然后将结果排序返回

python manage.py rebuild_index   创建索引文件

在这里插入图片描述

views.py
from django.core.paginator import *
from django.shortcuts import render
from django.conf import settings
from .models import *
from haystack.generic_views import SearchView
# 视图以通用视图实现
class MySearchView(SearchView):
    # 模版文件
    template_name = 'search.html'
    def get(self, request, *args, **kwargs):
        if not self.request.GET.get('q', ''):
            product = Product.objects.all().order_by('id')
            per = settings.HAYSTACK_SEARCH_RESULTS_PER_PAGE
            p = Paginator(product, per)
            try:
                num = int(self.request.GET.get('page', 1))
                page_obj = p.page(num)
            except PageNotAnInteger:
                # 如果参数page不是整型,则返回第1页数据
                page_obj = p.page(1)
            except EmptyPage:
                # 访问页数大于总页数,则返回最后1页的数据
                page_obj = p.page(p.num_pages)
            return render(request, self.template_name, locals())
        else:
            return super().get(*args, request, *args, **kwargs)

Celery异步任务和定时任务

pip3 install celery   安装Celery框架实现异步任务和定时任务的调度控制
pip3 install redis  实现python和redis数据库的连接
pip3 install django-celery-results  基于Celery封装的异步任务功能
pip3 install django-celery-beat 基于Celery封装的定时任务功能
pip3 install eventlet   python的协程并发库,这是celery实现异步并发运行的模式之一
settings 中配置异步功能
INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',

    # 添加异步任务功能
    'django_celery_results',
    # 添加定时任务功能
    'django_celery_beat',
]


# 设置存储Celery任务队列的Redis数据库
CELERY_BROKER_URL = 'redis://127.0.0.1:6379/0'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
# 设置存储Celery任务结果的数据库
CELERY_RESULT_BACKEND = 'django-db'

# 设置定时任务相关配置
CELERY_ENABLE_UTC = False
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'
modles.py 
from django.db import models

# Create your models here.
class PersonInfo(models.Model):
    name = models.CharField(max_length=20)
    age = models.IntegerField()
    hireDate = models.DateField()

    def __str__(self):
        return self.name
数据迁移
python manage.py makemigrations
python manage.py migrate
项目下创建celery.py  (settings同目录)
创建celery框架的实例化对象
import os
from celery import Celery
# 获取settings的配置信息
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "web2.settings")
# 定义celery对象,并将项目配置信息加载到对象中
# Celery的参数一般以项目名命名
app = Celery('web2')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

项目下的 __init__.py
将celery实例化对象和django 绑定
# django运行的时候自动加载celery实例化对象
from .celery import app as celery_app
__all__ = ['celery_app']

异步任务

应用下创建task.py
开发异步任务
from celery import shared_task
from .models import PersonInfo
import time

# 带参数的异步任务
@shared_task
def updateDate(id, kwargs):
    try:
        PersonInfo.objects.filter(id=id).update(**kwargs)
        return "Done"
    except Exception as e:
        print('error', e)
        return 'Fail'
开发视图,并在urls中添加路由地址
views.py
from django.http import HttpResponse
from .task import updateDate
def tasksyn(request):
    id = request.GET.get('id', 1)
    kwarg = dict(name='mike', age=19, hireDate='2024-04-13')
    updateDate.delay(id, kwarg)
    return HttpResponse('hello celery')
先启动django
uvicorn web2(项目名).asgi:application --reload --host 0.0.0.0 --port 8000
再启动异步任务
celery -A projectName worker -l info -P eventlet

定时任务

from celery import shared_task
from .models import PersonInfo
import time

# 带参数的异步任务
@shared_task
def updateDate(id, kwargs):
    try:
        PersonInfo.objects.filter(id=id).update(**kwargs)
        return "Done"
    except Exception as e:
        print('error', e)
        return 'Fail'

# 定时任务
@shared_task
def timing():
    now = time.strftime("%H:%M:%S")
    with open('output.txt', 'a') as f:
        f.write('The time is ' + now)
        f.write('\n')

在这里插入图片描述
进入后台,设置定时任务,
Name:给定时任务取名,任意
Task(registered):task.py 开发的定时任务
Interval Schedule:设置时间间隔
PS:如果任务带参数,可在Add periodic task 中设置Arguments 或者 Keyword argument

先启动django
uvicorn web2(项目名).asgi:application --reload --host 0.0.0.0 --port 8000
再启动异步任务
celery -A projectName worker -l info -P eventlet
再启动定时任务
celery -A projectName beat -l info -S django

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/1595062.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

再写-全景拼接

全景拼接 1. 将读取进行灰度转化&#xff0c;并且输出图像&#xff0c;关键点和计算描述 import cv2 import numpy as np# 将读取进行灰度转化&#xff0c;并且输出图像&#xff0c;关键点和计算描述 image_left cv2.imread("C:\\Users\\HONOR\\Desktop\\image\\pinjie…

001_IoT/物联网通信协议基础: HTTP、Websocket、MQTT、AMQP、COAP、LWM2M一文搞懂

001_IoT/物联网通信协议基础: HTTP、Websocket、MQTT、AMQP、COAP、LWM2M一文搞懂 文章目录 001_IoT/物联网通信协议基础: HTTP、Websocket、MQTT、AMQP、COAP、LWM2M一文搞懂创作背景通信模型ISO/OSI七层模型 和 TCP/IP四层模型网络通信数据包格式&#xff08;Ethernet II&…

【微信小程序——案例——本地生活(列表页面)】

案例——本地生活&#xff08;列表页面&#xff09; 九宫格中实现导航跳转——以汽车服务为案例&#xff08;之后可以全部实现页面跳转——现在先实现一个&#xff09; 在app.json中添加新页面 修改之前的九宫格view改为navitage 效果图&#xff1a; 动态设置标题内容—…

【5G PHY】5G无线链路监测原理简述

博主未授权任何人或组织机构转载博主任何原创文章&#xff0c;感谢各位对原创的支持&#xff01; 博主链接 本人就职于国际知名终端厂商&#xff0c;负责modem芯片研发。 在5G早期负责终端数据业务层、核心网相关的开发工作&#xff0c;目前牵头6G算力网络技术标准研究。 博客…

车载电子电器架构 —— 平行开发策略

车载电子电器架构 —— 平行开发策略 我是穿拖鞋的汉子,魔都中坚持长期主义的汽车电子工程师。 老规矩,分享一段喜欢的文字,避免自己成为高知识低文化的工程师: 屏蔽力是信息过载时代一个人的特殊竞争力,任何消耗你的人和事,多看一眼都是你的不对。非必要不费力证明自己…

架构师系列-搜索引擎ElasticSearch(八)- 集群管理故障恢复

故障转移 集群的master节点会监控集群中的节点状态&#xff0c;如果发现有节点宕机&#xff0c;会立即将宕机节点的分片数据迁移到其它节点&#xff0c;确保数据安全&#xff0c;这个叫做故障转移。 下图中node1是主节点&#xff0c;其他两个节点是从节点 节点故障 此时node1…

【LeetCode】回溯算法类题目详解

所有题目均来自于LeetCode&#xff0c;刷题代码使用的Python3版本 回溯算法 回溯算法是一种搜索的方法&#xff0c;在二叉树总结当中&#xff0c;经常使用到递归去解决相关的问题&#xff0c;在二叉树的所有路径问题中&#xff0c;我们就使用到了回溯算法来找到所有的路径。 …

计算机网络 实验指导 实验17

实验17 配置无线网络实验 1.实验拓扑图 Table PC0 和 Table PC1 最开始可能还会连Access Point0&#xff0c;无影响后面会改 名称接口IP地址网关地址Router0fa0/0210.10.10.1fa0/1220.10.10.2Tablet PC0210.10.10.11Tablet PC1210.10.10.12Wireless互联网220.10.10.2LAN192.16…

CSS-布局

display display 属性是用于控制 布局 的最重要的 CSS 属性。display 属性规定是否/如何显示元素。 每个 HTML 元素都有一个默认的 display 值&#xff0c;具体取决于它的元素类型。大多数元素的默认 display 值为 block 或 inline。 block block&#xff1a;块级元素。块级…

STL--list双向链表

功能 将数据进行链式存储 链表&#xff08;list&#xff09;是一种物理存储单元上非连续的存储结构&#xff0c;数据元素的逻辑顺序是通过链表中的指针链接实现的 链表的组成&#xff1a;链表由一系列结点组成 结点的组成&#xff1a;一个是存储数据元素的数据域&#xff0…

Java应用中文件上传安全性分析与安全实践

✨✨谢谢大家捧场&#xff0c;祝屏幕前的小伙伴们每天都有好运相伴左右&#xff0c;一定要天天开心哦&#xff01;✨✨ &#x1f388;&#x1f388;作者主页&#xff1a; 喔的嘛呀&#x1f388;&#x1f388; 目录 引言 一. 文件上传的风险 二. 使用合适的框架和库 1. Spr…

Tomcat服务器入门介及用postman工具简单接收数据 2024详解

Tomcat服务器 简介 Tomcat是一个开源的Servlet容器&#xff0c;也是一个支持Java Servlet和JSP技术的Web服务器。它由Apache软件基金会开发和维护。Tomcat的主要作用是将Java Servlet和JavaServer Pages&#xff08;JSP&#xff09;等动态网页技术部署到服务器上&#xff0c;…

Linux操作系统中关于用户管理的操作

创建新用户 useradd 【选项】 用户名 在/etc/passwd中以追加的方式在passwd的最后一行添加用户信息。 可以使用命令tail -n 1/etc/passwd查看文件的最后一行内容。 ls /home/首先/home/这是普通用户的家目录&#xff0c; 在/home/下会有一个跟用户名同名的家目录&#xf…

推荐一款基于vim的超可扩展文本编辑器neovim

一、简介 Vim是一个基于流行的Vi编辑器的文本编辑器&#xff0c;最初是在20世纪70年代发布的。Vim代表“改进的Vi”&#xff0c;它拥有广泛的用户基础和广泛的可用插件和扩展。 Neovim是Vim的一个分支&#xff0c;创建于2014年&#xff0c;旨在解决Vim的一些缺点&#xff0c;…

Node.js留言板(超详细注释)

目录结构如下 app.js // 一.引入模块 var http require(http);// 用于创建 HTTP 服务器和处理 HTTP 请求 var fs require(fs);// 用于读取和写入文件 var url require(url);// 用于解析URL// 创建留言数据对象 var msgs [{ name: 牛二, content: "我是妞儿", cr…

Hadoop+Spark大数据技术(微课版)曾国荪、曹洁版思维导图第四次作业 (第4章 HBase分布式DB)

1.简述Hbase的特点及与传统关系数据库的区别 HBase与传统关系数据库的区别 &#xff08;1&#xff09;数据类型 关系数据库具有丰富的数据类型&#xff0c;如字符串型、数值型、日期型、二进制型等。HBase只有字符串数据类型&#xff0c;数据的实际类型都是交由用户自己编写程序…

Spring+SpringMVC的知识总结

一:技术体系架构二:SpringFramework介绍三:Spring loC容器和核心概念3.1 组件和组件管理的概念3.1.1什么是组件:3.1.2:我们的期待3.1.3Spring充当组件管理角色(IOC)3.1.4 Spring优势3.2 Spring Ioc容器和容器实现3.2.1普通和复杂容器3.2.2 SpringIOC的容器介绍3.2.3 Spring IOC…

开源版中文和越南语贷款源码贷款平台下载 小额贷款系统 贷款源码运营版

后台 代理 前端均为vue源码&#xff0c;前端有中文和越南语 前端ui黄色大气&#xff0c;逻辑操作简单&#xff0c;注册可对接国际短信&#xff0c;可不对接 用户注册进去填写资料&#xff0c;后台审批&#xff0c;审批状态可自定义修改文字显示 源码免费下载地址抄笔记 (chaob…

【Vue】新手一步一步安装 vue 语言开发环境

文章目录 1、下载node.js安装包 1、下载node.js安装包 1.打开node.js的官网下载地址&#xff1a;http://nodejs.cn/download/ 选择适合自己系统的安装包&#xff1a;winds、mac 2. 配置node.js和npm环境变量 安装好之后&#xff0c;对npm安装的全局模块所在路径以及缓存所在路…

Linux网络基础 (二) ——(IP、MAC、端口号、TCPUDP协议、网络字节序)

文章目录 IP 地址基本概念源IP地址 & 目的IP地址 MAC 地址基本概念源MAC地址 & 目的MAC地址 端口号基本概念源端口号 & 目的端口号 TCP & UDP 协议基本概念TCP 与 UDP 的抉择 网络字节序大端、小端字节序 &#x1f396; 博主的CSDN主页&#xff1a;Ryan.Alask…