使用channels在WebSockets之上构建RESTful API。
项目描述
Channels API
Channels API通过channels在WebSockets上公开RESTful流式API。它提供了一个类似于Django Rest Framework的ModelViewSet的ResourceBinding。它基于DRF序列化类。
它需要Python 2.7或3.x,Channels <=1.1.8.1,Django <=1.11,以及Django Rest Framework 3.x
您可以从我在SF Django Meetup或PyBay 2016的演讲中了解更多关于channels-api的信息
目录
它是如何工作的?
API建立在channels的WebsocketBinding类之上。它通过客户端发送stream和payload参数来工作。这使得我们可以将消息路由到不同的流(或资源)以执行特定操作。因此,POST /user将有一个如下所示的消息
var msg = {
stream: "users",
payload: {
action: "create",
data: {
email: "test@example.com",
password: "password"
}
}
}
ws.send(JSON.stringify(msg))
为什么?
您已经在使用Django Rest Framework,并希望在WebSockets上公开类似的逻辑。
WebSockets可以在不请求的情况下向客户端发布更新。这对于资源可以在多个平台上被多个用户编辑的情况非常有用。
入门
本教程假定您熟悉channels并已完成入门
将channels_api添加到requirements.txt
pip install channels_api
将channels_api添加到INSTALLED_APPS
INSTALLED_APPS = (
'rest_framework',
'channels',
'channels_api'
)
添加您的第一个资源绑定
# polls/bindings.py
from channels_api.bindings import ResourceBinding
from .models import Question
from .serializers import QuestionSerializer
class QuestionBinding(ResourceBinding):
model = Question
stream = "questions"
serializer_class = QuestionSerializer
queryset = Question.objects.all()
将WebsocketDemultiplexer添加到您的channel_routing
# proj/routing.py
from channels.generic.websockets import WebsocketDemultiplexer
from channels.routing import route_class
from polls.bindings import QuestionBinding
class APIDemultiplexer(WebsocketDemultiplexer):
consumers = {
'questions': QuestionBinding.consumer
}
channel_routing = [
route_class(APIDemultiplexer)
]
这就完成了。您现在可以向服务器发送REST WebSocket请求。
var ws = new WebSocket("ws://" + window.location.host + "/")
ws.onmessage = function(e){
console.log(e.data)
}
var msg = {
stream: "questions",
payload: {
action: "create",
data: {
question_text: "What is your favorite python package?"
},
request_id: "some-guid"
}
}
ws.send(JSON.stringify(msg))
// response
{
stream: "questions",
payload: {
action: "create",
data: {
id: "1",
question_text: "What is your favorite python package"
}
errors: [],
response_status: 200
request_id: "some-guid"
}
}
添加channels调试页面(可选)
本页面有助于调试浏览器中的API请求并查看响应。它仅在DEBUG=TRUE时设计使用。
# proj/urls.py
from django.conf.urls import include
urlpatterns = [
url(r'^channels-api/', include('channels_api.urls'))
]
ResourceBinding
默认情况下,ResourceBinding实现了以下REST方法:
创建
检索
更新
列出
删除
订阅
请查看测试套件以获取每个方法的用法示例。
列表分页
分页由django.core.paginator.Paginator处理。
您可以通过覆盖设置来配置DEFAULT_PAGE_SIZE。
# settings.py
CHANNELS_API = {
'DEFAULT_PAGE_SIZE': 25
}
订阅
订阅是一种在资源被创建、更新或删除时从服务器程序化接收更新的方式。
默认情况下,channels-api实现了以下订阅:
创建资源
更新任何资源
更新此资源
删除任何资源
删除此资源
要订阅特定事件,请使用带有过滤参数的subscribe动作。
// get an event when any question is updated
var msg = {
stream: "questions",
payload: {
action: "subscribe",
data: {
action: "update"
}
}
}
// get an event when question(1) is updated
var msg = {
stream: "questions",
payload: {
action: "subscribe",
pk: "1",
data: {
action: "update"
}
}
}
自定义操作
要添加自定义动作,请使用detail_action或list_action装饰器。
from channels_api.bindings import ResourceBinding
from channels_api.decorators import detail_action, list_action
from .models import Question
from .serializers import QuestionSerializer
class QuestionBinding(ResourceBinding):
model = Question
stream = "questions"
serializer_class = QuestionSerializer
queryset = Question.objects.all()
@detail_action()
def publish(self, pk, data=None, **kwargs):
instance = self.get_object(pk)
result = instance.publish()
return result, 200
@list_action()
def report(self, data=None, **kwargs):
report = self.get_queryset().build_report()
return report, 200
然后在消息中将方法名称作为“action”传递。
// run the publish() custom action on Question 1
var msg = {
stream: "questions",
payload: {
action: "publish",
data: {
pk: "1"
}
}
}
// run the report() custom action on all Questions
var msg = {
stream: "questions",
payload: {
action: "report"
}
}
权限
Channels API提供了一种简单的权限类系统,灵感来自rest_framework。提供了两个权限类:AllowAny和IsAuthenticated。
要全局配置权限,请使用设置DEFAULT_PERMISSION_CLASSES如下所示:
# settings.py
CHANNELS_API = {
'DEFAULT_PERMISSION_CLASSES': ('channels_api.permissions.AllowAny',)
}
您还可以像这样在ResourceBinding本身上配置权限类:
from channels_api.permissions import IsAuthenticated
class MyBinding(ResourceBinding):
permission_classes = (IsAuthenticated,)
最后,要实现您自己的权限类,请覆盖BasePermission的has_permission。
from channels_api.permissions import BasePermission
class MyPermission(BasePermission):
def has_permission(self, user, action, pk):
if action == "CREATE":
return True
return False
项目详情
下载文件
下载您平台的文件。如果您不确定选择哪个,请了解更多关于安装包的信息。
源分布
构建分布
channels_api-0.4.1-py3-none-any.whl的哈希值
算法 | 哈希摘要 | |
---|---|---|
SHA256 | a59ea7756993483740254e3cb94d52a4f555d60c745fd9f6bacfec349168c397 |
|
MD5 | 52b4ca1920e043efac3a667a54a76c8c |
|
BLAKE2b-256 | 328d6bd99c04e882f4f02916d61260c79dc0a5a0224fbdb3c1e5e8a53890d29a |