生成代码工程

This commit is contained in:
root 2026-05-06 13:42:33 +08:00
parent c0263d7822
commit 995caf4847
9 changed files with 5408 additions and 1 deletions

View File

@ -1 +1,59 @@
任务自主规划与执行软件
# Todo Manager - 任务管理器示例工程
基于 FastAPI 的单体示例工程,使用内存假数据完成任务的增删改查核心流程。
## 技术栈
- Python 3.11+
- FastAPI (Web 框架)
- Pydantic (数据校验)
- Uvicorn (ASGI 服务器)
- Pytest + HTTPX (测试)
## 安装依赖
```bash
cd codegen-runs/codegen_8a0046bfd6e148b3ab096039e55d5f72
pip install -r requirements.txt
```
## 启动服务
```bash
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
```
服务启动后访问:
- API 文档: http://127.0.0.1:8000/docs
- 替代文档: http://127.0.0.1:8000/redoc
## API 接口
| 方法 | 路径 | 说明 |
|------|------|------|
| GET | /api/todos | 获取所有任务 |
| GET | /api/todos/{id} | 根据 ID 获取任务 |
| POST | /api/todos | 创建新任务 |
| PUT | /api/todos/{id} | 更新任务 |
| DELETE | /api/todos/{id} | 删除任务 |
### 请求/响应示例
**创建任务:**
```bash
curl -X POST "http://127.0.0.1:8000/api/todos" \
-H "Content-Type: application/json" \
-d '{"title": "学习 FastAPI", "description": "完成 Todo 示例项目"}'
```
**获取所有任务:**
```bash
curl "http://127.0.0.1:8000/api/todos"
```
## 运行测试
```bash
cd codegen-runs/codegen_8a0046bfd6e148b3ab096039e55d5f72
pytest tests/ -v
```

3
app/__init__.py Normal file
View File

@ -0,0 +1,3 @@
"""Todo Manager 应用包。"""
__version__ = "1.0.0"

108
app/main.py Normal file
View File

@ -0,0 +1,108 @@
"""FastAPI 主应用模块。
定义 Web 路由和启动入口提供 RESTful 风格的任务管理 API
"""
from fastapi import FastAPI, HTTPException
from app.models import TodoCreate, TodoUpdate, TodoResponse, TodoListResponse
from app.services import todo_service
app = FastAPI(
title="Todo Manager API",
description="一个简单的任务管理器示例 API使用内存假数据。",
version="1.0.0",
)
@app.get("/api/todos", response_model=TodoListResponse, tags=["Todos"])
def list_todos():
"""获取所有任务列表。
Returns:
包含任务列表和总数的响应
"""
items, total = todo_service.list_todos()
return TodoListResponse(data=items, total=total)
@app.get("/api/todos/{todo_id}", response_model=TodoResponse, tags=["Todos"])
def get_todo(todo_id: int):
"""根据 ID 获取单个任务。
Args:
todo_id: 任务唯一标识
Returns:
包含任务详细信息的响应
Raises:
HTTPException: 任务不存在时返回 404
"""
item = todo_service.get_todo(todo_id)
if item is None:
raise HTTPException(status_code=404, detail=f"Todo with id {todo_id} not found")
return TodoResponse(data=item)
@app.post("/api/todos", response_model=TodoResponse, status_code=201, tags=["Todos"])
def create_todo(data: TodoCreate):
"""创建新任务。
Args:
data: 包含标题和描述的任务创建请求
Returns:
包含新创建任务信息的响应
"""
item = todo_service.create_todo(data)
return TodoResponse(code=201, message="created", data=item)
@app.put("/api/todos/{todo_id}", response_model=TodoResponse, tags=["Todos"])
def update_todo(todo_id: int, data: TodoUpdate):
"""更新已有任务。
Args:
todo_id: 要更新的任务 ID
data: 包含需要更新字段的请求体
Returns:
包含更新后任务信息的响应
Raises:
HTTPException: 任务不存在时返回 404
"""
item = todo_service.update_todo(todo_id, data)
if item is None:
raise HTTPException(status_code=404, detail=f"Todo with id {todo_id} not found")
return TodoResponse(data=item)
@app.delete("/api/todos/{todo_id}", response_model=TodoResponse, tags=["Todos"])
def delete_todo(todo_id: int):
"""删除指定任务。
Args:
todo_id: 要删除的任务 ID
Returns:
删除成功的确认响应
Raises:
HTTPException: 任务不存在时返回 404
"""
deleted = todo_service.delete_todo(todo_id)
if not deleted:
raise HTTPException(status_code=404, detail=f"Todo with id {todo_id} not found")
return TodoResponse(message="deleted")
@app.get("/", tags=["Root"])
def root():
"""根路径,返回 API 基础信息。"""
return {
"name": "Todo Manager API",
"version": "1.0.0",
"docs": "/docs",
}

57
app/models.py Normal file
View File

