# ai-coding-test

**Repository Path**: liuwenlin/ai-coding-test

## Basic Information

- **Project Name**: ai-coding-test
- **Description**: 测试AI编码智能体
- **Primary Language**: Python
- **License**: Apache-2.0
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-09-11
- **Last Updated**: 2026-09-13

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# ai-coding-test

基于 **FastAPI + Python 3.14** 的全异步 Web API 脚手架，内置异步数据层（SQLAlchemy 2.x）、
异步缓存层（Redis / 内存降级）、可选消息队列（Kafka / RocketMQ）以及 Alembic 迁移、统一响应体、
请求上下文与日志、依赖探测健康探针。默认零外部依赖即可启动，方便按需接入业务模块。

## 特性

- **全异步链路**：FastAPI + SQLAlchemy 2.0 async + asyncmy/aiosqlite + redis-py。
- **分层架构**：`api -> service -> repository -> db`，`schemas` 负责出入参，`core` 负责横切能力。
- **优雅降级**：Redis / Kafka / RocketMQ 默认关闭；连接失败只记录 WARNING 并自动降级，不阻塞启动。
- **统一响应体**：`{"code": 0, "message": "ok", "data": ..., "trace_id": "..."}`，分页结构 `PageResult`。
- **请求可观测**：中间件注入 `X-Request-ID`、`X-Process-Time`，日志带 trace_id。
- **健康探针**：`/health/live`（存活）与 `/health/ready`（依赖探测：db / cache / kafka / rocketmq）。
- **可插拔缓存**：`CacheBackend` 抽象（`RedisCacheBackend` / `InMemoryCacheBackend` / `NullCacheBackend`）；Redis 不可用时自动降级为空实现，业务不受影响，并通过 `is_null` / `health()` 暴露降级与可用性状态。
- **消息队列抽象**：统一 `MessageEnvelope` / `MessagePublisher` / `MessageConsumer`，Kafka 与 RocketMQ 可切换。
- **Alembic 异步迁移**：`alembic/env.py` 从 `Settings` 读取数据库 URL。
- **测试齐全**：pytest + httpx ASGI 测试，15 个用例覆盖健康探针、CRUD 与 MQ。

## 目录结构

```
.
├── alembic/                     # 数据库迁移
│   ├── env.py                   # 异步迁移环境（读取 Settings.database.url）
│   └── versions/0001_create_items.py
├── app/
│   ├── main.py                  # create_app() 应用工厂 + lifespan
│   ├── config/settings.py       # pydantic-settings：app/database/redis/kafka/rocketmq
│   ├── core/                    # 横切能力
│   │   ├── context.py           # request_id contextvar
│   │   ├── logging.py           # setup_logging + JSON/控制台 Formatter
│   │   ├── response.py          # ApiResponse / PageResult / success / failure
│   │   ├── exceptions.py        # BizError 及子类
│   │   ├── handlers.py          # 全局异常处理器
│   │   └── middleware.py        # 纯 ASGI 请求上下文中间件
│   ├── db/                      # 数据层
│   │   ├── base.py              # Base + 命名约定 + TimestampMixin
│   │   ├── session.py           # init_engine / get_db_session
│   │   ├── models/item.py       # Item 模型
│   │   └── init_db.py           # 建表 + 种子数据
│   ├── cache/                   # 缓存层
│   │   ├── redis_client.py      # Redis 客户端生命周期
│   │   ├── backend.py           # Redis / Null CacheBackend
│   │   └── deps.py              # init_cache / get_cache / 内存降级实现
│   ├── mq/                      # 消息队列
│   │   ├── base.py              # MessagePublisher / MessageConsumer 抽象
│   │   ├── schemas.py           # MessageEnvelope
│   │   ├── kafka/               # producer / consumer / deps(KafkaManager)
│   │   └── rocketmq/            # compat(惰性导入) / producer / consumer / deps(RocketMQManager)
│   ├── schemas/                 # Pydantic 出入参
│   ├── repositories/            # 数据访问层
│   ├── services/                # 业务逻辑层（含缓存读写与失效）
│   ├── api/v1/endpoints/        # health / items / mq
│   └── lifecycle/manager.py     # 统一管理 db/cache/mq 生命周期
├── tests/                       # pytest 测试
├── Dockerfile
├── docker-compose.yml
└── pyproject.toml
```

## 快速开始

