myaps_api/AGENTS.md

MyAPS API - 开发者指南

本文档为代码助手（如华为云CodeArts代码智能体）提供项目开发所需的关键信息，包括构建命令、代码风格指南和最佳实践。

项目概述

MyAPS API 是一个基于FastAPI的企业级数据操作平台，支持数据清洗、监控、WebSocket通信和定时任务调度。

技术栈：

• 后端框架: FastAPI (>=0.110.0)
• 异步服务器: Uvicorn (>=0.29.0)
• 数据库ORM: Tortoise-ORM (>=1.1.7)
• 数据验证: Pydantic (>=2.0.0)
• 前端: 原生HTML/JS + Bootstrap CSS

统一日志系统

项目使用统一的日志系统（globalobjects/logger/），替代原有的logger.py和logger_v2.py。

特性

• 异步写入，不阻塞业务线程
• 多目标输出（控制台、文件、数据库、WebSocket）
• 敏感信息自动脱敏
• 日期前缀文件轮转
• API完全向后兼容

使用方法

from globalobjects import logger

# 基本日志
logger.debug("调试信息")
logger.info("普通信息")
logger.warning("警告信息")
logger.error("错误信息")
logger.exception("异常信息")  # 自动捕获异常堆栈

# 业务便捷方法
logger.success("推送数据", "单号001", "共5条")
logger.fail("查询失败", "表A", "连接超时")
logger.start("同步任务", "账套A01")
logger.stop("同步任务", "账套A01")
logger.query("用户表", count=100)
logger.insert("日志表", count=50)

# 配置
logger.set_level("DEBUG")
logger.set_db_initialized(True)

环境变量配置

LOG_LEVEL=INFO              # 日志级别
LOG_DIR=logs               # 日志目录
TO_CONSOLE=true            # 输出到控制台
TO_FILE=true               # 输出到文件
TO_DATABASE=true           # 写入数据库
LOG_STACK_TRACE=false      # 是否启用调用栈追踪

旧版本备份

• logger_v1_backup.py - 原logger.py备份
• logger_v2_backup.py - 原logger_v2.py备份

构建和运行命令

开发环境运行

# 启动开发服务器（默认端口8001）
./scripts/dev_server.sh start

# 查看状态
./scripts/dev_server.sh status

# 停止服务
./scripts/dev_server.sh stop

# 重启服务
./scripts/dev_server.sh restart

# 查看日志
./scripts/dev_server.sh logs
./scripts/dev_server.sh logs -f  # 实时查看

# 清除Python缓存
./scripts/dev_server.sh clear_cache

直接运行

# 使用uvicorn
uvicorn main:app --host 0.0.0.0 --port 8000

# 带热重载
uvicorn main:app --host 0.0.0.0 --port 8000 --reload

Docker运行

# 使用Docker Compose（包含Redis）
docker-compose up -d

# 单独构建
docker build -t myaps-api .
docker run -p 8000:8000 myaps-api

生产部署

# 使用Gunicorn
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

# 使用配置文件
gunicorn -c scripts/deploy/gunicorn.conf.py main:app

测试命令

运行所有测试

# 在项目根目录运行
python -m pytest tests/ -v

# 或直接使用pytest
pytest tests/ -v

运行单个测试文件

# 测试统一数据入口
python -m pytest tests/test_unified_router.py -v

# 测试业务规则
python -m pytest tests/test_business_rules.py -v

# 测试日志系统
python -m pytest tests/test_logger.py -v

运行单个测试类

# 测试特定的测试类
python -m pytest tests/test_unified_router.py::TestIsStagingMode -v
python -m pytest tests/test_unified_router.py::TestMapStagingResponseToDirect -v

运行单个测试方法

# 测试特定的测试方法
python -m pytest tests/test_unified_router.py::TestIsStagingMode::test_staging_mode_with_reserved_value -v

测试标记

项目使用pytest标记：

• @pytest.mark.asyncio - 异步测试标记
• 无其他特殊标记

测试覆盖率（如需要）

# 安装pytest-cov
pip install pytest-cov

# 运行测试并计算覆盖率
pytest --cov=. --cov-report=html tests/

代码风格指南

Python代码风格

项目遵循PEP 8标准，但没有配置正式的lint工具。建议遵循以下约定：

导入顺序

# 1. 标准库导入
import os
import sys
from pathlib import Path

# 2. 第三方库导入
from fastapi import FastAPI, APIRouter
from pydantic import BaseModel
import pandas as pd

# 3. 本地应用导入
from core.app import create_app
from core.settings import PORT
from apps.io_api.routers import rt as io_rt

命名约定