@ -0,0 +1,57 @@
"""任务模型定义模块。
定义任务相关的 Pydantic 数据模型用于请求校验与响应序列化
"""
from datetime import datetime, timezone
from typing import Optional
from pydantic import BaseModel, Field
class TodoItem(BaseModel):
"""任务项模型,对应数据库中的一条任务记录。"""
id: int = Field(..., description="任务唯一标识")
title: str = Field(..., min_length=1, max_length=200, description="任务标题")
description: str = Field(default="", max_length=1000, description="任务描述")
completed: bool = Field(default=False, description="是否完成")
created_at: datetime = Field(
default_factory=lambda: datetime.now(timezone.utc),
description="创建时间",
)
updated_at: datetime = Field(
default_factory=lambda: datetime.now(timezone.utc),
description="更新时间",
)
class TodoCreate(BaseModel):
"""创建任务请求模型。"""
title: str = Field(..., min_length=1, max_length=200, description="任务标题")
description: str = Field(default="", max_length=1000, description="任务描述")
class TodoUpdate(BaseModel):
"""更新任务请求模型,所有字段可选。"""
title: Optional[str] = Field(None, min_length=1, max_length=200, description="任务标题")
description: Optional[str] = Field(None, max_length=1000, description="任务描述")
completed: Optional[bool] = Field(None, description="是否完成")
class TodoResponse(BaseModel):
"""任务响应模型,包含标准返回结构。"""
code: int = Field(default=200, description="状态码")
message: str = Field(default="success", description="提示信息")
data: Optional[TodoItem] = Field(None, description="任务数据")
class TodoListResponse(BaseModel):
"""任务列表响应模型。"""
code: int = Field(default=200, description="状态码")
message: str = Field(default="success", description="提示信息")
data: list[TodoItem] = Field(default_factory=list, description="任务列表")
total: int = Field(default=0, description="任务总数")

122
app/services.py Normal file
View File

@ -0,0 +1,122 @@
"""业务逻辑层模块。
提供任务管理的核心业务逻辑使用内存存储模拟数据库操作
"""
from datetime import datetime, timezone
from typing import Optional
from app.models import TodoItem, TodoCreate, TodoUpdate
class TodoService:
"""任务管理服务类。
使用内存字典存储任务数据提供完整的增删改查操作
线程不安全仅适用于单线程演示场景
"""
def __init__(self) -> None:
"""初始化服务,创建空的内存存储和 ID 计数器。"""
self._store: dict[int, TodoItem] = {}
self._next_id: int = 1
def _generate_id(self) -> int:
"""生成下一个可用的任务 ID。
Returns:
自增的任务 ID
"""
current = self._next_id
self._next_id += 1
return current
def _now(self) -> datetime:
"""获取当前 UTC 时间。
Returns:
当前 UTC 时间戳
"""
return datetime.now(timezone.utc)
def list_todos(self) -> tuple[list[TodoItem], int]:
"""获取所有任务列表。
Returns:
(任务列表, 总数) 的元组
"""
items = list(self._store.values())
return items, len(items)
def get_todo(self, todo_id: int) -> Optional[TodoItem]:
"""根据 ID 获取单个任务。
Args:
todo_id: 任务 ID
Returns:
如果找到返回 TodoItem否则返回 None
"""
return self._store.get(todo_id)
def create_todo(self, data: TodoCreate) -> TodoItem:
"""创建新任务。
Args:
data: 创建任务的请求数据
Returns:
创建成功的 TodoItem
"""
now = self._now()
item = TodoItem(
id=self._generate_id(),
title=data.title,
description=data.description,
completed=False,
created_at=now,
updated_at=now,
)
self._store[item.id] = item
return item
def update_todo(self, todo_id: int, data: TodoUpdate) -> Optional[TodoItem]:
"""更新已有任务。
Args:
todo_id: 要更新的任务 ID
data: 包含更新字段的请求数据
Returns:
更新后的 TodoItem如果任务不存在返回 None
"""
existing = self._store.get(todo_id)
if existing is None:
return None
update_dict = data.model_dump(exclude_none=True)
if not update_dict:
return existing
updated_item = existing.model_copy(update=update_dict)
updated_item.updated_at = self._now()
self._store[todo_id] = updated_item
return updated_item
def delete_todo(self, todo_id: int) -> bool:
"""删除指定任务。
Args:
todo_id: 要删除的任务 ID
Returns:
删除成功返回 True任务不存在返回 False
"""
if todo_id in self._store:
del self._store[todo_id]
return True
return False
# 全局单例服务实例
todo_service = TodoService()

4878
events.ndjson Normal file

File diff suppressed because one or more lines are too long

5
requirements.txt Normal file
View File

@ -0,0 +1,5 @@
fastapi==0.104.1
uvicorn==0.24.0
pydantic==2.5.2
httpx==0.25.2
pytest==7.4.3

1
tests/__init__.py Normal file
View File

@ -0,0 +1 @@
"""测试包。"""

175
tests/test_basic.py Normal file
View File