```bash
# 1. 创建虚拟环境
python -m venv .venv
.venv\Scripts\activate        # Windows
source .venv/bin/activate     # Linux / macOS

# 2. 安装（按需选择 extra）
pip install -e ".[dev]"                 # 基础 + 测试工具（SQLite 可用）
pip install -e ".[dev,mysql,kafka]"     # 追加 MySQL / Kafka
pip install -e ".[dev,rocketmq]"        # 追加 RocketMQ（需原生库）

# 3. 准备配置
copy .env.example .env        # Windows
cp .env.example .env          # Linux / macOS

# 4. 启动
uvicorn app.main:app --reload
```

启动后：

- 接口文档：<http://127.0.0.1:8000/docs>
- 存活探针：<http://127.0.0.1:8000/health/live>
- 依赖探针：<http://127.0.0.1:8000/health/ready>

`APP_ENV=local` 时会在启动阶段自动建表并写入示例数据，开箱即可调用业务接口。

## 环境变量

配置前缀与分组如下（详见 `.env.example`）。

### 应用
| 变量 | 默认值 | 说明 |
| --- | --- | --- |
| `APP_ENV` | `local` | 运行环境：`local` / `dev` / `test` / `staging` / `prod` |
| `APP_NAME` | `ai-coding-test` | 应用名，用于日志与文档标题 |
| `APP_DEBUG` | `false` | 调试模式 |
| `API_V1_PREFIX` | `/api/v1` | 业务路由前缀 |
| `LOG_LEVEL` | `INFO` | 日志级别：`CRITICAL` / `ERROR` / `WARNING` / `INFO` / `DEBUG` / `NOTSET` |
| `LOG_FORMAT` | `console` | 日志格式：`console` / `json` |
| `CORS_ORIGINS` | `*` | 允许跨域来源，**逗号分隔字符串**（如 `https://a.com,https://b.com`）；`*` 表示任意来源且会自动关闭凭据透传；留空表示不允许跨域 |

> 说明：监听地址与端口由启动命令决定（`uvicorn --host --port`），不属于应用配置项。

### 数据库
| 变量 | 默认值 | 说明 |
| --- | --- | --- |
| `DB_URL` | `sqlite+aiosqlite:///./data/app.db` | 异步驱动 URL，MySQL 用 `mysql+asyncmy://...` |
| `DB_ECHO` | `false` | 是否打印 SQL |
| `DB_POOL_SIZE` / `DB_MAX_OVERFLOW` | `10` / `20` | 连接池参数（SQLite 下忽略） |
| `DB_POOL_PRE_PING` | `true` | 取连接前先探活 |

### 缓存（Redis）
| 变量 | 默认值 | 说明 |
| --- | --- | --- |
| `CACHE_ENABLED` | `false` | 关闭时使用空实现（等价无缓存，业务不受影响） |
| `REDIS_URL` | `redis://127.0.0.1:6379/0` | Redis 连接串 |
| `REDIS_MAX_CONNECTIONS` | `50` | 连接池上限 |
| `CACHE_DEFAULT_TTL` | `300` | 默认过期秒数 |
| `CACHE_NULL_TTL` | `30` | 空值哨兵过期秒数，用于缓存穿透防护 |

### Kafka
| 变量 | 默认值 | 说明 |
| --- | --- | --- |
| `KAFKA_ENABLED` | `false` | 是否启用 |
| `KAFKA_BOOTSTRAP_SERVERS` | `127.0.0.1:9092` | broker 地址，逗号分隔 |
| `KAFKA_CLIENT_ID` | `ai-coding-test` | 客户端标识 |
| `KAFKA_GROUP_ID` | `ai-coding-test-group` | 消费组 |
| `KAFKA_CONSUME_TOPICS` | `app.demo.topic` | 订阅主题，逗号分隔 |
| `KAFKA_AUTO_COMMIT` | `false` | 是否自动提交位点 |

### RocketMQ
| 变量 | 默认值 | 说明 |
| --- | --- | --- |
| `ROCKETMQ_ENABLED` | `false` | 是否启用，需已安装原生客户端库 |
| `ROCKETMQ_NAME_SERVER` | `127.0.0.1:9876` | NameServer 地址 |
| `ROCKETMQ_PRODUCER_GROUP` | `ai-coding-test-producer` | 生产者组 |
| `ROCKETMQ_CONSUMER_GROUP` | `ai-coding-test-consumer` | 消费者组 |
| `ROCKETMQ_CONSUME_TOPICS` | `app.demo.topic` | 订阅主题，逗号分隔 |

