33 lines
1.1 KiB
Python
33 lines
1.1 KiB
Python
|
from django.core.handlers.wsgi import WSGIRequest
|
||
|
|
from rest_framework import status
|
||
|
|
from rest_framework.decorators import action
|
||
|
|
from rest_framework.request import Request
|
||
|
|
from rest_framework.response import Response
|
||
|
|
from rest_framework.routers import DefaultRouter
|
||
|
|
from rest_framework.viewsets import ViewSet
|
||
|
|
|
||
|
|
from django_celery.xiao_queue队列.tasks import taskMessages
|
||
|
|
from sixGuoDjango import BaseResponse, ResData
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
class Celery测试接口ViewSets( ViewSet ):
|
||
|
|
|
||
|
|
|
||
|
|
@action( detail=False, methods=[ "get", "post", "put", "delete" ], url_path="test" )
|
||
|
|
def 测试路由接口( self, request: WSGIRequest | Request ):
|
||
|
|
""""""
|
||
|
|
taskMessages.delay( token="alkdsfjal;d" )
|
||
|
|
taskMessages.apply_async( kwargs={ 'token': 'your_token' }, )
|
||
|
|
httpResData = BaseResponse( resData=ResData( listData=[ "" ], dictData={ "demo": "xiaozizi" } ), ).model_dump( )
|
||
|
|
return Response( data=httpResData, status=status.HTTP_200_OK )
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
Celery测试接口ViewSetsRouter = DefaultRouter( )
|
||
|
|
Celery测试接口ViewSetsRouter.register( prefix="", viewset=Celery测试接口ViewSets, basename="" )
|