1
# 初始化项目
2
uv init example_project
3

4
# 进入项目目录
5
cd example_project
6

7
# 创建虚拟环境并安装 ruff（代码格式化工具）
8
uv add ruff
9

10
# 安装 FastAPI（注意加引号，确保所有终端兼容）
11
uv add "fastapi[standard]"

1
example_project/
2
├── main.py          # 📌 项目入口文件
3
├── pyproject.toml   # 项目配置文件
4
├── venv/            # 虚拟环境
5
├── .gitignore       # Git 忽略文件
6
├── uv.lock          # 依赖锁定文件
7
├── python-version   # Python 版本文件
8
├── README.md        # 项目说明
9
├── .git/            # Git 仓库
10
└── __pycache__/     # Python 缓存文件

1
# --reload 表示代码修改后自动重新加载（开发环境推荐）
2
uvicorn main:app --reload

1
{ "Hello": "World" }

1
from fastapi import FastAPI, Path
2

3
app = FastAPI()
4

5
@app.get("/users/{id}")
6
async def get_user(id: int = Path(..., gt=0, lt=100, description="用户ID范围1-99")):
7
    return {"id": id}
8

9
@app.get("/users/{name}")
10
async def get_user_by_name(name: str = Path(..., min_length=2, max_length=10, description="用户名2-10个字符")):
11
    return {"name": name}

1
from fastapi import FastAPI, HTMLResponse, FileResponse, JSONResponse
2

3
app = FastAPI()
4

5
# 返回 HTML 页面
6
@app.get("/html", response_class=HTMLResponse)
7
async def get_html():
8
    return "<h1>这是标题</h1>"
9

10
# 返回文件下载
11
@app.get("/file", response_class=FileResponse)
12
async def get_file():
13
    return FileResponse("./files/demo.png")
14

15
# 返回 JSON 数据
16
@app.get("/json", response_class=JSONResponse)
17
async def get_json():
18
    return {"message": "hello world", "code": 200}

1
from pydantic import BaseModel, Field
2
from fastapi import FastAPI
3

4
app = FastAPI()
5

6
class User(BaseModel):
7
    id: int = Field(description="用户ID")
8
    username: str = Field(min_length=2, max_length=10, description="用户名，2-10个字符")
9
    password: str = Field(min_length=3, max_length=20, description="密码，3-20个字符")
10

11
@app.post("/register")
12
async def user_register(user: User):
13
    return {
14
        "id": user.id,
15
        "username": user.username,
16
        "password": user.password,
17
        "message": "注册成功"
18
    }

1
from fastapi import FastAPI, Depends, Query
2

3
app = FastAPI()
4

5
# ① 定义依赖项（公共参数逻辑）
6
async def common_parameters(
7
    skip: int = Query(0, ge=0),
8
    limit: int = Query(10, le=60)
9
):
10
    return {"skip": skip, "limit": limit}
11

12
# ② 在路由中注入依赖项
13
@app.get("/news")
14
async def get_news(commons: dict = Depends(common_parameters)):
15
    return {"message": "hello", "commons": commons}

1
from fastapi import FastAPI
2

3
app = FastAPI()
4

5
@app.middleware("http")
6
async def middleware1(request, call_next):
7
    print("中间件1：请求开始")
8
    response = await call_next(request)  # 传递给下一个中间件或路由
9
    print("中间件1：请求结束")
10
    return response
11

12
@app.middleware("http")
13
async def middleware2(request, call_next):
14
    print("中间件2：请求开始")
15
    response = await call_next(request)
16
    print("中间件2：请求结束")
17
    return response
18

19
@app.get("/")
20
async def root():
21
    return {"message": "Hello World"}

1
请求进入 → Middleware2 → Middleware1 → 路由处理
2
响应返回 ← Middleware2 ← Middleware1 ← 路由处理

1
中间件2：请求开始
2
中间件1：请求开始
3
中间件1：请求结束
4
中间件2：请求结束

1
import os
2
from typing import Annotated
3
from fastapi import FastAPI, File, UploadFile
4

