Django 入门 — Python 全栈 Web 框架 Django ORM 数据库操作 Django Admin — 自动后台管理 Django 表单 — 自动生成与验证 Django 模板引擎 Django 认证系统 Django REST Framework — 构建 API Django 中间件 Django 测试 — 单元测试 Django 信号 Django Celery 异步任务 Django 配置管理 Django 模型字段 Django 部署上线 Django Model 深入 Django 视图深入 Django REST Framework 深入 Django Channels WebSocket Django + Celery 异步任务 Django 生产部署 Django QuerySet 查询集深入 Django Admin 定制 Django 缓存系统 Django 文件上传 Django Migrations 迁移深入 Django Serializer 序列化 Django 模型字段深入 Django CBV 类视图深入 Django Forms 表单深入 Django Sessions 会话深入 Django缓存实战 Django分页技术 Django文件上传 Django自定义User模型 Django管理命令 Django中间件实战 Django信号实战 Django测试入门 Django性能优化 Django 模型 Model 层 Django安全最佳实践 Django 视图 View 层 Django中间件实战 Django 模板 Template 层 Django信号实战 Django REST Framework 入门 Django性能优化 Django缓存实战 Django缓存入门 Django ORM 自定义聚合函数实战动态任务链与 Celery Canvas 高级编排悲观锁与 select_for_update 高并发扣库存实战 Django Channels 自定义内存层与广播模式零停机字段重命名迁移策略异步中间件处理流式响应与请求取消基于属性的访问控制（ABAC）后端实现原生 Django ContentTypes 实现多态查询优化基于 HTMX 的表单动态验证与部分渲染 Elasticsearch 集成：自定义分析器与同义词搜索深入Django自定义查询表达式：超越Q与F对象 Django异步信号管道：不阻塞主请求的订阅者链每个请求一个数据库：基于用户的动态多租户路由器 REST API的部分响应合约：实现类似GraphQL的字段筛选器构建可切换的存储抽象层：从本地到S3再到自定义文件系统 Celery动态工作流引擎：通过数据库定义的任务DAG 保护Webhook端点：基于HMAC签名和重放攻击防御使用DataLoader模式消灭N+1查询：在Django REST中批量加载关联数据模型数据版本控制与审计日志：回溯任意历史快照内置慢查询分析器：不依赖外部工具的实时SQL监控面板多数据库与自定义路由：突破单库读写分离异步任务管道：Celery 链式调度与动态重试实时 GraphQL：Subscriptions 与 Django Channels 实战类型安全 Django：Pydantic + mypy 消灭运行时错误附件存储工厂：对接阿里云/七牛/本地统一接口权限矩阵引擎：基于动作-资源的细粒度控制零停机迁移：大表无锁 DDL 与向后兼容部署 WebSocket 压测与高并发优化：Daphne + Uvicorn 调优 Admin 魔法组件：自定义 Widget 与动态表单交互缓存战术手册：多级缓存与缓存预热/失效策略异步中间件链：在Django中构建非阻塞请求处理管道编译型查询集缓存：利用ORM字节码缓存消除N+1查询信号重放防护：为Django信号添加幂等性与去重机制模板流式超媒体：使用Django模板引擎实现服务端推送SSR片段自定义存储后端：基于文件系统的透明加密与权限控制查询集聚合管道：构建像MongoDB一样的链式聚合查询声明式JSON表单渲染：从Django表单直接生成前端JSON Schema 零停机迁移策略：通过影子表与渐进式模式迁移改造生产数据库缓存失效图：基于依赖追踪的智能缓存驱逐系统 WSGI/ASGI混合协议适配器：在同一个Django应用中同时运行同步和异步视图信号驱动的消息中间件：解耦业务与通知自定义数据库路由器：实现多租户读写分离动态定时任务：在运行时创建与管理 Celery Beat 周期任务流式响应实战：用 StreamingHttpResponse 导出千万级 CSV 不爆内存子查询与窗口函数：在 ORM 中实现复杂报表聚合无刷新仪表盘：用 HTMX + Django 构建实时数据图表可插拔验证管道：设计模型与表单的渐进式验证引擎无限级嵌套评论：基于 MPTT 的时间戳排序与分页数据迁移高级模式：在已有表中安全添加非空字段秒传属性系统：自定义模板标签实现动态 HTML 属性注入 Django ORM 与 PostgreSQL 物化视图的智能同步在传统 WSGI 部署中实现 Django GraphQL 订阅 Django ORM 联合查询与异构模型的聚合分页设计可插拔的 Django 通用视图：动态模板与权限工厂混用 Django ORM 与 SQLAlchemy 2.0 实现异步分析报表利用 PostgreSQL 逻辑复制构建 Django 实时数据管道编写 Django 自定义查询表达式：JSONB 深度路径操作与索引用变异测试与随机序列保证 Django 业务逻辑的鲁棒性在 Django 响应管道中注入 HTTP/3 早期提示与资源推送 Django 嵌套事务与保留点深入：隔离级别与原子性陷阱在Django中利用PostgreSQL JSONB字段构建高级搜索引擎动态任务管道：基于Celery和Django构建可配置的数据处理流程基于数据库Schema的Django多租户架构实战基于Django Channels和GraphQL Subscriptions构建实时协作编辑器超越Filter: 用自定义SQL表达式实现复杂JOIN与聚合窗口函数 Django异步视图与ASGI流式响应：构建ChatGPT式流输出基于属性的测试与模糊测试：用Hypothesis武装Django模型与API 自定义存储后端：从零实现兼容S3的MinIO分布式文件管理数据迁移的艺术：用Django迁移脚本安全重构生产数据库 Django + HTMX 内联编辑工作流：无JS的复杂表单交互 DRF序列化器性能调优：缓存预取与惰性加载实战异步信号链：基于Daphne的跨进程事件驱动架构编写自有存储引擎：从SFTP到对象存储的零停机迁移公式驱动字段：在ModelField中嵌入Excel表达式引擎手写SQL查询调试中间件：自动捕获N+1与慢查询并告警像Git一样管理模型版本：行级差异快照与回滚基于属性的API模糊测试：自动发现视图层安全漏洞双向外键：无ORM声明式实现数据库级图关联在Django中玩转HTTP/3：使用aioquic实现服务端推送与0-RTT 运行时类型系统：让Django模型支持动态字段增删而不迁移 Django ASGI WebSocket ORM 桥接模式利用数据库约束与表达式索引增强数据完整性构建动态任务调度器：运行时修改Crontab与间隔 ORM进阶：子查询聚合与公用表表达式(CTE)自定义存储后端：实现预签名直传S3并记录文件元数据 GraphQL变异处理模式：结合SQLAlchemy作为只读后备存储 DRF视图集action级别权限矩阵与动态角色数据库路由器：基于租户标识的水平分片自定义模板引擎：带变动跟踪的晶片缓存标签异步数据迁移：在迁移中安全执行大批量数据变换 Django ORM 查询表达式树与执行计划优化 Django 异步信号与 Channels Layer 深度集成 Celery Beat 动态热加载与分布式锁调度 Pydantic V2 + Django Model 双向序列化与严格校验基于 SAML2 的 Django 自定义认证后端与属性映射 Django 迁移与物化视图的版本化管理 Django + Strawberry GraphQL Subscription 与 ASGI 长连接模式 Django 动态权限规则引擎与表达式解析基于 hypothesis 的 Django 属性基测试与自动数据工厂 Django 自定义存储后端：S3 兼容 MinIO 的分片上传与签名URL 基于Actor模型的Django ASGI WebSocket中间件设计用PostgreSQL SKIP LOCKED实现轻量级任务队列函数组合式中间件链：超越类的管道设计多租户字段级加密：每个租户独立密钥管理基于Git的自动API版本控制：URL零改动用Server-Sent Events实现GraphQL Subscription 物化路径树结构：用Ltree与PostgreSQL实现高效祖先查询 Celery Prefork池下的装饰器模式：任务上下文自动注入用JSON Schema替换DRF Serializer：声明式API校验基于表级哈希的查询缓存自动失效策略用物化路径实现 Django ORM 树形结构信号驱动的竞态条件防御模式编写自定义数据库函数：窗口聚合与排名 Celery Chord 动态任务编排与回调异步视图中的 ORM 连接池管理 PostgreSQL HStore 字段与零停机迁移编写可组合的权限后端：ABAC 属性级控制测试工厂模式：动态 Fixture 与 Scenario 构建中间件优先队列：按条件降级或限流请求 MPTT 树结构的 Redis 缓存失效策略编写自定义缓存后端：从内存到分布式文件系统构建声明式数据过滤器：超越 django-filter 的 AST 解析基于JSON Schema动态生成表单：无代码配置后台实现业务感知的读写分离路由器：按模型、操作类型分发异步信号与分布式锁：防止高并发下重复处理编写Markdown模板引擎：让.md文件直接支持Django标签为管理命令注入AI Agent：自然语言操作数据库基于IPFS的去中心化存储后端：文件永存且可验证窗口函数与子查询：实现复杂分析报表（无需RawSQL）GraphQL订阅与Django Channels：建立实时双向数据管道利用自定义中间件实现请求状态机基于信号的脱敏审计日志系统基于Redis优先级队列的异步任务编排多数据库分片读写分离与动态路由编写动态表单渲染的自定义模板标签与过滤器基于版本号的细粒度缓存失效策略实现自适应滑动窗口限流算法带断点续传能力的有状态数据迁移命令使用VCR模拟外部API并消除网络依赖基于Django Channels的GraphQL订阅实现实时推送