## 接口说明

| 方法 | 路径 | 说明 |
| --- | --- | --- |
| GET | `/health/live` | 存活探测，返回 `{"status": "up"}` |
| GET | `/health/ready` | 依赖探测，返回 database / cache / kafka / rocketmq 状态 |
| POST | `/api/v1/items` | 创建条目 |
| GET | `/api/v1/items` | 分页查询条目 |
| GET | `/api/v1/items/{item_id}` | 查询单个条目（带缓存） |
| PUT | `/api/v1/items/{item_id}` | 全量更新条目并失效缓存 |
| DELETE | `/api/v1/items/{item_id}` | 删除条目并失效缓存 |
| POST | `/api/v1/mq/{broker}/messages` | 发送消息，`broker` 取值 `kafka` / `rocketmq` |

统一响应体：

```json
{
  "code": 0,
  "message": "ok",
  "data": { "...": "业务数据" },
  "trace_id": "f0d1c0f0b1c24b6c8e2f5c3d9a7b4e10"
}
```

## 数据库迁移

```bash
alembic revision --autogenerate -m "add xxx table"   # 生成迁移
alembic upgrade head                                # 升级
alembic downgrade -1                                # 回滚一步
alembic current                                     # 当前版本
```

`alembic/env.py` 直接读取 `Settings.database.url`，因此切换数据库只需调整 `DB_URL`。

## 测试

```bash
python -m pytest -v
python -m pytest --cov=app --cov-report=term-missing   # 如需覆盖率
```

测试通过 httpx ASGI transport 直接调用应用，使用 SQLite 临时库与内存缓存，不依赖 Redis / Kafka / RocketMQ。

## 使用 Docker Compose

```bash
docker compose up -d --build          # 只启动 api（SQLite）
docker compose --profile redis up -d  # 追加 Redis
docker compose --profile kafka up -d  # 追加 Kafka
docker compose --profile mysql up -d  # 追加 MySQL
docker compose --profile rocketmq up -d
```

接入中间件时，把对应 `*_ENABLED` 改为 `true` 并按需覆盖连接地址即可。

## 如何新增业务模块

以 `product` 为例：

1. `app/db/models/product.py` 定义模型，并在 `app/db/models/__init__.py` 导出。
2. 生成迁移：`alembic revision --autogenerate -m "add product"`，再 `alembic upgrade head`。
3. `app/schemas/product.py` 定义 `ProductCreate` / `ProductUpdate` / `ProductRead`。
4. `app/repositories/product_repository.py` 编写数据访问，只依赖 `AsyncSession`。
5. `app/services/product_service.py` 编写业务逻辑，缓存读写统一使用 `get_cache()`。
6. `app/api/v1/endpoints/products.py` 编写路由，依赖注入用 `Depends(get_db_session)` 与 `Depends(get_cache)`。
7. 在 `app/api/v1/router.py` 中挂载新路由。
8. 在 `tests/api/test_products.py` 补充接口测试。

消息发送统一构造 `MessageEnvelope`，通过 `POST /api/v1/mq/{broker}/messages` 或直接依赖
`get_kafka_producer()` / `get_rocketmq_producer()` 使用。

## 扩展 MQ

新增 MQ 实现时，实现 `app/mq/base.py` 中的 `MessagePublisher` 与 `MessageConsumer`，
放入 `app/mq/<broker>/`，提供 `deps.py` 暴露单例管理器，并在 `app/lifecycle/manager.py`
与 `app/api/v1/endpoints/mq.py` 中注册即可，业务侧无需改动。

## 已知说明

- Python 3.14 下 Kafka（`aiokafka`）与 Redis（`redis>=5.2`）均有官方支持；`rocketmq-client-python`
  官方元数据仅声明到 Python 3.7，因此设计为**可选 extra 且默认关闭**，并在 `app/mq/rocketmq/compat.py`
  中通过 `asyncio.to_thread` 包装同步客户端，避免阻塞事件循环。
- 生产环境请勿使用 `APP_ENV=local` 自动建表行为，改用 Alembic 迁移。

## License

本项目基于仓库内 `LICENSE` 文件所述的许可协议发布。