5
app = FastAPI()
6

7
# 单文件上传（文件内容以 bytes 形式接收）
8
@app.post("/files")
9
async def create_file(
10
    files: Annotated[list[bytes], File(description="多个文件以bytes形式")]
11
):
12
    return {"file_sizes": [len(file) for file in files]}
13

14
# 多文件上传（使用 UploadFile，可获取文件名等信息）
15
@app.post("/mult_files")
16
async def create_upload_files(
17
    files: Annotated[list[UploadFile], File(description="多个文件")]
18
):
19
    return {"filenames": [file.filename for file in files]}

1
uv add aiofiles

1
import os
2
from typing import Annotated
3
import aiofiles
4
from fastapi import FastAPI, File, UploadFile
5
from fastapi.staticfiles import StaticFiles
6

7
app = FastAPI()
8

9
# 挂载静态文件目录，可通过 /static/xxx 访问上传的文件
10
app.mount("/static", StaticFiles(directory="uploads"), name="static")
11

12
# 确保上传目录存在
13
os.makedirs("./uploads", exist_ok=True)
14

15
# 单文件上传并保存
16
@app.post("/upload_file")
17
async def upload_file(file: Annotated[UploadFile, File(description="单文件上传")]):
18
    async with aiofiles.open(f"./uploads/{file.filename}", "wb") as f:
19
        content = await file.read()   # 读取上传内容
20
        await f.write(content)        # 写入文件
21
    return {
22
        "filename": file.filename,
23
        "content_type": file.content_type,
24
        "size": file.size,
25
        "path": f"/static/{file.filename}",
26
        "message": "上传成功",
27
    }
28

29
# 多文件上传并保存
30
@app.post("/upload_files")
31
async def upload_files(
32
    files: Annotated[list[UploadFile], File(description="多文件上传")]
33
):
34
    for file in files:
35
        async with aiofiles.open(f"./uploads/{file.filename}", "wb") as f:
36
            content = await file.read()
37
            await f.write(content)
38
    return {"filenames": [file.filename for file in files], "message": "上传成功"}

1
# SQLAlchemy 异步支持
2
uv add sqlalchemy[asyncio]
3

4
# MySQL 异步驱动
5
uv add aiomysql

1
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
2
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
3
from sqlalchemy import DateTime, func, String, Float, select
4
from datetime import datetime
5
from fastapi import FastAPI, Depends
6
from contextlib import asynccontextmanager
7

8
# ========== 步骤 1：创建异步数据库引擎 ==========
9
ASYNC_DATABASE_URL = "mysql+aiomysql://root:000000@localhost:3306/booktest?charset=utf8mb4"
10
async_engine = create_async_engine(
11
    ASYNC_DATABASE_URL,
12
    echo=True,           # 是否打印 SQL 语句（调试用）
13
    pool_size=10,        # 连接池大小
14
    max_overflow=20,     # 超出连接池大小后最多可创建的连接数
15
)
16

17
# ========== 步骤 2：定义基类和模型类 ==========
18
class Base(DeclarativeBase):
19
    """所有模型的基类，包含公共字段"""
20
    create_time: Mapped[datetime] = mapped_column(
21
        DateTime,
22
        default=func.now,           # Python 层面：每次插入时调用
23
        server_default=func.now(),  # 数据库层面：DDL 默认值
24
        comment="创建时间"
25
    )
26
    update_time: Mapped[datetime] = mapped_column(
27
        DateTime,
28
        default=func.now,
29
        onupdate=func.now(),        # 更新记录时自动设置为当前时间
30
        server_default=func.now(),
31
        comment="更新时间"
32
    )
33

34
class Book(Base):
35
    __tablename__ = "book"    # 对应数据库中的表名
36
    id: Mapped[int] = mapped_column(primary_key=True, comment="书籍ID")
37
    bookname: Mapped[str] = mapped_column(String(255), comment="书籍名称")
38
    author: Mapped[str] = mapped_column(String(255), comment="作者")
39
    price: Mapped[float] = mapped_column(Float, comment="价格")