@ -0,0 +1,175 @@
"""基础测试模块。
测试业务逻辑层 (TodoService) API (FastAPI 路由) 的核心功能
"""
import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app
from app.services import TodoService
from app.models import TodoCreate, TodoUpdate
# ============================================================
# 单元测试 — TodoService 业务逻辑
# ============================================================
class TestTodoService:
"""TodoService 的单元测试。"""
@pytest.fixture
def service(self):
"""每个测试使用全新的 TodoService 实例。"""
return TodoService()
def test_create_todo(self, service):
"""测试创建任务:应返回带 ID 和默认字段的有效任务。"""
data = TodoCreate(title="测试任务", description="测试描述")
item = service.create_todo(data)
assert item.id == 1
assert item.title == "测试任务"
assert item.description == "测试描述"
assert item.completed is False
def test_list_todos_empty(self, service):
"""测试空列表:新建服务应返回空列表。"""
items, total = service.list_todos()
assert items == []
assert total == 0
def test_list_todos_with_data(self, service):
"""测试列表:创建多个任务后应能正确返回。"""
service.create_todo(TodoCreate(title="任务1"))
service.create_todo(TodoCreate(title="任务2"))
items, total = service.list_todos()
assert total == 2
assert len(items) == 2
def test_get_todo_by_id(self, service):
"""测试根据 ID 获取任务。"""
created = service.create_todo(TodoCreate(title="查找测试"))
fetched = service.get_todo(created.id)
assert fetched is not None
assert fetched.title == "查找测试"
def test_get_todo_not_found(self, service):
"""测试获取不存在的任务应返回 None。"""
assert service.get_todo(999) is None
def test_update_todo(self, service):
"""测试更新任务:应正确修改标题和完成状态。"""
created = service.create_todo(TodoCreate(title="原标题"))
updated = service.update_todo(
created.id,
TodoUpdate(title="新标题", completed=True),
)
assert updated is not None
assert updated.title == "新标题"
assert updated.completed is True
def test_update_todo_not_found(self, service):
"""测试更新不存在的任务应返回 None。"""
result = service.update_todo(999, TodoUpdate(title=""))
assert result is None
def test_delete_todo(self, service):
"""测试删除任务:删除后应无法再获取。"""
created = service.create_todo(TodoCreate(title="待删除"))
deleted = service.delete_todo(created.id)
assert deleted is True
assert service.get_todo(created.id) is None
def test_delete_todo_not_found(self, service):
"""测试删除不存在的任务应返回 False。"""
result = service.delete_todo(999)
assert result is False
# ============================================================
# 集成测试 — FastAPI 路由
# ============================================================
@pytest.fixture
def client():
"""创建测试用的 AsyncClient绑定到 FastAPI 应用。"""
transport = ASGITransport(app=app)
return AsyncClient(transport=transport, base_url="http://test")
@pytest.mark.asyncio
async def test_api_root(client):
"""测试根路径返回 API 信息。"""
response = await client.get("/")
assert response.status_code == 200
data = response.json()
assert "name" in data
assert "version" in data
@pytest.mark.asyncio
async def test_api_create_todo(client):
"""测试 POST /api/todos 创建任务。"""
response = await client.post("/api/todos", json={
"title": "API 测试任务",
"description": "通过 API 创建",
})
assert response.status_code == 201
data = response.json()
assert data["code"] == 201
assert data["data"]["title"] == "API 测试任务"
@pytest.mark.asyncio
async def test_api_list_todos(client):
"""测试 GET /api/todos 获取任务列表。"""
# 先创建一条数据
await client.post("/api/todos", json={"title": "列表测试"})
response = await client.get("/api/todos")
assert response.status_code == 200
data = response.json()
assert data["code"] == 200
assert data["total"] >= 1
@pytest.mark.asyncio
async def test_api_get_todo(client):
"""测试 GET /api/todos/{id} 获取单个任务。"""
create_resp = await client.post("/api/todos", json={"title": "单个查询"})
todo_id = create_resp.json()["data"]["id"]
response = await client.get(f"/api/todos/{todo_id}")
assert response.status_code == 200
assert response.json()["data"]["title"] == "单个查询"
@pytest.mark.asyncio
async def test_api_get_todo_not_found(client):
"""测试 GET /api/todos/{id} 不存在的任务应返回 404。"""
response = await client.get("/api/todos/99999")
assert response.status_code == 404
@pytest.mark.asyncio
async def test_api_update_todo(client):
"""测试 PUT /api/todos/{id} 更新任务。"""
create_resp = await client.post("/api/todos", json={"title": "待更新"})
todo_id = create_resp.json()["data"]["id"]
response = await client.put(f"/api/todos/{todo_id}", json={
"title": "已更新",
"completed": True,
})
assert response.status_code == 200
assert response.json()["data"]["title"] == "已更新"
assert response.json()["data"]["completed"] is True
@pytest.mark.asyncio
async def test_api_delete_todo(client):
"""测试 DELETE /api/todos/{id} 删除任务。"""
create_resp = await client.post("/api/todos", json={"title": "待删除"})
todo_id = create_resp.json()["data"]["id"]
response = await client.delete(f"/api/todos/{todo_id}")
assert response.status_code == 200
# 再次查询应返回 404
get_resp = await client.get(f"/api/todos/{todo_id}")
assert get_resp.status_code == 404