简单版增删改查
修改 pyproject.toml
bash
[project]
name = "demo1"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.14"
dependencies = [
"fastapi[standard]>=0.162.0",
"pydantic>=2,<3",
"pydantic-settings>=2.16.2",
"uvicorn>=0.54.0",
]
[project.scripts]
dev = "main:dev"
prod = "main:prod"
test = "main:test"
[build-system]
requires = ["setuptools>=75"]
build-backend = "setuptools.build_meta"
[tool.setuptools]
py-modules = ["main"]
[tool.setuptools.packages.find]
where = ["."]
include = ["core*", "modules*"]
namespaces = true数据源
- core/datasources.py
python
book = [
{
"id": 1,
"name": "Python 3.11 入门",
"author": "张三",
"price": 99.0,
"description": "面向初学者讲解 Python 基础语法、常用数据类型和简单项目实践。",
},
{
"id": 2,
"name": "Python 3.11 进阶",
"author": "李四",
"price": 199.0,
"description": "深入介绍面向对象、异常处理、并发编程和 Python 高级特性。",
},
{
"id": 3,
"name": "FastAPI 实战",
"author": "王五",
"price": 128.0,
"description": "通过完整案例学习 FastAPI 路由、依赖注入、数据校验和接口部署。",
},
{
"id": 4,
"name": "数据结构与算法",
"author": "赵六",
"price": 118.0,
"description": "系统讲解常见数据结构、经典算法及其时间复杂度分析。",
},
{
"id": 5,
"name": "MySQL 数据库入门",
"author": "陈明",
"price": 89.0,
"description": "涵盖数据库设计、SQL 查询、索引优化和事务管理等基础知识。",
},
{
"id": 6,
"name": "Web 前端开发基础",
"author": "刘洋",
"price": 108.0,
"description": "介绍 HTML、CSS 和 JavaScript,帮助读者完成基础网页开发。",
},
{
"id": 7,
"name": "JavaScript 高级程序设计",
"author": "周杰",
"price": 139.0,
"description": "深入讲解 JavaScript 作用域、异步编程、原型链和模块化开发。",
},
{
"id": 8,
"name": "Linux 系统管理指南",
"author": "吴磊",
"price": 119.0,
"description": "介绍 Linux 常用命令、用户权限、进程管理和服务配置。",
},
{
"id": 9,
"name": "Docker 容器化实践",
"author": "郑浩",
"price": 129.0,
"description": "从镜像和容器基础开始,讲解应用容器化及 Docker Compose 实践。",
},
{
"id": 10,
"name": "Git 版本控制实战",
"author": "孙悦",
"price": 79.0,
"description": "通过实际工作流程掌握 Git 提交、分支、合并和冲突处理。",
},
{
"id": 11,
"name": "RESTful API 设计指南",
"author": "何静",
"price": 96.0,
"description": "讲解资源建模、状态码、接口版本管理和 API 文档设计规范。",
},
{
"id": 12,
"name": "Redis 入门与实践",
"author": "郭涛",
"price": 109.0,
"description": "介绍 Redis 数据类型、持久化、缓存策略和常见应用场景。",
},
{
"id": 13,
"name": "软件测试基础",
"author": "林雪",
"price": 88.0,
"description": "覆盖测试用例设计、单元测试、集成测试和基本质量保障方法。",
},
{
"id": 14,
"name": "微服务架构设计",
"author": "黄宇",
"price": 168.0,
"description": "分析微服务拆分、服务通信、配置管理和系统可观测性设计。",
},
{
"id": 15,
"name": "人工智能导论",
"author": "徐晨",
"price": 158.0,
"description": "介绍机器学习、神经网络和人工智能应用的基本概念与发展。",
},
]modules模块之Book模块
modules/book/schemas.py
py
from pydantic import BaseModel
class BookBase(BaseModel):
name: str
author: str
price: float
description: str
class BookCreate(BookBase):
pass
class BookUpdate(BookBase):
pass
class BookResponse(BookBase):
id: intmodules/book/router.py
py
from fastapi import APIRouter, HTTPException, status
from modules.book import service
from modules.book.schemas import BookCreate, BookResponse, BookUpdate
# 图书接口统一使用 /books 前缀,并在接口文档中归入 books 分组。
router = APIRouter(prefix="/books", tags=["books"])
# 查询全部图书,响应数据按照 BookResponse 模型进行序列化。
@router.get("", response_model=list[BookResponse])
def list_books():
return service.list_books()
# 新增图书,请求体不包含 ID,由服务层自动生成;成功后返回 201。
@router.post("", response_model=BookResponse, status_code=status.HTTP_201_CREATED)
def create_book(book_data: BookCreate):
return service.create_book(book_data)
# 根据路径中的图书 ID 查询详情,图书不存在时返回 404。
@router.get("/{book_id}", response_model=BookResponse)
def get_book(book_id: int):
existing_book = service.get_book(book_id)
if existing_book is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="图书不存在",
)
return existing_book
# 全量更新指定图书,保留原 ID;图书不存在时返回 404。
@router.put("/{book_id}", response_model=BookResponse)
def update_book(book_id: int, book_data: BookUpdate):
existing_book = service.update_book(book_id, book_data)
if existing_book is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="图书不存在",
)
return existing_book
# 删除指定图书,成功后返回 204 且无响应体;图书不存在时返回 404。
@router.delete("/{book_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_book(book_id: int) -> None:
if not service.delete_book(book_id):
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="图书不存在",
)modules/book/service.py
py
from core.datasources import book
from modules.book.schemas import BookCreate, BookUpdate
def list_books() -> list[dict]:
return book
def get_book(book_id: int) -> dict | None:
return next((item for item in book if item["id"] == book_id), None)
def create_book(book_data: BookCreate) -> dict:
new_book = {
"id": max((item["id"] for item in book), default=0) + 1,
**book_data.model_dump(),
}
book.append(new_book)
return new_book
def update_book(book_id: int, book_data: BookUpdate) -> dict | None:
existing_book = get_book(book_id)
if existing_book is None:
return None
existing_book.update(book_data.model_dump())
return existing_book
def delete_book(book_id: int) -> bool:
existing_book = get_book(book_id)
if existing_book is None:
return False
book.remove(existing_book)
return Truemodules模块之sys模块
modules/sys/schemas.py
py
from pydantic import BaseModel
class SysResponse(BaseModel):
environment:str
database_url:strmodules/sys/router.py
py
from fastapi import APIRouter, status
from modules.sys import service
from modules.sys.schemas import SysResponse
router = APIRouter(prefix="/sys", tags=["sys"])
@router.get("", response_model=SysResponse,status_code=status.HTTP_200_OK)
def list_sys():
return service.list_sys()modules/sys/service.py
py
from modules.sys.schemas import SysResponse
def list_sys() -> SysResponse:
# 从配置文件中获取环境变量和数据库连接字符串,不能写在开头,否则会报错
from core.config import settings
return SysResponse(
environment=settings.app_env,
database_url=settings.database_url,
)main.py
py
import os
import uvicorn
from fastapi import FastAPI
from modules.book.router import router as book_router
from modules.sys.router import router as sys_router
def create_app() -> FastAPI:
from core.config import settings
app = FastAPI(
title=settings.app_name,
debug=settings.debug,
)
# 注册路由
app.include_router(book_router)
app.include_router(sys_router)
return app
def _start(environment: str, *, reload: bool = False) -> None:
os.environ["APP_ENV"] = environment
uvicorn.run(
"main:create_app",
factory=True,
host="0.0.0.0" if environment == "prod" else "127.0.0.1",
port=8000,
reload=reload,
)
def dev() -> None:
_start("dev", reload=True)
def prod() -> None:
_start("prod")
def test() -> None:
_start("test")