上一篇 带你进入异步Django+Vue的世界 - Didi打车实战(4)
Demo: https://didi-taxi.herokuapp.com/
上一篇，前、后端已经完整支持了Websockets。
接下来，我们来实现创建订单、群发群收、修改订单功能。
Refactoring: Trip返回信息
后台返回Trip信息里，driver/rider是一个primary key，指向User。我们希望能直接看到ForeignKey: driver/rider的详细信息。
[{created: "2019-05-20T10:08:59.950536Z"
driver: null
drop_off_address: "牛首山"
id: "4a25dde1-dd0d-422a-9e5e-706958b65046"
pick_up_address: "总统府"
rider: {id: 5, username: "rider3", first_name: "", last_name: "", group: "rider"}
status: "REQUESTED"
updated: "2019-05-20T10:08:59.950563Z"}, ...]
Serializer添加ReadOnlyTripSerializer，关联UserSerializer即可。
# /backend/api/serializers.py
class ReadOnlyTripSerializer(serializers.ModelSerializer):
driver = UserSerializer(read_only=True)
rider = UserSerializer(read_only=True)
class Meta:
model = Trip
fields = '__all__'
然后修改DRF view, 用户HTTP访问/trip/时的TripView，
#
class TripView(viewsets.ReadOnlyModelViewSet):
lookup_field = 'id'
lookup_url_kwarg = 'trip_id'
permission_classes = (permissions.IsAuthenticated,)
queryset = Trip.objects.all()
serializer_class = ReadOnlyTripSerializer # changed
Channels 创建订单
当用户创建一个订单时，我们用Consumer来创建订单：
- 判断消息type是否为
create.trip - 调用DRF
trip = serializer.create()创建 - 注意Django的数据库操作，都是同步的，而Channels是异步的，所以需要加个装饰器：
@database_sync_to_async - 创建Trip记录后，再添加用户信息，调用
ReadOnlyTripSerializer() - 发送Websockets:
self.send_json() - 新订单创建时，通知所有的司机：
channel_layer.group_send( group='drivers', message={ 'type': 'echo.message', 'data': trip_data } )- 其中
'type': 'echo.message'，Channels会自动调用echo_message(event)函数，保证在drivers组里的司机们都能收到
- 其中
# api/consumers.py
from channels.db import database_sync_to_async # new
from channels.generic.websocket import AsyncJsonWebsocketConsumer
from api.serializers import ReadOnlyTripSerializer, TripSerializer # new
class TaxiConsumer(AsyncJsonWebsocketConsumer):
# modified
async def connect(self):
user = self.scope['user']
if user.is_anonymous:
await self.close()
else:
channel_groups = []
# Add a driver to the 'drivers' group.
user_group = await self._get_user_group(self.scope['user'])
if user_group == 'driver':
channel_groups.append(self.channel_layer.group_add(
group='drivers',
channel=self.channel_name
))
# Get trips and add rider to each one's group.
self.trips = set([
str(trip_id) for trip_id in await self._get_trips(self.scope['user'])
])
for trip in self.trips:
channel_groups.append(self.channel_layer.group_add(trip, self.channel_name))
await asyncio.gather(*channel_groups)
await self.accept()
# new
async def receive_json(self, content, **kwargs):
message_type = content.get('type')
if message_type == 'create.trip':
await self.create_trip(content)
# new
async def echo_message(self, event):
await self.send_json(event)
# new
async def create_trip(self, event):
trip = await self._create_trip(event.get('data'))
trip_id = f'{trip.id}'
trip_data = ReadOnlyTripSerializer(trip).data
# Send rider requests to all drivers.
await self.channel_layer.group_send(
group='drivers', message={
'type': 'echo.message',
'data': trip_data
}
)
# Add trip to set.
if trip_id not in self.trips:
self.trips.add(trip_id)
# Add this channel to the new trip's group.
await self.channel_layer.group_add(
group=trip_id, channel=self.channel_name
)
await self.send_json({
'type': 'create.trip',
'data': trip_data
})
# new
@database_sync_to_async
def _create_trip(self, content):
serializer = TripSerializer(data=content)
serializer.is_valid(raise_exception=True)
trip = serializer.create(serializer.validated_data)
return trip
前端 - 创建订单
点击导航条上的叫车按钮，显示对话框：
image.png
<template>
# /src/App.vue
<v-dialog v-model="dialog">
<v-card>
<v-card-title class="headline">你想去哪里？</v-card-title>
<v-card-text>
<v-layout row>
<v-flex xs12 sm8>
<v-text-field
name="from"
label="出发地点"
v-model="from"
type="text"
required></v-text-field>
</v-flex>
</v-layout>
<v-layout row wrap>
<v-flex xs12 sm8>
<v-text-field
name="dest"
label="目的地"
v-model="dest"
type="text"
required></v-text-field>
</v-flex>
</v-layout>
</v-card-text>
<v-card-actions>
<v-btn
color="red"
flat="flat"
@click="dialog = false"
>
Cancel
</v-btn>
<v-spacer></v-spacer>
<v-btn
color="green"
flat outline
:disabled="!(from && dest)"
@click="dialog = false; callTaxi()"
>
叫车
</v-btn>
</v-card-actions>
</v-card>
</v-dialog>
点击对话框里的“叫车”时，调用Vuex的createTrip action来发送WebSockets消息：
<script>
data () {
return {
dialog: false,
from: '',
dest: ''
}
},
methods: {
...mapActions(['clearAlert']),
menu_click (title) {
if (title === 'Exit') {
this.$store.dispatch('messages/signUserOut')
} else if (title === 'Call') {
this.dialog = true
}
},
callTaxi () {
let data = { pick_up_address: this.from, drop_off_address: this.dest, rider: this.user.id }
this.$store.dispatch('ws/createTrip', data)
}
}
同axios，所有与后台交互的WS操作，全部集中到wsService.js中，方便管理和更新。
# /src/services/wsService.js
// send Websockets msg to server
export default {
async createTrip (ws, payload) {
let data = JSON.stringify({
type: 'create.trip',
data: payload
})
await ws.send(data)
}
}
然后，Vuex store里，根据需求，添加不同的actions:
# /src/store/modules/ws.js
const actions = {
async createTrip ({ commit }, message) {
await wsService.createTrip(state.websocket.ws, message)
},
async updateTrip ({ commit }, message) {
await wsService.updateTrip(state.websocket.ws, message)
}
}
测试：按F12，浏览器Console窗口，点叫车按钮，输入数据，就能看到创建成功的WS消息了：
WS received: {
"type":"create.trip",
"data":{
"id":"69caf2d4-a9cb-4b3e-80d3-2412a2debe99","driver":null,
"rider":{"id":2,"username":"rider1","first_name":"","last_name":""},
"created":"2019-05-19T11:40:41.278098Z",
"updated":"2019-05-19T11:40:41.278126Z",
"pick_up_address":"南京",
"drop_off_address":"大理",
"status":"REQUESTED"}
}
收到后台WS消息后，setAlert消息，并且更新“当前订单”。这是前端业务逻辑，集中放在ws.js中
# /src/store.modules/ws.js
const actions = {
// handle msg from server
wsOnMessage ({ dispatch, commit }, e) {
const rdata = JSON.parse(e.data)
console.log('WS received: ' + JSON.stringify(rdata))
switch (rdata.type) {
case 'create.trip':
commit('messages/addTrip', rdata.data, { root: true })
break
case 'update.trip':
break
}
},
添加addTrip action，并且我们让trips按更新时间逆序排序：
# /scr/store/modules/messages.js
const getters = {
trips: state => {
return state.trips.sort((a, b) => new Date(b.updated) - new Date(a.updated))
},
}
const mutations = {
addTrip (state, messages) {
state.trips.splice(0, 0, message)
},
image.png
Channels 更新消息的群发群收
用户创建订单后，如果有司机接单，则用户应能即时得到通知。
用户退出时，司机也能收到通知。
实现：利用Channels group
Consumer
- 每个用户，维护一个trips列表
- 在新订单创建后，
channel_layer.group_add来新建一个group - 群，在群内的所有成员(乘客和司机)，会同时收到更新提醒 - 用户WS连接关闭（可能是退出程序，也可能是无信号），则Channels里解散用户所处的群，并把trips列表清空
# backend/api/consumers.py
import asyncio # new
from channels.db import database_sync_to_async
from channels.generic.websocket import AsyncJsonWebsocketConsumer
from api.serializers import ReadOnlyTripSerializer, TripSerializer
class TaxiConsumer(AsyncJsonWebsocketConsumer):
# new
def __init__(self, scope):
super().__init__(scope)
# Keep track of the user's trips.
self.trips = set()
async def connect(self): ...
async def receive_json(self, content, **kwargs): ...
# new
async def echo_message(self, event):
await self.send_json(event)
# changed
async def create_trip(self, event):
trip = await self._create_trip(event.get('data'))
trip_id = f'{trip.id}'
trip_data = ReadOnlyTripSerializer(trip).data
# Add trip to set.
self.trips.add(trip_id)
# Add this channel to the new trip's group.
await self.channel_layer.group_add(
group=trip_id,
channel=self.channel_name
)
await self.send_json({
'type': 'create.trip',
'data': trip_data
})
# new
async def disconnect(self, code):
# Remove this channel from every trip's group.
channel_groups = [
self.channel_layer.group_discard(
group=trip,
channel=self.channel_name
)
for trip in self.trips
]
asyncio.gather(*channel_groups)
# Remove all references to trips.
self.trips.clear()
await super().disconnect(code)
@database_sync_to_async
def _create_trip(self, content): ...
用户恢复WS连接时，应该能从数据库里，读取已有trip，然后重新添加用户到群里
Consumer
-
_get_trips读取数据库记录，排除已完成的订单 -
channel_layer.group_add添加用户到所有未完成订单的群里
# api/consumers.py
import asyncio
from channels.db import database_sync_to_async
from channels.generic.websocket import AsyncJsonWebsocketConsumer
from api.models import Trip # new
from api.serializers import ReadOnlyTripSerializer, TripSerializer
class TaxiConsumer(AsyncJsonWebSocketConsumer):
def __init__(self, scope): ...
# changed
async def connect(self):
user = self.scope['user']
if user.is_anonymous:
await self.close()
else:
# Get trips and add rider to each one's group.
channel_groups = []
self.trips = set([
str(trip_id) for trip_id in await self._get_trips(self.scope['user'])
])
for trip in self.trips:
channel_groups.append(self.channel_layer.group_add(trip, self.channel_name))
asyncio.gather(*channel_groups)
await self.accept()
async def receive_json(self, content, **kwargs): ...
async def echo_message(self, event): ...
async def create_trip(self, event): ...
async def disconnect(self, code): ...
@database_sync_to_async
def _create_trip(self, content): ...
# new
@database_sync_to_async
def _get_trips(self, user):
if not user.is_authenticated:
raise Exception('User is not authenticated.')
user_groups = user.groups.values_list('name', flat=True)
if 'driver' in user_groups:
return user.trips_as_driver.exclude(
status=Trip.COMPLETED
).only('id').values_list('id', flat=True)
else:
return user.trips_as_rider.exclude(
status=Trip.COMPLETED
).only('id').values_list('id', flat=True)
创建订单时，检查是否已存在记录。如果已存在，则跳过加群的步骤。
# api/consumers.py
async def create_trip(self, event):
trip = await self._create_trip(event.get('data'))
trip_id = f'{trip.id}'
trip_data = ReadOnlyTripSerializer(trip).data
# Handle add only if trip is not being tracked.
if trip_id not in self.trips:
self.trips.add(trip_id)
await self.channel_layer.group_add(
group=trip_id,
channel=self.channel_name
)
await self.send_json({
'type': 'create.trip',
'data': trip_data
})
更新订单
Consumer
- 如果司机/乘客更新了订单，则触发
update_trip动作 - 通过Serializer，更新订单状态
- 如果是司机接单，则把司机加入到群里
channel_layer.group_add() - 通知乘客，已有司机接单。
(group=trip_id, message={ 'type': 'echo.message', 'data': trip_data })- 注意
message={'type': 'echo.message'，Channels会自动寻找对应的方法函数：echo_message(event)
- 注意
# api/consumers.py
import asyncio
from channels.db import database_sync_to_async
from channels.generic.websocket import AsyncJsonWebsocketConsumer
from api.models import Trip
from api.serializers import ReadOnlyTripSerializer, TripSerializer
class TaxiConsumer(AsyncJsonWebsocketConsumer):
def __init__(self, scope): ...
async def connect(self): ...
async def receive_json(self, content, **kwargs):
message_type = content.get('type')
if message_type == 'create.trip':
await self.create_trip(content)
elif message_type == 'update.trip': # new
await self.update_trip(content)
async def echo_message(self, event): ...
async def create_trip(self, event): ...
# new
async def update_trip(self, event):
trip = await self._update_trip(event.get('data'))
trip_id = f'{trip.id}'
trip_data = ReadOnlyTripSerializer(trip).data
# Send updates to riders that subscribe to this trip.
await self.channel_layer.group_send(group=trip_id, message={
'type': 'echo.message',
'data': trip_data
})
if trip_id not in self.trips:
self.trips.add(trip_id)
await self.channel_layer.group_add(
group=trip_id,
channel=self.channel_name
)
await self.send_json({
'type': 'update.trip',
'data': trip_data
})
async def disconnect(self, code): ...
@database_sync_to_async
def _create_trip(self, content): ...
@database_sync_to_async
def _get_trips(self, user): ...
# new
@database_sync_to_async
def _update_trip(self, content):
instance = Trip.objects.get(id=content.get('id'))
# https://www.django-rest-framework.org/api-guide/serializers/#partial-updates
serializer = TripSerializer(data=content, partial=True)
serializer.is_valid(raise_exception=True)
trip = serializer.update(instance, serializer.validated_data)
return trip
引入User group概念
为了区分用户是乘客还是司机，需要把用户分组。
数据模型添加group计算字段，类似于Vue computed()：
class User(AbstractUser):
# photo = models.ImageField(upload_to='photos', null=True, blank=True)
@property
def group(self):
groups = self.groups.all()
return groups[0].name if groups else None
DRF Serializer在注册时，增加group字段的处理：
# /backend/api/serializers.py
class UserSerializer(serializers.ModelSerializer):
password1 = serializers.CharField(write_only=True)
password2 = serializers.CharField(write_only=True)
group = serializers.CharField()
# photo = MediaImageField(allow_empty_file=True)
def validate(self, data):
if data['password1'] != data['password2']:
raise serializers.ValidationError('两次密码不一致')
return data
def create(self, validated_data):
group_data = validated_data.pop('group')
group, _ = Group.objects.get_or_create(name=group_data)
data = {
key: value for key, value in validated_data.items()
if key not in ('password1', 'password2')
}
data['password'] = validated_data['password1']
user = self.Meta.model.objects.create_user(**data)
user.groups.add(group)
user.save()
return user
class Meta:
model = get_user_model()
fields = (
'id', 'username', 'password1', 'password2', 'first_name', 'last_name', 'group', #'photo',
)
read_only_fields = ('id',)
admin后台管理页面：
# /backend/api/admin.py
@admin.register(User)
class UserAdmin(DefaultUserAdmin):
list_display = (
'username', 'id', 'group', 'first_name', 'last_name', 'email', 'is_staff',
)
readonly_fields = (
'id',
)
注意：数据库不需要重新migrate，应该不是新字段，而且计算字段。
注意：已有用户，需要在admin里添加“group”字段。或者删除重新注册。
前端Sign-Up页面
我们在注册用户时，让用户选择不同角色：
image.png
更新一下Vue view:
<template>
# /src/views/Signup.vue
<v-radio-group v-model="group" row>
<v-radio label="乘客" value="rider"></v-radio>
<v-radio label="司机" value="driver"></v-radio>
</v-radio-group>
<v-layout>
<v-flex xs12>
<v-card-actions>
<v-spacer />
<v-btn round type="submit" :loading="loading" class="orange">Register</v-btn>
</v-card-actions>
</v-flex>
</v-layout>
<script>
data () {
return {
username: '',
password: '',
confirmPassword: '',
group: 'rider'
}
},
methods: {
onSignup () {
this.$store.dispatch('messages/signUserUp', { username: this.username, password2: this.confirmPassword, password1: this.password, group: this.group })
},
总结
后台对订单的更新、群发群收，已经全部ready了。
下一篇，会介绍前端如何处理订单更新
带你进入异步Django+Vue的世界 - Didi打车实战(6)
网友评论