Skip to content

简单版增删改查 ​

修改 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: int

modules/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 True

modules模块之sys模块 ​

modules/sys/schemas.py ​

py
from pydantic import BaseModel

class SysResponse(BaseModel):
    environment:str
    database_url:str

modules/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")