用Server-Sent Events实现GraphQL Subscription

不依赖WebSocket，利用Django的StreamingHttpResponse与Server-Sent Events (SSE)协议，为GraphQL Subscription提供轻量级实时推送，兼容CDN与普通HTTP代理。 · 难度：入门 · +10XP

SSE+GraphQL的优雅组合

GraphQL Subscription通常依赖WebSocket，但WebSocket在CDN和一层负载均衡后面临复杂配置。SSE是一种纯HTTP服务器推送技术，只需一个长连接。本教程展示如何用Django的StreamingHttpResponse配合asyncio队列，监听数据库变动或Redis Pub/Sub，当有事件发生时将GraphQL响应格式化为SSE事件流。客户端使用EventSource即可接收。同时解决SSE连接数与Django同步/异步模式的兼容问题。

# subscriptions/stream.py
from django.http import StreamingHttpResponse
import asyncio
def subscription_stream(request, channel):
    async def event_stream():
        queue = asyncio.Queue()
        # register listener
        async for message in listen(channel):
            await queue.put(f"data: {message}
")
            yield await queue.get()
    response = StreamingHttpResponse(event_stream(), content_type='text/event-stream')
    response['Cache-Control'] = 'no-cache'
    return response

← 上一节下一节 →

Ctrl+Enter

🚀 升级VIP
解锁全部课程+AI助手

🏆 学习排行

加载中...

📢 推荐

📚 大学视频课程 💼 实习招聘 🏠 便民信息

🔧 工具

💻 全屏编辑器 🔄 重置代码

📊 统计

📖 189 篇

✅ 0 完成

🔥 0 天