• 变量/函数: snake_case
• 常量: UPPER_CASE_WITH_UNDERSCORES
• 类名: PascalCase
• 模块/包: snake_case
• 私有成员: _private_variable 或 __really_private

类型提示

使用Python类型提示：

from typing import Optional, List, Dict, Any

def process_data(
    data: List[Dict[str, Any]],
    config: Optional[Dict] = None
) -> Dict[str, Any]:
    """处理数据的函数文档字符串"""
    pass

FastAPI特定约定

1. 路由函数: 使用异步(async/await)
2. Pydantic模型: 使用BaseModel派生
3. API响应: 统一返回JSON格式
4. 错误处理: 使用FastAPI异常处理器

数据库相关

• 表名: snake_case (如 t_material_staging)
• 列名: PascalCase (如 MaterialNo)
• 注意: 数据库字段名大小写敏感

API设计规范

路由定义

from fastapi import APIRouter, Depends, HTTPException

router = APIRouter(prefix="/api", tags=["materials"])

@router.post("/upload", response_model=UploadResponse)
async def upload_materials(
    data: UploadRequest,
    db_name: str = Query(..., description="数据库名称")
):
    """上传物料数据"""
    pass

请求/响应模型

from pydantic import BaseModel
from typing import Optional

class UploadRequest(BaseModel):
    file_content: str
    file_name: str
    overwrite: bool = False

class UploadResponse(BaseModel):
    success: bool
    message: str
    data: Optional[Dict] = None
    meta: Optional[Dict] = None

API字段名

• POST字段名: 使用小写格式 (如 file_content, db_name)
• Query参数: 使用snake_case (如 db_name, overwrite)
• 路径参数: 使用snake_case (如 {material_id})

错误处理

HTTP状态码

• 200: 成功
• 400: 请求参数错误
• 401: 未授权
• 403: 禁止访问
• 404: 资源不存在
• 500: 服务器内部错误

错误响应格式

{
    "success": false,
    "error": {
        "code": "VALIDATION_ERROR",
        "message": "字段验证失败",
        "details": {
            "field_name": "具体错误信息"
        }
    }
}

日志记录

项目使用loguru或原生logging：

from globalobjects import logger

logger.info("操作成功")
logger.warning("警告信息")
logger.error("错误信息", exc_info=True)

项目结构

myaps_api/
├── apps/                    # 应用模块
│   ├── common/             # 通用模块（监控、帮助）
│   ├── data_opt/           # 数据操作模块
│   └── io_api/             # I/O API模块
├── core/                   # 核心组件
│   ├── app.py             # FastAPI应用工厂
│   ├── database.py        # 数据库配置
│   ├── settings.py        # 应用设置
│   └── middleware.py      # 中间件
├── globalobjects/          # 全局对象管理
├── scripts/               # 脚本目录
│   ├── dev_server.sh     # 开发服务器管理
│   ├── deploy/           # 部署配置
│   └── migrate/          # 数据库迁移
├── static/               # 静态文件
├── storage/             # 数据存储
├── tests/               # 测试文件
├── .env                 # 环境变量配置
├── main.py              # 应用入口
└── requirements.txt     # 依赖包

开发工作流

1. 环境设置

# 创建虚拟环境（如不存在）
python -m venv venv

# 激活虚拟环境
source venv/bin/activate  # Linux/Mac
venv\Scripts\activate     # Windows

# 安装依赖
pip install -r requirements.txt

2. 代码规范检查

项目目前没有配置lint工具，建议添加以下工具：

可选配置（如需要）：

# 安装格式化工具
pip install black isort ruff

# 代码格式化
black .
isort .
ruff check --fix

# 类型检查（如需）
pip install mypy
mypy .

3. 运行测试

# 运行所有测试
pytest tests/

# 运行特定模块测试
pytest tests/test_unified_router.py

# 运行带详细输出
pytest -v

# 运行并输出覆盖率
pytest --cov=. --cov-report=html

4. 代码提交前检查

1. 确保测试通过
2. 检查代码格式（如配置了lint工具）
3. 更新相关文档
4. 验证API接口

数据库操作

迁移脚本

# 数据库迁移
python scripts/migrate/auto_migrate.py

# Tortoise-ORM迁移
python scripts/migrate/migrate_with_tortoise.py

模型定义

from tortoise.models import Model
from tortoise import fields

class MaterialStaging(Model):
    """物料暂存表"""
    id = fields.IntField(pk=True)
    MaterialNo = fields.CharField(max_length=50)
    MaterialName = fields.CharField(max_length=255)
    # 注意：列名使用PascalCase
    
    class Meta:
        table = "t_material_staging"  # 表名使用snake_case

前端约定

