开发指南

基本格式

Python 版本为 3.11+
使用 4 个空格缩进，不使用 Tab
行长度上限 120 字符
使用 UTF-8 编码
注释和文档字符串优先使用英文，业务说明可使用中文

async def get_device(hub, request: web.Request) -> web.Response:
    """GET /api/v1/devices/{device_id}"""
    did = request.match_info["device_id"]
    session = hub.get_device(did)
    if not session:
        return web.json_response({"success": False, "error": "device not found"}, status=404)
    return web.json_response({"success": True, "device": session.to_dict()})

命名规范

对象	规则	示例
模块/文件	`snake_case`	`api_device.py`、`jwt_utils.py`
类	`PascalCase`	`Hub`、`DeviceSession`、`MockDevice`
函数/方法	`snake_case`	`proxy_request()`、`get_device_list()`
私有函数/方法	`_snake_case`	`_on_register()`、`_load_bindings()`
变量	`snake_case`	`device_id`、`last_heartbeat`
常量	`UPPER_SNAKE_CASE`	`DEVICE_TOKEN`、`SYSTEM_IMAGE_SIZE`
环境变量	`UPPER_SNAKE_CASE`	`CLOUD_WS_HOST`、`CLOUD_JWT_SECRET`

导入顺序

标准库、第三方库、本地模块分组，每组之间空一行：

import asyncio
import json
import logging
import os

import websockets
from aiohttp import web

import config
from gateway.hub import Hub

类型注解

所有公共函数和 API 处理器应添加类型注解：

async def proxy_request(
    hub: Hub,
    device_id: str,
    method: str,
    path: str,
    params: dict = None,
    body=None,
    timeout: float = None,
) -> dict:
    ...

异步模式

使用 async/await 而非回调
HTTP 处理器统一签名为 async def handler(hub, request: web.Request) -> web.Response
不在协程中执行阻塞式同步 I/O；必须用时通过 asyncio.to_thread() 或 run_in_executor() 处理
WebSocket 消息处理使用 asyncio.ensure_future() 调度无需等待的异步发送

# 正确：异步等待代理结果
result = await proxy_request(hub, did, method, path)

# 正确：不等待的发送任务
asyncio.ensure_future(self.send_to(session, "register_ack", {"success": True}))

路由注册

所有路由在 server.py 的 create_app() 中集中注册，设备路由通过 _with_hub() 装饰器注入 Hub：

routes = [
    ("GET", "/api/v1/devices", api_device.get_devices),
    ("POST", "/api/v1/devices/{device_id}/hardware/move", api_hardware.move),
]
for method, path, handler in routes:
    app.router.add_route(method, path, _with_hub(handler))

错误处理

API 响应统一使用 {"success": bool, "error": str, ...} 格式
成功响应使用 {"success": True, ...} 携带业务数据
错误响应使用 {"success": False, "error": "描述"}，配合合适的 HTTP 状态码

# 成功
return web.json_response({"success": True, "devices": devices})

# 404
return web.json_response({"success": False, "error": "device not found"}, status=404)

# 400
return web.json_response({"success": False, "error": "invalid request body"}, status=400)

模块边界

gateway/：Hub 和 Proxy，不包含业务逻辑
routes/：API 处理器，通过 Hub 和 Proxy 与设备交互
auth/：认证和鉴权中间件
tools/：开发和测试工具，不部署到生产环境
firmware/：固件存储，运行时写入

安全要求

不在日志中记录 JWT 令牌、设备 Token 或密码
所有设备 ID 由客户端提供，服务端不做信任假设
env 文件不提交到代码仓库（已加入 .gitignore）
生产配置校验在启动时执行，不符合要求拒绝启动
API 路由通过中间件进行认证检查，白名单路由例外

提交前检查清单

未引入新的阻塞式同步 I/O
新 API 路由正确注册且权限配置合理
错误响应格式统一为 {"success": False, "error": "..."}
日志不泄露密钥、Token 或敏感配置
config.validate_production_config() 校验通过
模拟设备和冒烟测试通过