ARTICLE DETAIL

资讯详情

深耕网站视觉设计与运营推广的一线实战洞察。

Django投票应用:ORM、Token与WebSocket实时推送

Django投票应用:ORM、Token与WebSocket实时推送 “投票应用”这个项目几乎是Django官方教程的同义词但很多新手做到一半就卡住环境配好了、app也建了模型一跑就报错之后想加个WebSocket实时图表又不知道从哪下手。这篇文章我准备带着你完整走一遍投票应用的搭建过程同时把新手最常搜的那几个点——ORM查询与删除、cookie设置token、后台数据推送到前端——全部串起来讲清楚。内容面向刚接触Django的人也适合想快速回顾全流程的老朋友。1. 项目启动从零到Django投票应用的地基1.1 环境准备与Django版本选择我先说结论做这类实战项目别在全局环境里装Django一定要用虚拟环境。原因很简单真实开发中不同项目依赖的Django版本可能不一样今天做一个投票应用用了Django 4.x明天接一个老项目可能还在Django 2.2全局环境会让依赖互相打架。mkdir django-polls-project cd django-polls-project python -m venv venvWindows下激活虚拟环境是venv\Scripts\activatemacOS/Linux是source venv/bin/activate。激活后安装Django我建议直接装当前稳定版本比如Django 4.x或5.xpip install django装完可以用python -m django --version看一眼。Django版本差异主要体现在路由写法、时区配置、以及一些新特性上对于投票应用这个规模的项目4.x和5.x的写法几乎一致不用过于纠结。1.2 使用django-admin创建项目与应用Django项目和应用是两个概念我用一句生活化比喻讲清楚项目是“服务器机房”应用是“机房里的一个业务部门”。一个项目可以包含多个应用投票应用就是其中一个业务部门。django-admin startproject config . python manage.py startapp polls这里有个小技巧创建项目时末尾加一个点表示在当前目录生成配置文件而不是再嵌套一层目录。很多人不加点结果项目文件夹里又套了一个文件夹后面manage.py路径全乱了。startapp polls会生成polls目录里面有models.py、views.py、admin.py等文件。这里必须强调一下很多人误以为只要执行了这个命令应用就生效了其实还要把应用注册到项目的settings.py里Django才知道有这个业务部门的存在。1.3 应用注册与初始配置打开config/settings.py找到INSTALLED_APPS这个列表把polls加进去INSTALLED_APPS [ django.contrib.admin, django.contrib.auth, django.contrib.contenttypes, django.contrib.sessions, django.contrib.messages, django.contrib.staticfiles, polls, ]在创建应用之后我还会顺手把LANGUAGE_CODE改成zh-hansTIME_ZONE改成Asia/Shanghai。这不是什么花哨操作纯粹是为了让后台显示中文、时间对得上新手往往忽略这一点导致后台一堆英文日期看着别扭。做完这些先跑一下python manage.py runserver浏览器打开http://127.0.0.1:8000/看到火箭图标页面说明项目地基已经打好了。2. 数据模型与ORM操作让投票数据跑起来2.1 设计Question和Choice模型模型的本质是数据库表的设计蓝图。Django通过Python类来描述数据库表不需要手写SQL这也是Django能快速开发的核心原因之一。投票应用至少需要两张表问题表Question和选项表Choice。问题表存储投票标题和发布日期选项表存储每个投票选项的文字和它的票数。关键设计是选项与问题之间的关联关系一个问题对应多个选项这是典型的一对多关系用ForeignKey外键表达# polls/models.py from django.db import models class Question(models.Model): question_text models.CharField(max_length200, verbose_name问题内容) pub_date models.DateTimeField(verbose_name发布时间) def __str__(self): return self.question_text class Choice(models.Model): question models.ForeignKey(Question, on_deletemodels.CASCADE, related_namechoices) choice_text models.CharField(max_length200, verbose_name选项内容) votes models.IntegerField(default0, verbose_name票数) def __str__(self): return self.choice_text这里有几个新手容易犯的错第一__str__方法不写后台看到的对象全是Question object (1)没法调试第二on_deletemodels.CASCADE不理解含义结果删掉一个问题时它的所有选项也被自动删除有时这不是你想要的结果。如果希望删除问题时保留选项记录可以改成on_deletemodels.SET_NULL但对应外键字段要加nullTrue。我在设计Choice时加了related_namechoices这样可以在Question对象上直接通过question.choices.all()取到所有选项比默认的choice_set语义更清晰。2.2 初次数据迁移两条命令背后的原理模型写好后马上执行两条命令python manage.py makemigrations polls python manage.py migrate第一条命令是根据模型变化生成迁移文件第二条命令是把迁移文件真正应用到数据库。很多新手不理解为什么非要分两步直接把第一步做完就去跑服务器控制台一直报“没有这个表”。这里打个比方makemigrations是写“施工方案”migrate是照着方案“施工”两个动作必须连贯完成。如果之后修改了模型比如给Choice增加一个字段需要重新跑这两条命令。千万别手工改数据库表否则Django的迁移记录会和你实际表结构对不上后面排查起来非常痛苦。2.3 ORM执行查询常用姿势Django最有生产力的部分就是ORM它让你用Python语法操作数据库。我把新手最常用的查询操作集中说一下顺便解决热搜词里的“django执行查询”。先进入Django shellpython manage.py shell创建数据from polls.models import Question, Choice from django.utils import timezone q Question(question_text你最喜欢哪个城市, pub_datetimezone.now()) q.save()这里有个细节pub_date建议用django.utils.timezone.now()而不是datetime.datetime.now()因为Django启用了时区支持后用标准库的now()会产生naive datetime警告后期存储到数据库容易出现时区偏移。查询数据# 获取全部 Question.objects.all() # 按条件过滤返回QuerySet Question.objects.filter(question_text__contains城市) # 获取单个对象 question Question.objects.get(id1) # 关联查询获取某个问题的所有选项 question.choices.all() # 排序 Choice.objects.order_by(-votes)filter返回的是QuerySet一整个可迭代数据集get返回的是单个模型对象。如果用get查询时匹配到多条记录会抛MultipleObjectsReturned异常如果一条都没有会抛DoesNotExist异常。所以get适合确定唯一记录的场景不确定的时候建议先.filter().first()。2.4 删除对象的安全操作“django执行查询-删除对象”是很多人的痛点。删除操作最要命的不是不会写而是删错了救不回来。我每次在项目里执行删除之前都会先查一次Question.objects.filter(pub_date__year2023).count()确认数量没问题再删。删除单个对象q Question.objects.get(id1) q.delete()批量删除Choice.objects.filter(choice_text__startswith北京).delete()这里有两个大坑第一默认批量删除不会调用模型里重写的delete()方法如果你重写了delete()做额外清理批量删除时不会触发第二只要表中存在外键关联删除父对象会级联删除子对象所以删Question前必须考虑是否想让Choice一起消失。我个人的习惯是对于重要业务数据优先做“软删除”——给模型加一个is_active布尔字段查询时默认过滤掉is_activeFalse的记录。虽然投票应用规模小但养成这个习惯以后接手复杂业务系统时会少踩很多坑class Question(models.Model): # ... 其他字段 is_active models.BooleanField(defaultTrue)3. 后台管理与投票业务逻辑3.1 注册模型到Django Admin后台Django的admin后台是公认的“白送的福利”一张表都不用手写注册进去就能增删改查。我在开发阶段几乎全靠这个后台录入测试数据。打开polls/admin.pyfrom django.contrib import admin from .models import Question, Choice class ChoiceInline(admin.TabularInline): model Choice extra 2 class QuestionAdmin(admin.ModelAdmin): list_display (question_text, pub_date, is_active) fieldsets [ (None, {fields: [question_text]}), (日期信息, {fields: [pub_date]}), ] inlines [ChoiceInline] admin.site.register(Question, QuestionAdmin)通过TabularInline你可以在编辑问题的页面直接添加多个选项省去来回切换页面的麻烦。list_display设置后台列表页显示的字段fieldsets用来分组布局这些都是最常用的admin自定义能力。这里提一个新词django unfold它是Django admin的现代化皮肤方案给后台换上类似Notion风格的界面。项目上线后如果想给运营同学一个更好看的后台直接搜“django unfold”配置即可几分钟就能替换默认admin样式投入产出比很高。3.2 路由、视图与模板三板斧投票应用的标准流程是首页展示所有问题点进问题详情选择选项后提交展示投票结果。我建议新手先老老实实用函数视图走一遍理解“请求-视图-响应”的完整流程再考虑去用ListView、DetailView这类通用视图。函数视图的可读性强有问题时好排查。在polls/urls.py中配置路由项目根路由用path(polls/, include(polls.urls))引入from django.urls import path from . import views urlpatterns [ path(, views.index, nameindex), path(int:question_id/, views.detail, namedetail), path(int:question_id/vote/, views.vote, namevote), path(int:question_id/results/, views.results, nameresults), ]视图函数先跳过投票逻辑展示基础写法from django.shortcuts import render, get_object_or_404 from .models import Question def index(request): questions Question.objects.filter(is_activeTrue).order_by(-pub_date) return render(request, polls/index.html, {questions: questions}) def detail(request, question_id): question get_object_or_404(Question, pkquestion_id) return render(request, polls/detail.html, {question: question}) def results(request, question_id): question get_object_or_404(Question, pkquestion_id) return render(request, polls/results.html, {question: question})这里出现了一个核心经验用get_object_or_404而不是自己写try...except。当查询不到记录时直接返回404代码干净且符合HTTP语义。模板部分建议在polls/templates/polls/目录下建文件detail.html大致长这样h1{{ question.question_text }}/h1 form action{% url polls:vote question.id %} methodpost {% csrf_token %} {% for choice in question.choices.all %} input typeradio namechoice idchoice{{ forloop.counter }} value{{ choice.id }} label forchoice{{ forloop.counter }}{{ choice.choice_text }}/labelbr {% endfor %} input typebutton value投票 /form模板里最容易犯的错是忘了加{% csrf_token %}。Django默认开启了CSRF中间件所有POST请求都必须带这个令牌否则会返回403。曾经有个同事在做一个带内嵌表单的页面时忘记加上去排查了整整一个下午。3.3 投票表单与CSRF防护投票视图是业务核心逻辑其实非常简单拿到用户选中的Choice对象把它的votes字段加1然后重定向到结果页from django.http import HttpResponseRedirect from django.urls import reverse from .models import Choice def vote(request, question_id): question get_object_or_404(Question, pkquestion_id) try: selected_choice question.choices.get(pkrequest.POST[choice]) except (KeyError, Choice.DoesNotExist): return render(request, polls/detail.html, { question: question, error_message: 请先选择一个选项。, }) selected_choice.votes 1 selected_choice.save(update_fields[votes]) return HttpResponseRedirect(reverse(polls:results, args(question.id,)))注意两个点第一request.POST[choice]如果取不到值会抛KeyError必须捕获第二投票成功后使用HttpResponseRedirect重定向而不是直接渲染结果页这符合“POST-重定向-GET”模式可以防止用户刷新页面时重复提交投票。很多教程到这一步就停了但投票应用还有一个常见的业务需求用户只能投一次票。如果用Cookie做判断逻辑很简单——投票前先检查Cookie有没有标记如果投过就提示“你已经投过票了”投票后通过HttpResponse设置一个标记Cookie。3.4 Cookie与Token设置投票状态“django cookie设置token”是很多人做登录状态和用户识别时反复搜索的需求。这里我讲一个可以直接用的方案投票成功后在响应中写入一条HttpOnly的Cookie值可以是一个带签名的token。from django.http import HttpResponseRedirect from django.urls import reverse import hashlib, time def vote(request, question_id): # ... 前面的校验逻辑保持不变 ... # 生成一个简单的签名token token hashlib.sha256(( f{question_id}|{selected_choice.id}|{request.META.get(REMOTE_ADDR, )}|{time.time()} ).encode()).hexdigest() response HttpResponseRedirect(reverse(polls:results, args(question.id,))) response.set_cookie( keyfvoted_{question.id}, valuetoken, max_age86400, httponlyTrue, samesiteLax, ) return response这里说一下几个Cookie参数的用意max_age86400表示Cookie有效期一天httponlyTrue可以防止前端JavaScript读取这个Cookie降低XSS风险samesiteLax限制跨站请求携带Cookie算是最基础的安全防护。如果你需要更完整的用户登录体系建议直接用Django的session框架或JWT不必自己造轮子。用Django内置session时判断重复投票只需要检查request.session.get(fvoted_{question.id})逻辑更加简单。Cookie适合存标识类信息敏感数据还是放服务端session更稳妥。4. 投票结果实时推送Django WebSocket实战4.1 HTTP轮询与WebSocket方案对比投票应用做到这里功能已经能跑了。但真实项目往往还有更“现代”的需求多个人在投票页面能不能不刷新就自动显示最新结果这就是热搜词“python django websocket实现后台有数据前端推送”要解决的问题。Django默认是纯HTTP请求-响应模式浏览器发起请求服务器返回结果连接就断了。想要“后台有数据自动推给前端”传统做法是前端定时轮询比如每5秒发一次Ajax请求查最新票数。这种方式简单但效率低用户多起来服务器会扛不住大量空轮询。WebSocket则是一个长连接服务器可以随时主动往客户端推送数据。Voting应用这种场景有人投完票服务端把最新结果推送出去所有在线页面马上更新体验比轮询好很多。4.2 集成Channels与ASGI配置在Django中实现WebSocket最主流的方案是Channels。它把Django从纯HTTP扩展为支持WebSocket等协议。安装依赖pip install channels channels-redis然后把channels加到INSTALLED_APPS中并修改config/asgi.pyimport os from django.core.asgi import get_asgi_application from channels.routing import ProtocolTypeRouter, URLRouter from channels.auth import AuthMiddlewareStack from django.urls import path from polls.consumers import VoteConsumer os.environ.setdefault(DJANGO_SETTINGS_MODULE, config.settings) django_asgi_app get_asgi_application() application ProtocolTypeRouter({ http: django_asgi_app, websocket: AuthMiddlewareStack( URLRouter([ path(ws/votes/, VoteConsumer.as_asgi()), ]) ), })这里需要重点解释Channel Layer的作用。它相当于一个跨消费者通信的“消息总线”不同的WebSocket连接可以在不同进程里但通过Redis作为底层通道消息可以在它们之间传递。投票事件发生时视图函数把消息发布到channel layer所有消费者再收到通知后把结果推给各自的浏览器。如果没有配置channel layer多个服务器进程之间将无法同步推送消息。开发环境可以用InMemoryChannelLayer生产环境我推荐Redis也就是刚刚安装的channels-redis。在settings.py里配置CHANNEL_LAYERS { default: { BACKEND: channels_redis.core.RedisChannelLayer, CONFIG: { hosts: [(127.0.0.1, 6379)], }, }, }4.3 WebSocket消费者与后端推送实现消费者才是WebSocket的核心逻辑。我写一个最简单的消费者接收“投票事件”消息然后把最新结果广播给所有客户端# polls/consumers.py import json from channels.generic.websocket import AsyncWebsocketConsumer class VoteConsumer(AsyncWebsocketConsumer): async def connect(self): self.group_name votes await self.channel_layer.group_add(self.group_name, self.channel_name) await self.accept() async def disconnect(self, close_code): await self.channel_layer.group_discard(self.group_name, self.channel_name) async def vote_broadcast(self, event): # 收到消息后把数据发送到WebSocket客户端 await self.send(text_datajson.dumps(event[data]))然后在投票视图里投票成功后通过channel_layer发送广播。因为视图是同步函数需要用到async_to_sync来调用异步的channel layer方法from asgiref.sync import async_to_sync from channels.layers import get_channel_layer def vote(request, question_id): # ... 投票逻辑 ... votes_data { str(choice.id): choice.votes for choice in question.choices.all() } channel_layer get_channel_layer() async_to_sync(channel_layer.group_send)( votes, { type: vote_broadcast, data: { question_id: question.id, votes: votes_data, } } ) return HttpResponseRedirect(reverse(polls:results, args(question.id,)))这套流程就是“Django后端有数据时主动推送”的完整语义视图里更新了数据库然后把变更事件发到中间层消费者收到后再推给前端。整个过程不需要前端轮询。4.4 前端订阅与断线重连前端页面需要通过原生WebSocket API连接后端并处理动态更新。我习惯在模板里放一段JavaScriptlet ws new WebSocket(ws://${window.location.host}/ws/votes/); ws.onmessage function(e) { const data JSON.parse(e.data); console.log(最新数据, data.votes); // 这里可以更新页面上对应的票数元素 for (const [choiceId, votes] of Object.entries(data.votes)) { const el document.querySelector(#choice-votes-${choiceId}); if (el) el.textContent votes; } };一个很容易被忽略的问题是断线重连。WebSocket连接会因为网络切换、服务器重启等原因断开如果不做重连机制页面就变成“僵尸连接”永远收不到新数据。我的经验是至少加一个简单的重试逻辑function connectWebSocket() { const ws new WebSocket(ws://${window.location.host}/ws/votes/); ws.onclose function() { setTimeout(connectWebSocket, 3000); }; return ws; } let ws connectWebSocket();断线重连的间隔建议设在2到5秒之间太短会导致服务器承受大量瞬时连接太长又会体验延迟。有条件和精力的话还可以在连接恢复后拉一次全量数据作为补偿。5. 新手避坑清单与常用命令速查5.1 十个常见问题与排查我见过的Django新手项目里以下十个问题几乎轮流出现。这里做成一个速查表方便你排查时直接对照。问题现象大概率原因解决办法启动后找不到模板模板没放在正确目录或没按app注册名建目录检查templates/polls/目录是否存在应用是否在INSTALLED_APPS里提交表单返回403模板缺少{% csrf_token %}表单内加上{% csrf_token %}标签ImportError: No module named polls执行命令时不在项目根目录或应用没有注册确认当前目录包含manage.py检查INSTALLED_APPS数据库表不存在改了模型但没执行迁移依次执行makemigrations和migrate后台没有应用数据表admin.py没注册模型在admin.py中调用admin.site.register(Model)静态文件加载不出来模板中未使用{% static %}标签或未配置STATIC_URL模板开头加{% load static %}用{% static css/app.css %}引用时间显示成UTCTIME_ZONE没改在settings.py中设置TIME_ZONE Asia/Shanghaiget()返回多条记录条件不够精确改用.filter().first()或增加约束条件批量删除后数据救不回来没有备份且级联删除触发删除前执行.count()确认使用时考虑软删除WebSocket连不上没配ASGI路由或没装channels检查是否启动python manage.py runserver确认ASGI路由正确INSTALLED_APPS有channels这表格里的每一个问题我都在带新人的过程中真实遇到过尤其前三个基本每次培训都会有人踩。5.2 ORM与性能优化要点投票应用规模不大但既然做项目就不能只满足于“能跑”。在ORM使用上有几个常见的性能问题值得提前了解等以后做复杂系统时这些原则能帮你少走弯路。第一个是N1查询问题。列表页展示问题时如果每个问题都要取一次选项列表循环中就会反复执行数据库查询。解决办法是用prefetch_related(choices)一次把关联数据取出来questions Question.objects.filter(is_activeTrue).prefetch_related(choices)第二个是更新字段时尽量只更新变化的字段。前面投票时我用了save(update_fields[votes])这样Django只生成包含votes字段的UPDATE语句避免把整行数据重新写一遍在高并发场景下能减少写冲突。第三个是避免在循环里操作数据库。很多人喜欢这样写for question in questions: question.votes 1 question.save()这会产生N条SQL语句。改用批量更新可以大幅提升效率from django.db.models import F Choice.objects.filter(question_id1).update(votesF(votes) 1)这里用到了F表达式它让数据库在内部完成字段的自增操作既快又避免了并发时的竞态条件问题。5.3 WebSocket常见故障WebSocket这一部分的坑比普通HTTP多。第一个是忘记安装Redis只配置了RedisChannelLayer但本机没有启动Redis服务连接会一直报错。开发环境下可以先临时改用InMemoryChannelLayer应急但要注意它不支持跨进程通信。第二个坑是部署到生产环境时Django自带的runserver不一定能正确处理ASGI应用。真实部署时需要用uvicorn或daphne来启动ASGI服务并通过Nginx配置WebSocket的Upgrade请求头否则客户端连接时直接被拒绝。第三个问题是消费者里用了同步ORM操作。记住在AsyncWebsocketConsumer中直接调用数据库同步ORM会导致事件循环阻塞。如果确实需要查询数据库要么用database_sync_to_async包裹要么改用AsyncWebsocketConsumer外的同步环境执行。这个坑比较隐蔽遇到页面卡顿或连接超时可以优先排查。5.4 常用命令速查我把整个投票应用开发过程中高频使用的命令和它们的作用整理成表格方便直接抄作业。这些命令在任何一个Django项目里都是通用的。命令作用使用时机python -m venv venv创建虚拟环境项目开始pip install django安装Django环境准备django-admin startproject config .创建项目项目开始python manage.py startapp polls创建应用新增业务模块python manage.py makemigrations生成迁移文件修改模型后python manage.py migrate应用数据库迁移生成迁移文件后python manage.py createsuperuser创建管理员账号使用admin后台前python manage.py shell进入交互式环境调试ORM或业务逻辑python manage.py runserver启动开发服务器日常开发pip install channels channels-redis安装WebSocket相关库需要实时推送时这些命令的执行前提是虚拟环境已激活。我之前不止一次看到有人忘记激活虚拟环境结果系统提示“Django不是内部或外部命令”这种问题排查起来很简单看一眼命令行前有没有(venv)标记即可。做完整套投票应用我的体会是Django真正的学习曲线不在“创建应用”这些起步操作而在ORM的关联查询、WebSocket的异步通信机制、以及生产环境部署时的配置细节。投票应用虽然小但把这几个知识点从头到尾跑通了Django框架的骨架基本就掌握得差不多了。最后再分享一个小技巧如果你做的项目比较复杂给Django后台换皮肤时可以直接用django unfold精简配置后比默认admin好看很多配合自定义的admin配置运营和产品同事都会觉得这个系统“很专业”。投票应用后续可以继续扩展的功能也很多用户投票历史记录、按时间段统计趋势图、多用户权限分组、导出Excel报表思路都在上面了关键是先把地基打扎实。
返回列表