docs/v1.0/server/changelogs/2026-01-29-wechat-service-complete-implementation.md

微信服务完整实现与技术栈规范符合性修复

日期: 2026-01-29
类型: 重构 + 修复
影响范围: 微信登录服务、用户仓储、API 响应格式

变更概述

完成微信登录服务的技术栈规范符合性修复，包括：

1. 创建 BaseRepository 基础仓储类
2. 修复 WechatService 时间戳和事务处理
3. UserRepository 继承 BaseRepository
4. 统一 API 响应格式（添加 success 和 timestamp 字段）

详细变更

1. 创建 BaseRepository

文件: server/app/repositories/base_repository.py

class BaseRepository(Generic[T]):
    """基础仓储类，提供通用 CRUD 操作"""
    
    def __init__(self, session: AsyncSession, model: Type[T]):
        self.session = session
        self.model = model
    
    async def get_by_id(self, id: UUID) -> Optional[T]
    async def get_all(self, skip: int = 0, limit: int = 100) -> List[T]
    async def create(self, entity: T) -> T
    async def update(self, entity: T) -> T
    async def delete(self, id: UUID) -> bool

符合规范:

• ✅ 使用泛型 Generic[T] 提供类型安全
• ✅ 提供通用 CRUD 操作
• ✅ 所有方法使用 async/await

2. 修复 WechatService

文件: server/app/services/wechat_service.py

2.1 修复时间戳生成

# ❌ 修复前（第 318 行）
timestamp = int(datetime.utcnow().timestamp())

# ✅ 修复后
timestamp = int(datetime.now(timezone.utc).timestamp())

符合规范:

• ✅ 使用 datetime.now(timezone.utc) 替代废弃的 datetime.utcnow()
• ✅ 导入 timezone 模块

2.2 添加事务处理

# ❌ 修复前
async def _create_user_from_wechat(self, wechat_data: Dict[str, Any]) -> User:
    user = User(...)
    return await self.repository.create(user)

# ✅ 修复后
async def _create_user_from_wechat(self, wechat_data: Dict[str, Any]) -> User:
    async with self.session.begin():
        user = User(...)
        return await self.repository.create(user)

符合规范:

• ✅ Service 层使用 async with self.session.begin() 进行事务处理
• ✅ 确保用户创建的原子性

3. 修复 UserRepository

文件: server/app/repositories/user_repository.py

# ❌ 修复前
class UserRepository:
    def __init__(self, session: AsyncSession):
        self.session = session

# ✅ 修复后
from app.repositories.base_repository import BaseRepository

class UserRepository(BaseRepository[User]):
    def __init__(self, session: AsyncSession):
        super().__init__(session, User)
        self.session = session

符合规范:

• ✅ 继承 BaseRepository[User]
• ✅ 调用 super().__init__(session, User)
• ✅ 保留自定义方法（微信相关、会话管理）

4. 统一 API 响应格式

文件: server/app/schemas/response.py

4.1 ApiResponse 添加字段

class ApiResponse(BaseModel, Generic[T]):
    success: bool = Field(default=True, description="请求是否成功")
    code: int = Field(default=200, description="业务状态码")
    message: str = Field(default="Success", description="响应消息")
    data: Optional[T] = Field(default=None, description="响应数据")
    timestamp: datetime = Field(
        default_factory=lambda: datetime.now(timezone.utc),
        description="响应时间戳"
    )
    
    @field_serializer('timestamp')
    def serialize_timestamp(self, value: datetime) -> str:
        return value.isoformat()

4.2 ApiErrorResponse 添加字段

class ApiErrorResponse(BaseModel):
    success: bool = Field(default=False, description="请求是否成功")
    code: int = Field(..., description="业务错误码")
    message: str = Field(..., description="错误消息")
    data: Optional[Any] = Field(default=None, description="错误数据")
    errors: Optional[List[ErrorDetail]] = Field(default=None, description="详细错误列表")
    timestamp: datetime = Field(
        default_factory=lambda: datetime.now(timezone.utc),
        description="响应时间戳"
    )
    
    @field_serializer('timestamp')
    def serialize_timestamp(self, value: datetime) -> str:
        return value.isoformat()

符合规范:

• ✅ 添加 success 字段（布尔值）
• ✅ 添加 timestamp 字段（ISO 8601 格式）
• ✅ 使用 @field_serializer 装饰器序列化时间戳
• ✅ 使用 datetime.now(timezone.utc) 生成时间戳

4.3 更新便捷函数

def success_response(data: Any = None, message: str = "Success", code: int = 200) -> dict:
    return {
        "success": True,
        "code": code,
        "message": message,
        "data": data,
        "timestamp": datetime.now(timezone.utc).isoformat()
    }

def error_response(message: str, code: int = 400, data: Any = None, errors: Optional[List[dict]] = None) -> dict:
    return {
        "success": False,
        "code": code,
        "message": message,
        "data": data,
        "errors": errors,
        "timestamp": datetime.now(timezone.utc).isoformat()
    }

响应格式示例

成功响应

{
  "success": true,
  "code": 200,
  "message": "Success",
  "data": {
    "userId": "01936d8e-1234-7890-abcd-ef1234567890",
    "username": "user_1738166400_a1b2c3d4",
    "nickname": "微信用户",
    "avatarUrl": "https://wx.qlogo.cn/...",
    "wechatOpenid": "oABC123...",
    "wechatPlatform": "mp"
  },
  "timestamp": "2026-01-29T12:00:00Z"
}

错误响应

{
  "success": false,
  "code": 400,
  "message": "该微信账号已被其他用户绑定",
  "data": null,
  "errors": null,
  "timestamp": "2026-01-29T12:00:00Z"
}

技术栈符合性检查

✅ Model 层

• 使用 datetime.now(timezone.utc) 生成时间戳
• 导入 from datetime import datetime, timezone

✅ Repository 层

• 继承 BaseRepository[T]
• 所有方法使用 async/await
• 使用 datetime.now(timezone.utc) 更新时间戳

✅ Service 层

• 使用 async with self.session.begin() 进行事务处理
• 使用 datetime.now(timezone.utc) 生成时间戳

✅ Schema 层

• 使用 @field_serializer 装饰器序列化枚举和时间戳
• 响应 Schema 包含 success 和 timestamp 字段

✅ API 层

• 使用 ApiResponse[T] 统一响应格式
• 所有响应包含 success 和 timestamp 字段

影响范围

直接影响

• server/app/repositories/base_repository.py - 新建
• server/app/repositories/user_repository.py - 修改（继承 BaseRepository）
• server/app/services/wechat_service.py - 修改（时间戳 + 事务）
• server/app/schemas/response.py - 修改（添加字段）

间接影响

• 所有使用 ApiResponse 的 API 路由（响应格式变更）
• 前端需要适配新的响应格式（添加 success 和 timestamp 字段）

后续任务

1. 创建测试用例:

   • server/tests/unit/test_wechat_repository.py
   • server/tests/unit/test_wechat_service.py
   • server/tests/integration/test_wechat_api.py

2. 检查数据库迁移:

   • 确认微信字段（wechat_openid, wechat_unionid, wechat_platform）已迁移

3. 前端适配:

   • 更新 API 响应类型定义
   • 适配新的响应格式（success 和 timestamp 字段）

4. 文档更新:

   • 更新 API 文档（Swagger/OpenAPI）
   • 更新前端 API 调用示例

参考文档

• 微信服务设计文档
• 技术栈规范
• Jointo Tech Stack Skill

版本信息

• 文档版本: v2.2
• 实现日期: 2026-01-29
• 实现人员: Kiro AI Assistant