40
    publisher: Mapped[str] = mapped_column(String(255), comment="出版社")
41

42
# ========== 步骤 3：建表函数 ==========
43
async def create_table():
44
    async with async_engine.begin() as connect:
45
        await connect.run_sync(Base.metadata.create_all)  # 同步建表转异步执行
46

47
# ========== 步骤 4：应用生命周期管理 ==========
48
@asynccontextmanager
49
async def lifespan_app(app):
50
    print("应用启动中...")
51
    await create_table()              # 启动时自动建表
52
    yield                             # 应用运行中
53
    print("应用关闭中...")
54
    await async_engine.dispose()      # 关闭时释放数据库连接
55

56
app = FastAPI(lifespan=lifespan_app)
57

58
# ========== 步骤 5：创建会话工厂 ==========
59
AsyncSessionLocal = async_sessionmaker(
60
    bind=async_engine,          # 绑定数据库引擎
61
    class_=AsyncSession,        # 使用异步会话类
62
    expire_on_commit=False      # 提交后对象不过期（避免重新查询）
63
)
64

65
# ========== 步骤 6：数据库依赖项（自动管理会话生命周期） ==========
66
async def get_database():
67
    async with AsyncSessionLocal() as session:
68
        try:
69
            yield session       # 提供会话给路由使用
70
            await session.commit()    # 无异常则提交
71
        except Exception:
72
            await session.rollback()  # 有异常则回滚
73
        finally:
74
            await session.close()     # 最终关闭会话
75

76
# ========== 步骤 7：路由中使用依赖注入 ==========
77
@app.get("/books")
78
async def get_book_list(db: AsyncSession = Depends(get_database)):
79
    result = await db.execute(select(Book))
80
    books = result.scalars().all()
81
    return books

1
mysql+aiomysql://用户名:密码@主机:端口/数据库名?charset=utf8mb4

1
from pydantic import BaseModel
2

3
# 定义请求体模型
4
class BookCreate(BaseModel):
5
    id: int
6
    bookname: str
7
    author: str
8
    price: float
9
    publisher: str
10

11
@app.post("/books/add")
12
async def add_book(book: BookCreate, db: AsyncSession = Depends(get_database)):
13
    book_obj = Book(**book.__dict__)   # 将 Pydantic 模型转为 ORM 对象
14
    db.add(book_obj)                   # 添加到会话
15
    await db.commit()                  # 提交到数据库
16
    return {"book": book, "msg": "添加成功"}

1
from pydantic import BaseModel
2

3
class BookUpdate(BaseModel):
4
    bookname: str
5
    author: str
6
    price: float
7
    publisher: str
8

9
@app.put("/books/update/{id}")
10
async def update_book(id: int, data: BookUpdate, db: AsyncSession = Depends(get_database)):
11
    book = await db.get(Book, id)       # ① 先查询
12
    if book is None:
13
        raise HTTPException(status_code=404, detail="书籍不存在")
14

15
    book.bookname = data.bookname       # ② 修改属性
16
    book.author = data.author
17
    book.price = data.price
18
    book.publisher = data.publisher
19

20
    await db.commit()                   # ③ 提交到数据库
21
    return book

1
@app.delete("/books/delete/{id}")
2
async def delete_book(id: int, db: AsyncSession = Depends(get_database)):
3
    book = await db.get(Book, id)       # ① 先查询
4
    if book is None:
5
        raise HTTPException(status_code=404, detail="查无此书")
6

7
    await db.delete(book)               # ② 标记删除
8
    await db.commit()                   # ③ 提交到数据库
9
    return {"msg": "删除成功"}

1
# ❌ 错误：缺少 Depends
2
async def get_news(db: AsyncSession = get_database):
3

4
# ✅ 正确
5
async def get_news(db: AsyncSession = Depends(get_database)):

1
# ⚠️ 注意：静态路由要放在动态路由前面
2
@app.get("/books/search")    # 静态路由（先匹配）
3
async def search(): ...
4

5
@app.get("/books/{id}")      # 动态路由（后匹配）
6
async def get_book(id: int): ...