我想将 django 与 aiohttp/asyncio 集成以进行异步编程和 websockets 处理。我知道 django 有 celery 和 django-channels 来分别执行异步任务和 websocket 服务器，但是 aiohttp 预先内置了异步和 websocket 服务器，我发现该框架在创建功能时比 celery/django 通道更具可扩展性和简单性到 webscraping（我不知道是否可以在 celery I 中进行 webscraping，还没有尝试过）。
并且它也完美支持异步和等待。
但我的问题是：我们如何在一个项目中同时实现 django 和 aiohttp？我们可以使用 aiohttp 服务器来代替使用 django 的开发服务器来为站点提供服务。
我们是否能够将 django 与 aiohttp 功能集成（例如让我们举个例子：如果我想将用户提交的输入的网站抓取到我的数据库中。我可以在获取网站并将以下网站发布到我的函数中时使用 await 调用吗？我的 django 数据库？或者将函数结果发布到另一个 django 函数？）
我想知道集成的缺点，如果有的话？
在发布您的答案时，请您发布一个示例实际集成示例，而不是通过 github 向我建议这些库。
Following tutorial on https://testdriven.io/, I have created a websocket test for testing my websocket connection:
# tests/test_websockets.py
from django.contrib.auth import get_user_model
from django.contrib.auth.models import Group
from django.test import Client
from channels.db import database_sync_to_async
from channels.layers import get_channel_layer
from channels.testing import WebsocketCommunicator
from nose.tools import assert_equal, assert_is_none, assert_is_not_none, assert_true
import pytest
from dc_wb.routing import application
from posts.models import Trip
TEST_CHANNEL_LAYERS = {
'default': {
'BACKEND': 'channels.layers.InMemoryChannelLayer',
},
}
@database_sync_to_async
def create_user(*,username='rider@example.com',password='pAssw0rd!',group='rider'):
# Create user.
user = get_user_model().objects.create_user(
username=username,
password=password
) …我不知道如何使用令牌身份验证连接和身份验证 django 通道，因为 javascript websockets 不支持将标头传递给服务器
class TokenAuthMiddleware:
"""
Token authorization middleware for Django Channels 2
"""
def __init__(self, inner):
self.inner = inner
def __call__(self, scope):
headers = dict(scope['headers'])
if b'authorization' in headers:
try:
token_name, token_key = headers[b'authorization'].decode().split()
if token_name == 'Token':
token = Token.objects.get(key=token_key)
scope['user'] = token.user
except Token.DoesNotExist:
scope['user'] = AnonymousUser()
return self.inner(scope)
TokenAuthMiddlewareStack = lambda inner: TokenAuthMiddleware(AuthMiddlewareStack(inner))
我找到了有关使用令牌进行身份验证的答案，但问题是我不明白如何将标头传递给服务器
token = Token.objects.get(key='175f76fd9b63a9477bf5f9a6f2e9a7f12ac62d65')
if token.user:
scope['user'] = token.user
else:
scope['user'] = AnonymousUser()
return self.inner(scope)
TokenAuthMiddlewareStack = lambda inner: TokenAuthMiddleware(AuthMiddlewareStack(inner)) …这是我的views.py代码
class DirectView(mixins.CreateModelMixin):
serializer_class=DirectSerializer
def perform_create(self, serializer):
serializer.save(user=self.request.user)
def post(self,request,*args,**kwargs):
return self.create(request,*args,**kwargs)
和我的 urls.py
path('direct/',DirectView.as_view(),name='direct'),
但每当我尝试运行服务器时，我都会收到错误
AttributeError: type object 'DirectView' has no attribute 'as_view'
我不明白问题是什么？