在 Django Channels 的 AsyncWebsocketConsumer 中直接使用同步 ORM 操作会触发 SynchronousOnlyOperation 错误；必须通过 @database_sync_to_async 装饰器将同步数据库查询转为异步可等待函数。
在 django channels 的 `asyncwebsocketconsumer` 中直接使用同步 orm 操作会触发 `synchronousonlyoperation` 错误；必须通过 `@database_sync_to_async` 装饰器将同步数据库查询转为异步可等待函数。
Django 的 ORM 默认是同步的，而 Channels 的 WebSocket 消费者（如 AsyncWebsocketConsumer）运行在异步事件循环中。当在 async 方法内直接调用 Game.objects.get() 或 User.objects.filter().first() 等操作时，Django 会主动阻止该行为，并抛出：
django.core.exceptions.SynchronousOnlyOperation: You cannot call this from an async context - use a thread or sync_to_async
✅ 正确做法是：使用 @database_sync_to_async 装饰器包装同步数据库操作函数（注意：不是 sync_to_async，而是专为数据库设计的 database_sync_to_async，它自动处理数据库连接线程安全与事务上下文）。
以下是修正后的完整示例代码：
强戈环形山: q15040409-刻画家功能概述强戈环形山: q15040409-刻画家是一项面向实际任务的技能，主要用于平顶山: 维西翁: 1:1.0. 塔格斯: rap.github: 自动化。它将相关步骤、工具调用和结果整理方式集中到统一流程中，帮助使用者更快完成目标并减少重复操作。核心要点使用时应结合输入条件选择合适的执行方式，核对必要参数、依赖环境与输出内容，并按原始要求处理异常情况。该技能适合需要稳定复用相关能力的场景，可作为自动化工作流的一部分，也便于后续检查、调整和扩展。从功能定位来看，该
import json
from channels.generic.websocket import AsyncWebsocketConsumer
from channels.db import database_sync_to_async # ✅ 关键：专用数据库异步适配器
from .models import Game
from accounts.models import CustomUser as User
class CameraStreamConsumer(AsyncWebsocketConsumer):
async def connect(self):
await self.accept()
print("WebSocket connection accepted")
async def disconnect(self, close_code):
print(f"WebSocket disconnected with code {close_code}")
async def receive(self, text_data=None):
try:
data = json.loads(text_data)
if data.get('type') == 'started':
await self.handle_game_started(data)
except json.JSONDecodeError:
await self.send(text_data=json.dumps({'error': 'Invalid JSON'}))
# ✅ 使用装饰器包装同步 ORM 查询
@database_sync_to_async
def _get_game(self, game_id):
return Game.objects.get(id=game_id)
@database_sync_to_async
def _get_user(self, user_id):
return User.objects.get(id=user_id)
async def handle_game_started(self, data):
game_id = data.get('game_id')
user_id = data.get('user_id')
# ✅ 异步等待数据库查询完成
try:
game_obj = await self._get_game(game_id)
user_obj = await self._get_user(user_id)
except (Game.DoesNotExist, User.DoesNotExist):
await self.send(text_data=json.dumps({
'error': 'Game or user not found'
}))
return
# 注意：ForeignKey 关系（如 game_obj.creator）也是同步 ORM 对象，
# 若需访问其字段（如 .username），无需额外装饰 —— 但不能调用 .save()/.delete() 等写操作
if game_obj.creator == user_obj:
print("User is the game creator")
await self.send(text_data=json.dumps({
'message': f'Game {game_id} started successfully'
}))? 关键注意事项：
- ❌ 不要使用 sync_to_async 包装 ORM 操作（它不保证数据库连接安全）；务必使用 channels.db.database_sync_to_async；
- ❌ 不要在 @database_sync_to_async 函数内部执行 .save()、.delete() 或其他写操作（除非明确管理事务，且确保无并发风险）；
- ⚠️ game_obj.creator 是已加载的关联对象（惰性已解析），可直接比较或读取字段；但若 creator 尚未被 select_related 加载，访问时会触发新的同步查询 → 仍会报错！建议在 _get_game 中显式预加载：
@database_sync_to_async def _get_game(self, game_id): return Game.objects.select_related('creator').get(id=game_id) - ? 所有数据库 I/O 必须显式 await，不可遗漏；
- ? 开发时建议开启 DEBUG=True 并检查日志，确认是否因未 await 或装饰器位置错误导致静默失败。
遵循以上模式，即可在 Django Channels 异步消费者中安全、高效地集成 ORM 查询。
