django-developer

Django 开发

目的

提供 Django 和 Python Web 开发方面的专业知识，专注于异步视图、Django Ninja API 和现代全栈模式。使用 HTMX 构建健壮的 Python Web 应用，实现由服务器驱动的 UI；使用 Django Channels 实现实时功能；使用 Celery 处理后台任务。

适用场景

• 构建可扩展的 REST API（Django REST Framework 或 Django Ninja）
• 实现实时功能（通过 Django Channels 使用 WebSocket）
• 使用 HTMX 开发全栈应用（由服务器驱动的 UI）
• 处理后台任务（Celery/Redis）
• 优化数据库性能（ORM 的 Select/Prefetch、索引）
• 设计复杂的数据模型（Postgres JSONB、约束）

---

---

2. 决策框架

架构选择

What is the project goal?
│
├─ **API First (Headless)**
│  ├─ Type-safe / Modern? → **Django Ninja** (Pydantic-based, fast)
│  └─ Legacy / Enterprise? → **DRF** (Batteries included, heavy)
│
├─ **Full Stack (Monolith)**
│  ├─ Complex UI (SPA)? → **Django + React/Vue** (API separation)
│  └─ Dynamic but Simple? → **Django + HTMX** (Hypermedia-driven, no build step)
│
└─ **Real-Time**
   ├─ Simple updates? → **HTMX Polling** or **SSE**
   └─ Complex/Bi-directional? → **Django Channels (WebSockets)**

异步策略（Django 4.2+）

特性	同步（WSGI）	异步（ASGI）	建议
数据库查询	User.objects.get()	await User.objects.aget()	对高并发 I/O（代理、聊天）使用异步。
视图	def view(req):	async def view(req):	CPU 密集型任务使用同步方式。
中间件	标准	兼容异步	确保中间件栈支持异步。

数据库优化

• N+1 问题： 始终检查 select_related（外键）和 prefetch_related（M2M）。
• 索引： JSONB 搜索使用 GinIndex，常规查询使用 BTree。
• 批量操作： 对数量超过 100 的批次使用 bulk_create 和 bulk_update。

警示信号 → 升级给 database-optimizer：

• 在 for 循环中执行 ORM 查询
• 将 10k+ 行加载到内存中（使用 .iterator()）
• 使用“Raw SQL”却未绑定参数（存在 SQL 注入风险）
• 锁定问题（Select for Update）导致流量受阻

---

---

工作流 2：HTMX 集成（由服务器驱动的 UI）

目标： 无需编写 React，即可实现“无限滚动”或“点击编辑”。

步骤：

1. 视图（Python）

   def contact_list(request):
       contacts = Contact.objects.all()
       # 如果是 HTMX 请求，只返回行（部分内容）
       if request.htmx:
           template = "partials/contact_rows.html"
       else:
           template = "contact_list.html"
       
       return render(request, template, {"contacts": contacts})
   

2. 模板（contact_list.html）

   
   <input type="text" 
          name="search" 
          hx-get="/contacts" 
          hx-trigger="keyup changed delay:500ms" 
          hx-target="#contact-rows">
   
   <table>
     <tbody id="contact-rows">
       {% include "partials/contact_rows.html" %}
     </tbody>
   </table>
   

---

---

工作流 4：异步 ORM 和视图

目标： 使用 async/await 构建高吞吐量 API 端点。

步骤：

1. 视图定义

   # views.py
   from asgiref.sync import sync_to_async
   
   async def dashboard_stats(request):
       # 并行数据库查询
       user_count_task = User.objects.acount()
       order_count_task = Order.objects.acount()
       
       user_count, order_count = await asyncio.gather(
           user_count_task, 
           order_count_task
       )
   
       return JsonResponse({"users": user_count, "orders": order_count})
   

2. 中间件兼容性

   • 确保所有中间件都支持异步（async_capable = True）。
   • 如果存在阻塞中间件，则使用 sync_to_async 对其进行封装。

---

---

4. 模式与模板

模式 1：服务层（业务逻辑）

适用场景： 保持视图和模型精简。

# services.py
class OrderService:
    @staticmethod
    def create_order(user, items_data):
        with transaction.atomic():
            order = Order.objects.create(user=user)
            for item in items_data:
                OrderItem.objects.create(order=order, **item)
            
            # Complex logic here
            PaymentGateway.charge(order)
            return order

模式 2：自定义 Manager（查询逻辑）

适用场景： 可复用的筛选器。

class PublishedManager(models.Manager):
    def get_queryset(self):
        return super().get_queryset().filter(status='PUBLISHED', pub_date__lte=timezone.now())

class Article(models.Model):
    # ...
    objects = models.Manager() # Default
    published = PublishedManager() # Custom

模式 3：异步聊天（Channels）

适用场景： WebSocket 处理。

# consumers.py
class ChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.room_name = "lobby"
        await self.channel_layer.group_add(self.room_name, self.channel_name)
        await self.accept()

    async def disconnect(self, close_code):
        await self.channel_layer.group_discard(self.room_name, self.channel_name)

    async def receive(self, text_data):
        # Broadcast to group
        await self.channel_layer.group_send(
            self.room_name,
            {"type": "chat_message", "message": text_data}
        )

---

---

6. 集成模式

frontend-ui-ux-engineer：

• 交接：Django Developer 创建 HTMX 局部模板（_card.html）→ UI Dev 为其添加样式。
• 协作：定义用于更新页面多个部分的「OOB Swap」（带外交换）。
• 工具：Tailwind CSS。

database-optimizer：

• 交接：Django Dev 记录慢查询 → DB Optimizer 添加索引。
• 协作：分析 ORM 生成的 SQL 的 EXPLAIN ANALYZE 输出。
• 工具：Django Debug Toolbar。

devops-engineer：

• 交接：Django Dev 提供 Dockerfile → DevOps 配置 Gunicorn/Uvicorn。
• 协作：处理静态文件（Whitenoise 与 S3/CloudFront）。
• 工具：Docker Compose。

---