HTML/JS约定

1. 样式: 使用Bootstrap 5.x
2. 图标: 使用Bootstrap Icons (bi-*)
3. 字体: 整个页面统一使用等宽字体
4. 布局: 偏好卡片布局而非列表布局
5. 颜色: 特定颜色使用十六进制值（如#ff9300）

错误提示

1. 布局: 横版布局，错误类型横向排列
2. 样式: 统一错误高亮提示样式
3. 交互: 手型光标用于可交互元素，插入光标用普通指针

验证规则显示

• 必填字段、枚举字段、业务规则块左右排列
• 内容在每个块内垂直分布
• 必填星号紧贴字段名右侧

部署配置

环境变量

复制.env.example（如存在）或创建.env文件：

# 应用配置
PORT=8000
HOST=0.0.0.0
LOG_LEVEL=INFO

# 数据库配置
DB_HOST=localhost
DB_PORT=3306
DB_USER=root
DB_PASSWORD=password
DB_NAME=myaps

# 功能开关
TURNON_BINLOG_LISTENER=False
TRUNON_SCHEDULER=False

# Staging模式配置
# 用于数据清洗模式的特殊数据库名称标识（默认--s）
STAGING_DB_NAME=--s

Gunicorn配置

生产环境使用Gunicorn多进程配置：

# scripts/deploy/gunicorn.conf.py
workers = 4
worker_class = "uvicorn.workers.UvicornWorker"
bind = "0.0.0.0:8000"

监控和调试

监控功能

• HTTP请求监控（HTTPMonitorMiddleware）
• 资源使用监控
• 数据库监控（binlog监听）
• 定时任务监控

调试技巧

1. 使用scripts/dev_server.sh logs -f实时查看日志
2. 检查logs/目录下的日志文件
3. 使用FastAPI自动生成的API文档（/docs和/redoc）

华为云CodeArts集成

项目已配置华为云CodeArts Doer：

• 配置文件位于.codeartsdoer/
• 技能配置在.codeartsdoer/skills/
• 规范配置在.codeartsdoer/specs/

注意事项

1. 代码质量: 确保新代码遵循项目现有模式
2. API兼容性: 修改API时保持向后兼容
3. 错误处理: 所有API端点应有适当的错误处理
4. 日志记录: 重要操作需记录日志
5. 测试覆盖: 新功能应包含单元测试
6. 配置管理: 避免硬编码，使用环境变量和配置对象
7. 数据库命名: 表名用snake_case，列名用PascalCase，注意大小写敏感

故障排除

常见问题

1. 端口占用: 修改.env中的PORT配置
2. 数据库连接失败: 检查数据库配置和环境变量
3. 依赖安装失败: 使用离线包或调整pip源
4. 权限问题: 确保logs/和storage/目录可写

Tortoise ORM 初始化竞态条件

问题描述： 启动时偶尔出现"Tortoise ORM 初始化超时"错误，即使数据库可连接。

根本原因： FastAPI启动时，register_tortoise异步初始化与lifespan检查存在竞态条件：

• PostgreSQL首次连接可能需要3-5秒
• 启动事件和lifespan并行执行
• 早期请求到达时，ORM可能未完全初始化

解决方案（已实施）： 使用事件驱动的智能等待机制（非硬编码等待）：

1. DatabaseInitManager (core/db_init_manager.py)

   • 事件驱动：初始化完成后主动通知等待者
   • 精确计时：记录实际初始化耗时
   • 状态追踪：监控初始化进度

2. 启动事件通知 (core/database.py)

   • @app.on_event("startup") 中标记初始化完成
   • 通知所有等待的协程

3. 智能等待 (core/lifespan.py, apps/common/utils/db_helpers.py)

   • 使用asyncio.Event而非轮询
   • 实际等待时间 = 数据库真实初始化时间
   • 超时保护：最多等待30秒

对比传统方案：

# ❌ 旧方案：硬编码轮询
for i in range(20):  # 固定等待10秒
    await asyncio.sleep(0.5)
    if Tortoise._inited:
        break

# ✅ 新方案：事件驱动
result = await db_init_manager.wait_for_init(max_wait=30.0)
# 实际等待时间 = 数据库初始化实际耗时（通常1-3秒）

相关配置：

• STAGING_DB_NAME: 清洗模式数据库标识（默认--s）
• THIS_DB_*: PostgreSQL连接配置

调试指南

如需调试指导（如断点设置），请参考：

1. VS Code调试配置在.vscode/
2. 使用Python内置pdb模块
3. 或使用更高级的调试器如debugpy

---

最后更新: 2026-05-21
维护者: MyAPS开发团队
版本: 1.0.0