我的后端之路

今天是学习的第一天,怀揣着紧张和激动的心情,踏上了这条后端开发员的道路(笑)

---

#今天稍稍了解了一下软件和硬件的基础知识,大致上掌握的还不错.(其实想弄一下,,听课时的截图的,太烂了,遂不弄了)

预科小知识

1-对于计算机的快捷按键我大致比较熟悉,快速的略过了
2-快捷键易如反掌
3-Dos命令
   md<目录名>
   rd目录名>
   cd > <文件名>
   del  <文件名>

卸载JDK

1-删除Java的安装目录
2-删除Java_HOME
3-删除path下关于Java的目录
4-java-version(命令行检测是否完全删除)

安装JDK

1-官网下载
2-同意协议
3-下载对应版本(17为ds推荐,8则是公司会用(?))
4-记住安装路径 C:\Program Files\Java\jdk-17(个人习惯,避免权限问题,开发工具识别等) 
5-配置环境变量(java bin)
6-环境变量-->Java-HOME(路径同上)
7-测试JDK是否安装成功
  1.打开cmd  //输入java -version

骗你你呀.立刻切换为python形态!

基于python

---

Day 1

花括号的作用

1. {变量名},插入为变量的值
2. 用于表达式计算, 格式化数字{pi:.2f} //Π .2代表小数点后两位, f代表为浮点数(数据类型)
3. 访问字典
   eg. person={“name”:”张三”,”age”:20}
   print(f”姓名:{person[‘name’]},年龄:{person[‘age’]}”)

*中括号的作用

1. 创建带元素的列表,空元素也可
2. 访问序列元素 //同理也可以访问字符串,元组(类似c++数组)
   eg.
   fruits = [‘apple’, ‘banana’, ‘orange’, ‘grape’]
   print(fruits[0]) # ‘apple’ - 第一个元素
   print(fruits[1]) # ‘banana’ - 第二个元素
   print(fruits[-1]) # ‘grape’ - 最后一个元素
   print(fruits[-2]) # ‘orange’ - 倒数第二个元素`
3. 修改访问元素(字典值)
   eg.fruits = [‘apple’, ‘banana’, ‘cherry’]
   fruits[1] = ‘blueberry’
   print(fruits) # [‘apple’, ‘blueberry’, ‘cherry’]
4. 取出字典中键对应的值[“键名”] /一般用get(),找不到键会返回None,而不是报错
   *\n 用于换行

---

定义函数(含参)

def 函数名( 参数 ):
函数体的内容
返回值

(默认函数)

1. 默认参数必须在非参数函数之前
2. 默认值只在函数定义时计算一次 //用None做默认值更安全,每次调用都创建新列表
3. 可以混合使用位置参数和默认参数
4. 默认函数本身可以被重新定义

3.全局or局部变量

A.全局变量
  1.在函数外部定义
  2.在整个模块中可见.
B.局部变量
  1.在函数内部定义
  2.只在函数内部可见
C.变量作用域优先级LEGB规则

==L==ocal -局部作用域
==E==nclosing -闭包函数作用域
==G==lobal -全局作用域
==B==uilt-in -内置作用域
D.常见method
1.global关键字 //修改为全局变量
2.使用nonlocal关键字 //修改为闭包变量

4.文件读写

首先明确
==文件对象可以存储在变量中,用于操纵文件==
==文件内容可以存储在变量中,作为字符串处理==
==文件本身(磁盘上的实体)不能直接存储在变量中==
**推荐 with 语句,自动开关
      # 将文件内容存储在变量中
      with open('example.txt', 'r', encoding='utf-8') as file:
      file_content = file.read()    // 文件内容存储在变量 file_content 中

      print(file_content)      // 可以多次使用文件内容
      print(f"文件长度: {len(file_content)} 字符")

      # 文件内容可以被修改并写回
      modified_content = file_content.upper()
      with open('example_upper.txt', 'w', encoding='utf-8') as file:
      file.write(modified_content)
**基本操作
      # 将文件对象存储在变量中
      read_file = open('input.txt', 'r', encoding='utf-8')    //r和r+模式只能读取已存在的文件,不能创建文件
      write_file = open('output.txt', 'w', encoding='utf-8')

      # 通过变量操作文件
      data = read_file.read()          // 从输入文件读取
      write_file.write(data)           // 写入输出文件

      # 记得关闭文件
      read_file.close()
      write_file.close()

##表格汇总
|’r’|只读|
|’w’|只写|
|’a’|追加|
|’x’|创建|
|’r+’|读写|
|’w+’|读写|
|’a+’|读写追加|

Day2

字典中简单的映射关系

eg.
def create_multiple_files():
//批量创建多个文件
file_contents = {
“readme.txt”: “这是说明文件”,
“config.json”: ‘{“name”: “app”, “version”: “1.0”}’,
“notes.md”: “# 学习笔记\n## Python文件操作”
}

    for filename, content in file_contents.items():
        with open(filename, "w", encoding="utf-8") as file:
            file.write(content)
        print(f"创建文件: {filename}")

• · ✅ file_contents 字典包含：文件名 → 文件内容 的映射
  · ✅ filename 变量获取：字典的键（文件名）
  · ✅ content 变量获取：字典的值（文件内容）
  · ✅ file.write(content) 中的 content：就是当前循环对应的文件内容
  *字典=键值对集合
• 键(key)–值(value)
• 一个键对应一个值
• 键是唯一的,值可以重复
• 字典名 = {
  键1: 值1,
  键2: 值2,
  键3: 值3
  }

字典可以套娃即字典中的字典

1.遍历键:
for key in person.keys():
print(key)
2. 遍历值:
for value in person.values():
print(value)
3.同时遍历值和键:
for key, value in person.items():
print(key, “→”, value)

python中的类(class) //类名首字母大写

1.类中的函数,我们统称为method(方法),定义时需要加上self,类似于c++的隐式指针this
2.def __init__(self,)      //类似与c++的构造函数,负责初始化的

元组与数组

** 需要list. 来调用     //list储存的是地址,
**1** 对于元组:元组是一个不可变的对象，赋值后无法更改其值!
1.排列位置的索引是从0开始,-1是最后一位
2.append(),从list的最后添加
3.insert(位置,值)
4.remove(值),第一次出现的值
5.a=[0:3] a中的第一位到第三位   //切片
6.index(value)  //从第一次出现该值找索引
7.count(value)   //出现该值的次数
8.sort()  排列,从小到大   //sort(reverse=True)为降序

错误处理try

 常见结构     
      try:
          # 可能引发异常的代码
          pass
      except ExceptionType:
          # 处理特定类型的异常
          pass

      except AnotherExceptionType:
          # 处理另一种类型的异常
          pass
      else:
          # 如果没有异常发生，执行这里的代码
          pass
      finally:
          # 无论是否发生异常，都会执行这里的代码
          pass
e.g  # 不好的做法
      try:
          do_something()
      except:
          pass  # 静默忽略所有异常

      # 好的做法
      try:
          do_something()
      except SpecificError as e:
          logger.error(f"处理SpecificError: {e}")
          # 采取适当的恢复措施
      except AnotherError as e:
          logger.error(f"处理AnotherError: {e}")
          # 采取不同的恢复措施

zip函数的用法

**1**   # zip()返回的是迭代器对象,不是列表   //只能遍历一次
**2**   #转化为列表后可以看到所有内容  ==以下情况优先考虑==       //可以遍历多次
        1.需要重复访问
        2.需要修改数据
        3.需要索引或切片操作

1. # 将两个列表对应位置的元素组合
    names = ['Alice', 'Bob', 'Charlie']
    ages = [25, 30, 35]

    zipped = zip(names, ages)
    print(list(zipped))     # [('Alice', 25), ('Bob', 30), ('Charlie', 35)]  
2. # 使用 * 操作符可以"解压"zip对象
    zipped = [('Alice', 25), ('Bob', 30), ('Charlie', 35)]
    names, ages = zip(*zipped)
    print(names)   # ('Alice', 'Bob', 'Charlie')
    print(ages)    # (25, 30, 35)
3. #  同时遍历多个列表
    e.g  names = ['Alice', 'Bob', 'Charlie']
          ages = [25, 30, 35]
          cities = ['New York', 'London', 'Tokyo']

          for name, age, city in zip(names, ages, cities):
              print(f"{name} is {age} years old and lives in {city}")
4. # lambda
   1. 基本用法
      lambda 参数:表达式
      相当于构造一个简单的函数
5. # map()
   1. map(function,iterable,...)  
   2. lterable(可迭代对象)指的是可以逐个返回其元素的对象,常见的包括list tuple(元组) str dic(字典)

pickle模块

1.基本文件序列化
2.批量数据处理
3.智能缓冲系统
---
****pickle.load()
1. 从文件加载对象
2. 加载各种数据类型

****pickle.dump(A,B)  //A保存到B
1.保存简单对象到文件
2.使用不同的协议版本 

---

开始练手

积累的知识点

1.计数模式（counting pattern）”，核心思想是 “查旧值 + 更新 + 写回”。
eg. skill_count[skill] = skill_count.get(skill, 0) + 1
读取 skill_count 这个字典中键 skill 对应的值；
如果存在，就取出该值；
如果不存在，就用默认值 0；
在这个值基础上加 1；
再把新值写回 skill_count[skill]。
2.
applicants_data = [
{
“name”: “张三”,
“score”: 85,
“skills”: [“Python”]
},
{
“name”: “李四”,
“score”: 90,
“skills”: [“C++”, “Java”]
}
]
for person in applicants_data:
图解:
applicants_data (list)
│
├── [0] ── person ── dict
│ ├── “name”: “张三”
│ ├── “score”: 85
│ └── “skills”: [“Python”]
│
└── [1] ── person ── dict
├── “name”: “李四”
├── “score”: 90
└── “skills”: [“C++”, “Java”]
3.isinstance(变量, 类型)
判断变量是不是指定类型
isinstance(score, (int, float)) //支持多个类型检查

1 week

*args位置可变参数

1.args 是一个 元组（tuple），里面存放所有的位置参数。你可以像普通元组那样操作它:    //请注意，args 只是一个名称。您不需要使用名称
2. e.g def add(*args):
          result=0
          for x in args:
              result+=x
          return result

**kwargs

1.kwargs 是一个 字典（dict），保存了所有的“键=值”形式的参数,可以像字典一样访问
2.e.g      def show_info(**kwargs):
            for key, value in kwargs.items():
                print(f"{key}: {value}")

            show_info(name="小明", score=90, subject="数学")
**定义函数参数时的顺序**:
  位置参数 → 默认参数 → *args → **kwargs

decorator:

def 装饰器名(被装饰的函数):
  def wrapper(*args, **kwargs):
      # 1️⃣ 执行函数前的代码
      result = 被装饰的函数(*args, **kwargs)
      # 2️⃣ 执行函数后的代码
      return result
  return wrapper
@装饰器名
def 被装饰函数():

相当于:被装饰函数 = 装饰器名(被装饰函数)

异步编程(async,await)

  Asyncio 是 Python 用于异步编程的库，专门用来处理耗时的 IO 操作，比如网络请求、文件读写、数据库查询等 
(1)1.定义协程函数
  2.包装协程为任务
  3.建立事件循环
(2)await的用法
  1.暂停当前的协程
  2.获取await后的协程结果
(3)如何写一个协程
   e.g
    import asyncio
    async def worker():
        print("开始")
        await asyncio.sleep(1)
        print("结束")

    async def main():
        # 包装协程
        task = asyncio.create_task(worker())

        # 等待任务结束
        await task

    asyncio.run(main())   //启动事件循环
  (4)返回协程结果
  await task(最简单)
  asyncio.gather()获取所有结果,顺序与传入顺序一样
  asyncio.as_completed()  //立即返回结果无需等待
  (5)其他asyncio的异步库
  1.aiohttp   //网络连接
  2.aiofiles   //文件

if name == “main“: 是 Python 程序的入口判断，用来区分：

    当前文件是被直接运行,就执行if后面的代码
    还是被别的文件 import 导入

---

关于http

1.请求方法
  | 方法       | 用途         |
  | -------- | ---------- |
  | `GET`    | 获取资源（无副作用） |
  | `POST`   | 创建资源       |
  | `PUT`    | 替换资源       |
  | `PATCH`  | 局部更新资源     |
  | `DELETE` | 删除资源       |
2.| 状态码                        
  | 场景             |
  | --------------------------- | -------------- |
  | `200 OK`                    | 一切正常           |
  | `201 Created`               | 创建成功           |
  | `400 Bad Request`           | 参数错误           |
  | `401 Unauthorized`          | 未认证            |
  | `403 Forbidden`             | 不允许访问          |
  | `404 Not Found`             | 资源不存在          |
  | `422 Unprocessable Entity`  | FastAPI 用于校验失败 |
  | `500 Internal Server Error` | 服务端异常          |
3.请求结构
  POST /login HTTP/1.1
  Host: example.com
  Content-Type: application/json

  {"username": "aa", "password": "bb"}

---

万事开头难,不要乱装东西(血的教训)

---

RESTful的路径(URL)的 设计原则

-使用名词,不用动词
e.g GET /api/user #获取所有用户
-用复数形式表示资源合集(约定俗成)
e.g POST /api/users #嗨，请在你的用户数据库里，为我添加一个新成员
-层级关系表达关联
e.g GET /api/users/1/posts #用户1的所有数据

RESTful设计风格详解

1. RESTful 的核心思想
   -资源（Resource）是核心：一切数据（如用户、文章、订单）都视为资源。
   -每个资源都有一个唯一的 URI（统一资源标识符，通常是一个 URL 路径）
   -使用标准 HTTP 方法（GET/POST/PUT/DELETE）操作资源
   -无状态（Stateless）：每次请求都应该包含足够的信息，服务器不保存客户端状态
   -可缓存、分层系统、统一接口等（更高级特性）
2. RESTful中的资源是什么:
   -用户(User)
   -文章(Articles)
   -订单(Order)
   -商品(Products)

如何用Query参数(查询参数)

1. 基本定义:Query 参数是放在 URL 问号（?）后面的一组键值对，用于向服务器传递可选的、非敏感的参数信息，常用于过滤、分页、排序、搜索等
   e.g 假设我们有一个获取用户列表的 API：
   GET /api/users
   你想添加如下查询条件：
   只获取状态为 active 的用户
   每页 10 条
   第 2 页
   那么 URL 应该写成：
   GET /api/users?status=active&page=2&per_page=10
   2.# 在不同场景下如何加 Query 参数

   工具/语言	示例
   浏览器 / 直接访问	在地址栏输入： https://api.example.com/users?name=Alice&age=20
   cURL	bash<br>curl "https://api.example.com/users?status=active&page=2"<br>
   Postman	在请求的 Params 标签页中添加键值对，Postman 会自动拼接到 URL
   Python (requests)	python<br>import requests<br>params = {"status": "active", "page": 2}<br>response = requests.get("https://api.example.com/users", params=params)<br>
   JavaScript (fetch)	javascript<br>fetch("https://api.example.com/users?status=active&page=2")<br>

   ✔ 注意：Query 参数用于 GET 请求居多，但技术上其他方法也可以带，只是不常见

如何写好 JSON Body

JSON Body 是发给 API 的请求数据，用于 POST / PUT 等操作。

基本语法:

{
  "key": "value"
}


键必须用双引号 " "
值可以是：
* 字符串 `"text"`
* 数字 `123`
* 布尔 `true / false`
* 数组 `[1, 2, 3]`
* 对象 `{ "name": "Tom" }`

e.g：
json
{
  "name": "Apple",
  "price": 3.5
}

## JSON Body 必须符合 API 的 Pydantic 模型

例如你的 FastAPI 模型：

```python
class Item(BaseModel):
    name: str
    price: float

那么 JSON Body 必须写成：

{
  "name": "Apple",
  "price": 3.5
}

如果字段缺失、类型不对，FastAPI 会返回：
422 Unprocessable Entity

数据多时也可以写嵌套结构：

{
  "order_id": 10,
  "customer": {
    "name": "Alice",
    "email": "alice@example.com"
  },
  "items": [
    { "name": "Apple", "price": 3.5 },
    { "name": "Banana", "price": 2.0 }
  ]
}

记住

* 必须用 `{}` 包住
* key 要用 `"双引号"`
* 值类型必须正确（string / number / boolean / object / array）
* 不要多逗号
* 必须符合 FastAPI 对应模型

---

关于requests库的几点:

1.是python中最流行,最易用的HTTP请求库,用于发送HTTP请求(GET,POST) 与Web服务(api,网站)交互,适合爬虫,接口调用,数据获取等场景
2.语法
(1). GET:requests.get(url)
(2). data = {“name”: “Tom”, “age”: 20}
POST:requests.post(url.json=data) //json=data 会自动设置成Content-Type: application/json
(3). response.status_code(获取状态码)
(4). response.json(将json文件响应解析为python字典)
(5). GET 请求带查询参数:
params = {“name”: “Alice”, “age”: 25}
response = requests.get(“https://httpbin.org/get“, params=params)
print(“完整 URL:”, response.url) # 自动拼接参数//params=params,requests 会自动将这个字典转换为 URL 的查询字符串（即 ?name=Alice&age=25），并拼接到 URL 后面。

关于typing模块:

Python 提供的类型提示工具库，里面包含了很多用于静态类型检查的类型构造器（如 List, Dict, Tuple, Union 等）

关于Union模型:

当你希望函数参数或返回值可以接受多种不同类型时就会用到 Union。   

Fastapi

==pip install fastapi[all]== 安装必要的库

from fastapi import FastAPI: 导入 FastAPI 的核心类，用于创建应用程序实例。

app = FastAPI()
app: 这是 FastAPI 的主要实例。所有的 API 路由（路径）和配置都将挂载在这个对象上。
在运行服务器时（例如使用 Uvicorn），你会引用这个 app 对象。

(“/“): 指定请求的路径（URL），这里是根目录

1. 基本骨架
   app/
   ├── main.py
   ├── routers/
   │ ├── users.py
   │ └── items.py

2. 运行:
   uvicorn main:app –reload

3. FastAPI 使用Python装饰器语法 来注册接口
   e.g :@app.get(“/items”)

     def read_items():
        
        return {"items": []}
   

   意思是 :@app.get(“/items”) 是装饰器

        表示这个函数是一个 处理 GET /items 请求的函数
        
        每个装饰器对应一个 HTTP 方法
   

4. 常用的decorator:

   1. @app.get(“/path”)
   2. @app.post(“/path”)
   3. @app.put(“/path”)
   4. @app.patch(“/path”)
   5. @app.delete(“/path”)

5. 定义带参数的路径:

   e.g: @app.get(“/items/{item_id}”)
   def read_item(item_id: int, q: Union[str, None] = None):
   return {“item_id”: item_id, “q”: q}

A. 路径参数 (Path Parameter)
  
  * 路由: /items/{item_id} 中的 {item_id} 是一个占位符。
    * 你可以通过定于枚举定义参数值的可选项     //class optional(str,Enum):
  * 参数: 函数参数 item_id: int 与路径中的占位符名称一致。
  * 类型校验: : int 是 Python 的类型提示。FastAPI 利用它进行自动校验：
    * 如果你访问 /items/5，FastAPI 会把 5 转换成整数传递给函数。
    * 如果你访问 /items/foo，FastAPI 会自动返回一个清晰的错误信息（因为 "foo" 不是整数）。
  
B. 查询参数 (Query Parameter)

定义: 参数 q 出现在函数定义中，但没有出现在路径装饰器 /items/{item_id} 中。因此，FastAPI 自动将其识别为查询参数（URL 中 ? 后面的部分）。
类型: Union[str, None] = None
  这意味着 q 可以是字符串，也可以是 None。
  = None 设置了默认值。这意味着这个参数是可选的。
* 一句话:只要参数没有出现在路径 {} 中，FastAPI 就会把它当成查询参数。 

用法示例:
/items/5?q=something -> item_id 是 5, q 是 "something"。
/items/5 -> item_id 是 5, q 是 None

---

关于pydantic

BaseModel 是 Python Pydantic 库 的核心概念。简单来说，它是一个用来定义数据结构、验证数据和转换数据的类。

在 FastAPI 中，Pydantic 模型扮演着至关重要的角色，它负责处理请求体（Request Body）和响应体（Response Body）。

1. 核心定义:

   1. 要创建一个 Pydantic ，你需要定义一个继承自 BaseModel 的类，并声明字段及其类型。

   2. 代码示例
      e.g:
      from pydantic import BaseModel
      from typing import Union

          ///完整代码:
          from fastapi import FastAPI
          from pydantic import BaseModel, EmailStr
      
          app = FastAPI()
      
          # 请求模型：客户端传入的数据
          class UserCreateRequest(BaseModel):
              username: str
              email: EmailStr
              password: str
      
          # 响应模型：返回给客户端的数据
          class UserResponse(BaseModel):
              id: int
              username: str
              email: EmailStr
      
          @app.post("/users", response_model=UserResponse)
          
      def create_user(user: UserCreateRequest): 
      此处定义请求体(即通过post传输的数据)为"user"参数,
      那么客户端发来的请求体JSON 数据就会被自动接收和验证(pydantic)   
      ```               
          new_user = {
              "id": 1,
              "username": user.username,  # 修正这里
              "email": user.email
          }
          return new_user
      

   3. Pydantic 的三大功能:
      当你使用上面这个 Item 类来接收数据时，Pydantic 会自动完成以下工作：
      A. 数据验证 (Validation)
      它会检查传入的数据是否符合定义的类型。

      1. 如果 price 传入的是一个无法转为数字的字符串（如 “hello”），Pydantic 会报错。
      2. 如果缺少了必填字段 name，Pydantic 也会报错。

   B. 数据转换/解析 (Parsing)

    3. 如果你给 price 传入字符串 "10.5"，Pydantic 会自动把它转换成浮点数 10.5。
    4. 如果你给 is_offer 传入字符串 "True" 或 "on"，它会自动转换成布尔值 True。
   

   C. JSON Schema 生成
   FastAPI 使用 Pydantic 模型来自动生成 API 文档（Swagger UI）。
   它会告诉前端开发者：“这个接口需要一个对象，必须包含字符串类型的 name 和浮点数类型的 price”。

2. 定义请求模型(UserCreateRequest) //请求模型用于接收客户端发送给 API 的数据（通常是 POST、PUT 请求的 JSON Body）。

  1. BaseModel 是所有 Pydantic 模型的基类

  2. 字段类型用 Python 的类型注解（如 str, int, EmailStr）

  3. Pydantic 会自动校验类型并在错误时抛出异常 
      e.g:
    
        from pydantic import BaseModel, EmailStr
        
        class UserCreateRequest(BaseModel):         
          username: str
          email: EmailStr
          password: str
          1. BaseModel 是所有 Pydantic 模型的基类

          2. 字段类型用 Python 的类型注解（如 str, int, EmailStr）

          3. Pydantic 会自动校验类型并在错误时抛出异常
            
            将模型类作为函数参数的类型提示
          
##    3. 定义响应模型(UserResponse)
           
       接上:
        
       ///
        class UserResponse(BaseModel):
          id: int
          username: str
          email: EmailStr
        @app.post("/users", response_model=UserResponse) 
              
           # 1. @app.post("/users")：告诉 FastAPI，当客户端发起 POST 请求到 /users 路径时，就调用下面的函数。
           # 2.response_model=UserResponse：规定返回的数据必须符合 UserResponse 模型的结构和类型，FastAPI 会自动校验并过滤掉多余字段。
        
        def create_user(user: UserCreateRequest): 
          
           # 发生了什么:
           1. 参数 user 类型是 UserCreateRequest（一个 Pydantic 模型）。
              在函数里你可以直接用 user.name、user.email、user.age，而不用手动解析 JSON
           
           2. FastAPI 会自动把客户端传来的 JSON 请求体解析成这个模型对象(User)，并进行类型校验。              
           
           3. 如果请求体缺少字段或类型不匹配，会直接返回 422 错误。
              
           # 模拟数据库保存 (实战中会用MySQL)
        new_user = { 
          "id": 1, 
          "username": 
          user.username,     
          "email": user.email 
        } 
           # username 和 email 来自请求体
        return new_user
                
           # 1.FastAPI 会把这个字典转换成 UserResponse 模型对象。

           # 2. 如果字典里有多余字段（比如密码），会被自动过滤掉。

           # 3. 最终返回给客户端的是一个 JSON 响应，结构严格符合 UserResponse。 
    
##   4. 关于Field
      1. Field 是 Pydantic 提供的一个函数，用来给模型字段添加 约束条件、默认值 和 元数据。它常用于==请求体验证和文档说明==
      2.  基本用法:
          e.g
          ```Python
          from pydantic import BaseModel, Field, EmailStr
          class User(BaseModel):
            name: str = Field(..., min_length=2, max_length=50, description="用户名，长度2-50")
              email: EmailStr = Field(..., description="合法的邮箱地址")
              age: int = Field(default=18, ge=0, le=120, description="年龄，默认18，范围0-120")
  3.  Pydantic Field 参数速查表
      
      | 参数                  | 说明                     |
      |-----------------------|--------------------------|
      | `default`             | 默认值                   |
      | `...`                 | 必填字段                 |
      | `title`               | 字段标题                 |
      | `description`         | 字段描述                 |
      | `min_length` / `max_length` | 字符串长度限制       |
      | `ge` / `le`           | 数值范围（大于等于 / 小于等于） |
      | `gt` / `lt`           | 数值范围（大于 / 小于）  |
      | `regex`               | 正则表达式验证           |
      | `example`             | 示例值（用于文档）       |

5. 关于Path:

   1. 用于“路径参数”添加校验规则、默认值说明、范围限制、描述信息等高级配置。 
   2. * 限制路径参数的范围（如必须 > 0）
      * 限制路径参数的范围（如必须 > 0）
      * 添加描述信息
      * 设置别名
      * 添加正则表达式\
      * 让文档更清晰

Fastapi的项目拆分:

将 FastAPI 项目拆分为 main.py, schemas.py, routers (以及通常还有 models.py 和 crud.py)，是为了遵循 “关注点分离” (Separation of Concerns) 的原则

1. schemas.py: 公司的“公文格式” (Pydantic 模型)

**作用：定义数据的“长相”（输入和输出的格式）。**

• 原来的做法：你在 main.py 里定义 class UserCreate(BaseModel)。
• 拆分后：
  * 这里只放 Pydantic 模型。
  * 它不管数据库怎么存，也不管业务逻辑怎么跑，它只管数据验证。
  * 比如：前端发来的 JSON 必须有哪些字段？接口返回给前端的 JSON 长什么样？
  为什么要拆？

 *   **复用**：用户注册需要校验数据，用户修改资料也需要校验。把模型放在独立文件，各个路由都能引用，不用重复写。
 *   **清晰**：打开这个文件，一眼就能看出你的 API 支持接收什么数据，会返回什么数据，相当于一份活文档。

• init 的作用
  • 让目录变成一个可导入的包(package),加上__init__之后,python才知道schema是一个包,可以从里面导入模块.

2. routers :公司的办事部门,(文件夹or模块) //api的路由

• 原来的做法：你在 main.py 里写了 @app.get("/users/"), @app.post("/items/")，导致文件巨长。

• 拆分后：
  * 你会建立一个 routers 文件夹。
  * routers/users.py: 专门处理和用户有关的 URL（登录、注册、查信息）。
  * routers/items.py: 专门处理和商品有关的 URL（增删改查商品）。
  * 这里面不使用 app = FastAPI()，而是使用 **router = APIRouter()**。

• 关于APIRouter
  APIRouter 是 FastAPI 提供的一个 路由分组工具。
  router = APIRouter() //当前文件里的所有接口都注册到这个 router 上

  • 基本用法:
    • 建一个路由文件（如 routers/user.py）
  • 高级功能:
    • 给整个路由组加前缀
      • router = APIRouter(prefix=”/api”) //所有接口自动变成 /api/user
    • 给整个路由组加标签（用于自动文档）
      • router = APIRouter(tags=[“User”]) //Swagger UI 会自动分组。

3. main.py: 公司的前台(入口文件)

作用：统筹全局，启动服务。

• 拆分后：
  * main.py 变得非常干净、短小。
  * 它不再包含具体的业务逻辑。
  * 它的主要工作是：
  1. 初始化 app = FastAPI()。
  2. 配置全局设置（比如跨域 CORS、数据库连接）。
  3. 把各个部门（routers）注册进来

4. 还有一个常见的 models.py (数据库模型)

• models.py (ORM, 如 SQLAlchemy): 给数据库看的，对应数据库里的表结构。

5. utils=utilities=工具,工具函数

• 放通用工具函数
  • 统一响应格式
  • 时间处理函数
  • 日志工具
  • Token 生成工具
  • 字符串处理工具
  • 文件处理工具

6.总结:标准目录结构长什么样

一个成熟的 FastAPI 项目通常是这样的：

my_project/
├── main.py           # 【入口】启动APP，连接各个路由，配置CORS
├── schemas.py        # 【公文】Pydantic模型，定义请求/响应的JSON格式
├── models.py         # 【仓库】数据库表结构 (SQLAlchemy类)
├── database.py       # 【基建】数据库连接配置
├── requirements.txt  # 【清单】依赖包列表
└── routers/          # 【部门】具体的API业务逻辑
    ├── __init__.py
    ├── users.py      # 处理 /users/ 相关的请求
    └── items.py      # 处理 /items/ 相关的请求

7 更自动一点:

 * 自动加载所有 routers:
    
    routers/
    │── __init__.py
    │── users.py
    │── items.py
  
  1. 在__init__.py:
    
from .users import router as users_router
from .items import router as items_router

all_routers = [
    users_router,
    items_router
]
  2. 然后在 main.py：
    from routers import all_routers

    for r in all_routers:
        app.include_router(r)

  **这样团队成员新增 router 时，只需要在 __init__.py 注册即可。**

### 8.关于哈希密码
  
  1. FastAPI 官方推荐的密码哈希库:  pip install passlib[bcrypt]
  2. 使用慢哈希:如bcrypt,  优点:自动加盐
  3. 使用 CryptContext 管理算法
    
    
pwd_context = CryptContext(
    schemes=["bcrypt"],  //schemes与项目结构中的schemas无关
    deprecated="auto"
)
  4. 密码哈希只在“写入数据库前”发生
  5. 写哈希函数放在security
  6. 调用哈希函数   //验证(verify)

### 9.FastAPI的错误处理

  1. 流程:service→router→http→json(前端看到)
  2. raise  作用:停止程序,将错误类型及原因返回给上层 
     
raise "错误类型"(错误原因)
  3. try和except负责链接service和router
  4. try: 尝试 
     except: 捕获错误,转为http返回错误/进行处理
  
  5. 错误处理流程(全局,不含try和expect)
    
    路由函数 → raise BusinessException
            ↓
    FastAPI 捕获异常
            ↓
    匹配到 business_exception_handler
            ↓
    构造 ResponseModel → JSONResponse
            ↓
    返回给前端 (HTTP 状态码 + JSON)


  ### 10.关于依赖
  1. 函数作为依赖（Depends）
  - **定义**：依赖函数是 FastAPI 的依赖注入机制，用来封装通用逻辑（认证、数据库、参数校验）。
  - **作用**：
    - 自动调用并注入返回值到路由函数参数。
    - 解耦业务逻辑与通用逻辑。
    - 提供可复用、可测试的模块。

  ---

   2. 全局依赖 vs 路由依赖
  | 特性       | 全局依赖 | 路由依赖 |
  |------------|----------|----------|
  | 作用范围   | 所有路由 | 单个路由或路由组 |
  | 常见场景   | 统一认证、日志、全局参数 | 特定认证、数据库连接、局部参数校验 |
  | 声明位置   | `FastAPI()` 实例化时 | 路由装饰器或 `APIRouter` |

  ---

   3. 常见使用场景
 - **认证**：验证用户身份（如 JWT Token）。
 - **数据库连接**：管理请求级别的数据库会话。

  
def get_db():
  db = SessionLocal()
  try:
    yield db
  finally:
    db.close()

@app.get("/items")
def read_items(db=Depends(get_db)):
  return db.query(Item).all()
  
  - **参数验证**：对请求参数进行额外逻辑校验。
  
def pagination_params(limit: int = Query(10, ge=1, le=100), skip: int = 0):
    if limit % 2 != 0:
        raise HTTPException(status_code=400, detail="limit 必须是偶数")
    return {"limit": limit, "skip": skip}
  ---

  4. FastAPI DI 的优势
  - **逻辑解耦**：路由只关注业务逻辑。
  - **自动注入**：无需手动传参。
  - **复用性**：依赖函数可在多个路由使用。
  - **可测试性**：可用 `dependency_overrides` 替换依赖函数。
  - **层级依赖**：依赖函数可以嵌套依赖其他函数。

  ---

  5. APIRouter 与依赖
  - **APIRouter 可以包含多个路由**，方便模块化管理。
  - 可以在 **整个 APIRouter** 上声明依赖，也可以只在 **某一个路由** 上声明依赖。
  - 示例：
    
@router.get("/{user_id}", dependencies=[Depends(check_admin)])
def get_user(user_id: int):
    return {"id": user_id}
    
  6.  FastAPI 中 `dependencies` 的好处总结
  
   语义清晰
  - 名字直观地表达了用途：就是用来声明“依赖”。
  - 一眼就能看出这是和依赖注入相关的参数，而不是其他功能。

    统一入口
  - 在路由装饰器、APIRouter、甚至全局应用层都用同一个名字 `dependencies`。
  - 开发者不需要记不同的参数名，学习成本低。
  - 统一的 API 设计让代码更一致，团队协作时更容易理解。

   区分副作用依赖与参数依赖
  - **参数依赖**：写在函数参数里，返回值会传给路由函数。
  - **dependencies 参数**：只执行副作用，不传值。
  - 固定名字让框架能明确区分这两类依赖，避免歧义。

   可扩展性
  - `dependencies` 接受列表，可以放多个依赖函数。
  - 框架会自动按顺序执行，支持认证、日志、权限检查等逻辑的模块化组合。

   与 OpenAPI 文档集成
  - FastAPI 会把 `dependencies` 中的依赖函数纳入 OpenAPI 文档生成逻辑。
  - 例如认证依赖会自动出现在接口的安全说明里。
  - 固定名字保证了文档生成的一致性。

    ---

   📊 总结表

  | 特点         | 好处                           |
  |--------------|--------------------------------|
  | 名字直观     | 一眼看出是依赖相关             |
  | 统一入口     | 路由、APIRouter、应用层一致    |
  | 区分用途     | 副作用依赖 vs 参数依赖         |
  | 列表形式     | 支持多个依赖组合               |
  | 文档集成     | 自动生成安全说明               |

  
  **Pydantic alias 与 validator 简要总结**

  **alias**：用于定义字段别名，解决外部数据字段名和内部模型字段名不一致的问题。
  **validator**：用于自定义字段校验逻辑，在模型初始化时对字段值进行额外检查或转换。

   对比
  | 功能 | alias | validator |
  |------|-------|-----------|
  | 作用 | 字段别名映射 | 自定义校验逻辑 |
  | 解决问题 | 外部字段名和内部字段名不一致 | 内置校验不够，需要额外逻辑 |
  | 使用位置 | `Field(..., alias="xxx")` | `@validator("field_name")` |
  | 典型场景 | 前后端/第三方 API 字段不同 | 检查格式、范围、自动转换 |

关于PostgreSQL

PostgreSQL 查询语句速查表

1. 基本结构

SELECT column_name(s), aggregate_function(column_name)
FROM table_name
WHERE condition
GROUP BY column_name(s)
HAVING condition
ORDER BY column_name(s) ASC|DESC;

---

2. 关键字说明

SELECT

• 指定要查询的列或聚合结果。 *: 通配符，表示选择表中的所有列
• 示例：

  SELECT name, age FROM Students;

FROM

• 指定数据来源表。
• 示例：

  SELECT * FROM Sales;

WHERE

• 在分组前过滤行。
• 示例：

  SELECT * FROM Sales WHERE Amount > 100;

GROUP BY

• 按列分组，每组生成一行结果。
• 示例：

  SELECT Product, SUM(Amount)
  FROM Sales
  GROUP BY Product;

HAVING

• 在分组后过滤分组结果，常与聚合函数一起使用。
• 示例：

  SELECT Product, SUM(Amount) AS TotalSales
  FROM Sales
  GROUP BY Product
  HAVING SUM(Amount) > 200;

ORDER BY

• 对结果排序。
• 示例：

  SELECT Product, SUM(Amount) AS TotalSales
  FROM Sales
  GROUP BY Product
  ORDER BY TotalSales DESC;

---

3. 常用聚合函数

函数	作用
COUNT()	统计数量
SUM()	求和
AVG()	平均值
MAX()	最大值
MIN()	最小值

---

4. WHERE vs HAVING 对比

特性	WHERE	HAVING
作用阶段	分组前过滤行	分组后过滤分组结果
可否用聚合函数	❌ 不可	✅ 可
常见用途	过滤原始数据	过滤统计结果

---

5. 执行顺序

1. FROM → 选表
2. WHERE → 过滤行
3. GROUP BY → 分组
4. HAVING → 过滤分组结果
5. SELECT → 选择列和聚合结果
6. ORDER BY → 排序

---

---

关于 SQLModel — 把 SQLAlchemy 和 Pydantic 缝在一起

学了 PostgreSQL 的 SQL 语句之后，下一个问题就是：怎么在 Python 里操作数据库？

原生方案是写裸 SQL 字符串然后用 asyncpg 执行，但很快你就会发现两个痛点：

1. SQL 字符串没有类型检查，字段名写错了编译期发现不了
2. 查询结果是一个 tuple/dict，要手动映射成 Python 对象

于是 ORM 出场了。Python 这边最主流的是 SQLAlchemy，但它的模型定义和 Pydantic 的 schema 是两套东西，你得维护两份几乎一样的代码。SQLModel 解决了这个问题：一个类同时是数据库表 + Pydantic 模型。

定义一张表

关于 SQLModel — 把 SQLAlchemy 和 Pydantic 缝在一起

学了 PostgreSQL 的 SQL 语句之后，下一个问题就是：怎么在 Python 里操作数据库？

原生方案是写裸 SQL 字符串然后用 asyncpg 执行，但很快你就会发现两个痛点：

1. SQL 字符串没有类型检查，字段名写错了编译期发现不了
2. 查询结果是一个 tuple/dict，要手动映射成 Python 对象

于是 ORM 出场了。Python 这边最主流的是 SQLAlchemy，但它的模型定义和 Pydantic 的 schema 是两套东西，你得维护两份几乎一样的代码。SQLModel 解决了这个问题：一个类同时是数据库表 + Pydantic 模型。

SQLModel = SQLAlchemy (数据库侧) + Pydantic (校验侧)

定义一张表

from sqlmodel import Field, SQLModel

class Task(SQLModel, table=True):          # table=True → 这是一张数据库表
    id: int | None = Field(default=None, primary_key=True)
    content: str = Field(min_length=1)      # Pydantic 校验照样生效
    is_done: bool = Field(default=False)

注意这个 table=True：

• 有它 → 数据库表，SQLAlchemy 会拿它做映射
• 没它 → 纯 Pydantic 模型，只做数据校验，不碰数据库

这就是 SQLModel 最妙的地方：同一个类，既是 ORM 模型又是校验模型，不用写两份。

但是！请求和响应通常不是同一回事

实际项目中你一般会拆成三个：

class TaskBase(SQLModel):        # 公共字段
    content: str

class Task(TaskBase, table=True):  # 数据库表（含内部字段）
    id: int | None = Field(default=None, primary_key=True)
    updated_at: datetime = Field(...)
    attachment_path: str | None = None   # 内部路径，不想暴露给前端

class TaskCreate(TaskBase):      # 创建请求（只要 content）
    pass

class TaskRead(TaskBase):        # 查询响应（含 id、时间等）
    id: int
    is_done: bool
    created_at: datetime
    model_config = {"from_attributes": True}  # 允许从 ORM 对象直接构造

一句话：Base 抽公共，Table 管存储，Create/Read 管输入输出。 各司其职。

---

异步数据库操作 — 为什么要 async？

FastAPI 本身就是异步框架，如果你的数据库操作是同步的（阻塞），那每个请求都会卡住整个线程，异步的优势全没了。

同步 vs 异步对比

同步（阻塞）:
  请求1 ──[查DB,等500ms]── 返回
  请求2          ──[查DB,等500ms]── 返回
  请求3                  ──[查DB,等500ms]── 返回
  总耗时: 1500ms

异步（非阻塞）:
  请求1 ──[发查询]──等──[收结果]── 返回
  请求2 ──[发查询]──等──[收结果]── 返回
  请求3 ──[发查询]──等──[收结果]── 返回
  总耗时: ~500ms  ← 三个请求同时在"等"

异步不加速单次查询，但可以让服务器同时处理更多请求。 对于 IO 密集的 Web 服务来说，这是质的区别。

异步引擎的配置

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/db",   # 注意是 asyncpg，不是 psycopg
    pool_size=10,           # 连接池大小
    max_overflow=20,        # 池满了最多再创建 20 个
    pool_pre_ping=True,     # 拿连接前先 ping 一下，确认没断
    pool_recycle=3600,      # 1 小时后强制回收连接（防 PostgreSQL 自动断开）
)

AsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

expire_on_commit=False 这个坑我踩过：默认 True 的情况下，commit 之后你再访问 ORM 对象的属性，SQLAlchemy 会报 “Instance is expired”，因为默认行为是 commit 后把所有属性标记为过期，下次访问时重新查库。异步场景下这个重查操作如果没有在正确的协程里执行，直接炸。 关掉就完事了。

Session 依赖注入

async def get_session():
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()        # 正常结束 → 提交
        except HTTPException:
            raise                          # 业务拒绝（如 404）→ 不 rollback
        except Exception:
            await session.rollback()       # 意外异常 → 回滚
            raise

把这个函数挂在 Depends(get_session) 上，FastAPI 会：

1. 请求进来 → 创建 session
2. 路由函数执行 → 用这个 session 做 CRUD
3. 正常返回 → 自动 commit
4. 抛异常 → 自动 rollback

CRUD 函数里绝对不要自己 commit！ commit 的权力只属于 session 的拥有者（即 get_session），CRUD 最多只做 flush。这是一个铁律，违反了事务边界就全乱了。

---

关于 Alembic — 数据库的 Git

代码用 Git 管版本，数据库表结构呢？Alembic 就是数据库的版本控制工具。

alembic/
├── env.py              # 配置：连哪个库、用哪个 metadata
├── script.py.mako      # 迁移文件模板
└── versions/
    └── 20260317_0001_init_tables.py   # 第 1 个版本：建表

核心命令

alembic revision --autogenerate -m "add user table"   # 自动生成迁移文件
alembic upgrade head     # 应用所有未执行的迁移
alembic downgrade -1     # 回退一个版本

迁移文件长什么样

def upgrade():
    op.create_table(
        "task",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("content", sa.String(), nullable=False),
        sa.PrimaryKeyConstraint("id"),
    )
    op.create_index(op.f("ix_task_content"), "task", ["content"])

def downgrade():
    op.drop_index(op.f("ix_task_content"), table_name="task")
    op.drop_table("task")

每个迁移都有 upgrade() 和 downgrade()，确保可以前进也可以回退。一旦迁移文件提交到 Git，就绝对不能改了 — 因为别人可能已经执行过了。 要改只能再新建一个迁移文件。

---

文件上传安全 — 这次我真的被教育了

做到简历 ZIP 上传功能的时候，我突然意识到文件上传这块水太深了。一个简单的 file.read() + open(path, "wb").write() 背后全是坑。

攻击面一览

攻击方式	原理	后果
扩展名伪造	把 .exe 改成 .jpg 上传	绕过前端校验
Content-Type 伪造	改 HTTP 头里的 MIME 类型	绕过简单后端校验
Magic Number 伪造	文件头也伪造了	绕过文件头校验
Zip Slip	ZIP 里放 ../../etc/passwd	覆盖系统文件
Zip Bomb	一个 1KB 的 ZIP 解压出 100TB	撑爆磁盘
符号链接	ZIP 里放符号链接指向 /etc/shadow	读取敏感文件

我的防御链路（从外到内）

用户上传文件
    ↓
① 扩展名白名单        → ".exe" 直接拒绝
    ↓
② Content-Type 校验   → MIME 类型必须在白名单
    ↓
③ Magic Number 校验   → 读文件头 4~16 字节，验证真实类型
    ↓                      JPEG: FF D8 FF
    ↓                      PNG:  89 50 4E 47 0D 0A 1A 0A
    ↓                      PDF:  %PDF-
    ↓                      ZIP:  PK\x03\x04
    ↓
④ python-magic MIME复核 → 基于文件内容（非扩展名/Content-Type）检测真实 MIME
    ↓
⑤ DOCX 结构校验       → OOXML 必须含 [Content_Types].xml + word/document.xml
    ↓
⑥ 大小限制            → 单文件、总解压量都设上限
    ↓
⑦ 写入磁盘

对于 ZIP 解压，还有额外的：

⑧ 路径穿越检测 → 禁止 ../ 和绝对路径
⑨ 符号链接检测 → os.path.is_symlink()
⑩ 文件数限制   → 最多 300 个文件
⑪ 总解压量限制 → 最多 200MB
⑫ 白名单过滤   → 只复制 resume.json 里引用过的文件，其余丢弃

最关键的心得：永远不要信任任何一个单一的校验（因为都可以伪造），要层层设防。 这叫”纵深防御”。

顺便说一下阻塞 IO 和异步的配合

文件读写是阻塞操作，直接放在 async 函数里会卡住整个事件循环。

# ❌ 错误：直接在协程里同步写文件
with open(path, "wb") as f:
    f.write(data)   # 把事件循环卡住了

# ✅ 正确：扔到线程池
await asyncio.to_thread(_save_file, path, data)

asyncio.to_thread 把同步函数丢到线程池里跑，主线程的事件循环继续处理其他请求。凡是涉及 open()、zipfile、shutil 的，全部走这个通道。

---

Pydantic Settings — 别再硬编码配置了

一开始写项目都喜欢把配置直接写死在代码里：

UPLOAD_DIR = "uploads"         # 改环境你就哭吧
MAX_SIZE = 50 * 1024 * 1024

然后某天要部署到服务器，路径不一样、大小限制不一样……开始到处改代码。pydantic-settings 就是来解决这个的。

配置分层

config.py         → 定义结构 + 默认值（代码级别，提交到 Git）
.env              → 环境差异（如 DATABASE_URL），每个环境不同
.env.local        → 本地私有覆盖（不提交 Git），放你自己的调试配置
.env.example      → 模板文件（提交 Git），告诉队友哪些配置需要填

实现

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    DATABASE_URL: str                         # 必填，没有默认值，不写就报错
    DEBUG: bool = False                       # 有默认值
    MAX_ATTACHMENT_BYTES: int = 50 * 1024 * 1024
    ALLOWED_IMAGE_EXTS: set[str] = {".jpg", ".jpeg", ".png", ".gif", ".webp"}

    model_config = SettingsConfigDict(
        env_file=(".env", ".env.local"),       # 后者的值覆盖前者
        env_file_encoding="utf-8",
    )

@lru_cache    # 单例，只读一次文件
def get_settings():
    return Settings()

settings = get_settings()

项目里所有地方都 from app.core.config import settings，永远不要硬编码路径和限制。这是后期维护的保命操作。

---

错误处理 — 让前端能”看懂”你的错误

以前写接口，错误返回千奇百怪：

{"detail": "Task not found"}
{"error": "Something went wrong"}
"Internal Server Error"

前端同学根本没法写 if-else 来处理不同错误。统一结构化错误码：

{
  "detail": {
    "code": "TASK_NOT_FOUND",
    "message": "Task not found"
  }
}

前端只要 if (detail.code === "TASK_NOT_FOUND") 就行了，不用去匹配自然语言字符串。

错误码集中管理

# app/core/error_codes.py
TASK_NOT_FOUND = "TASK_NOT_FOUND"
TASK_CONFLICT = "TASK_CONFLICT"
ATTACHMENT_TYPE_NOT_ALLOWED = "ATTACHMENT_TYPE_NOT_ALLOWED"
ATTACHMENT_MAGIC_NUMBER_INVALID = "ATTACHMENT_MAGIC_NUMBER_INVALID"
# ... 一共 35 个错误码

路由中的用法

def _api_error(status_code: int, code: str, message: str) -> HTTPException:
    return HTTPException(
        status_code=status_code,
        detail={"code": code, "message": message}
    )

# 使用
raise _api_error(404, ec.TASK_NOT_FOUND, "Task not found")
raise _api_error(400, ec.ATTACHMENT_MAGIC_NUMBER_INVALID, "Invalid JPEG signature")

**这样就形成了前后端的”错误契约”**：code 是契约本身（稳定，不可随意改），message 是给人看的（可以改措辞）。

---

测试 — 最容易被跳过但其实最救命的

说实话，一开始我也觉得写测试浪费时间，”我自己测一下不就行了？”

然后经历了一次：改了 A 功能，B 功能默默炸了，两天后才发现。测试不是测”现在对不对”，而是防止”未来被人改坏”。

测试金字塔

      /\
     /E2E\           ← 少：全链路测试
    /------\
   / 集成测试 \       ← 中：数据库+接口联测
  /----------\
 /  单元测试    \     ← 多：单个函数/类的测试
/--------------\

Mock 的艺术

API 测试不想连真实数据库怎么办？monkeypatch：

def test_read_task_not_found(client: TestClient, monkeypatch):
    async def fake_get_task(session, task_id):
        return None    # 模拟"查不到"

    monkeypatch.setattr(task_api.crud_task, "get_task", fake_get_task)

    response = client.get("/tasks/999")
    assert response.status_code == 404
    detail = response.json()["detail"]
    assert detail["code"] == "TASK_NOT_FOUND"

FastAPI 的 TestClient 不需要启动服务器，直接内存里发请求，测试秒跑完。

conftest.py — 共享装备库

# tests/conftest.py
@pytest.fixture
def client():
    app = FastAPI()
    app.include_router(task_router)
    app.dependency_overrides[get_session] = fake_session   # 替换掉真实 DB
    with TestClient(app) as c:
        yield c

conftest.py 里的 fixture 会被同级目录下的所有测试文件自动发现和复用，不用每个文件都写一遍。

CI — 每次提交自动跑

# .github/workflows/ci.yml
steps:
  - name: Lint (ruff)
    run: ruff check app tests
  - name: Type check (mypy)
    run: mypy app
  - name: Run tests
    run: pytest -q tests/

每次 push 或提 PR，GitHub Actions 自动跑一遍 ruff + mypy + pytest。任何一步挂了，PR 上直接一个大红叉，绝不允许烂代码合进主分支。

---

Docker — “在我机器上能跑啊”

手动部署最烦的就是环境差异：Python 版本、系统库、路径……Docker 一句话解决：把代码和环境一起打包。

FROM python:3.11-slim
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1
WORKDIR /app
COPY requirement.txt .
RUN pip install --no-cache-dir -r requirement.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

配合 docker-compose 把 PostgreSQL 一起拉起来：

services:
  db:
    image: postgres:16
    environment:
      POSTGRES_USER: arms
      POSTGRES_PASSWORD: arms
      POSTGRES_DB: arms
    ports:
      - "5432:5432"

  app:
    build: .
    depends_on:
      - db
    environment:
      DATABASE_URL: postgresql+asyncpg://arms:arms@db:5432/arms
    ports:
      - "8000:8000"

一条 docker compose up --build，环境就绪。新队友拉下来不用装任何东西（除了 Docker），直接跑。

---

项目整体心得

这次做 ARMS 后端，我觉得自己最大的成长是：

1. “能用”和”能抗住攻击”是两码事

写 CRUD 很容易，但文件上传这个口子如果不层层设防，分分钟被人搞。Magic number、MIME 复核、Zip Slip、Zip Bomb……每一个攻击向量都要有对应的防御。不是过度设计，是基本素养。

2. 事务边界要想清楚

CRUD 层不 commit，Service 层决定何时提交，Session 层统一处理 commit/rollback。这三层职责一旦混乱，数据一致性就没了。**”谁拥有 session，谁负责提交”** — 这个原则比具体代码更重要。

3. 异步不是银弹

async/await 只对 IO 密集型任务有效。CPU 密集的计算（比如图片处理）即使写在 async 函数里一样会卡。关键是识别出阻塞操作（文件读写、网络请求、数据库查询），然后该丢线程池就丢线程池。

4. 配置驱动 > 硬编码

所有路径、大小限制、白名单都走 settings，换来的是：切环境时改一个 .env 文件就搞定，不用满世界搜索代码。

5. 测试是给自己写的

手动测一次两分钟，自动化测一次两秒。项目越小越应该写测试 — 因为小项目容易改，改动就越容易引入 bug。测试就是你的安全网。

6. pre-commit 是好文明

提交前自动 ruff + mypy，把低级错误拦截在本地。省得推到 GitHub 上 CI 挂了再灰溜溜地修。

---

技术栈小结

层	选型	一句话理由
框架	FastAPI	快、自动文档、原生异步
ORM	SQLModel	一套模型同时搞定 DB + 校验
数据库	PostgreSQL + asyncpg	真正的异步 PostgreSQL 驱动
迁移	Alembic	数据库的 Git
配置	pydantic-settings	多环境分层，类型安全
校验	Pydantic v2	字段级、模型级自定义校验
测试	pytest + httpx + TestClient	轻量、mock 方便
质量	ruff + mypy + pre-commit	提交前自动检查
部署	Docker + docker-compose	一键启动，环境一致
CI	GitHub Actions	每次 push 自动 lint + 类型检查 + 跑测试

---

下一步想做 / 正在做的

• 补充数据库集成测试（现在全是 mock，事务行为没有真正被验证）
• 把 task.py 路由文件拆一拆，510 行太长了
• 引入结构化日志（JSON 格式日志，方便接入 ELK / Grafana）
• 加 health check 端点
• 如果用户量上来，加上 Redis 缓存

后端之路还在继续……虽然经常踩坑到凌晨三点，但每解决一个问题那种”啊哈！”的瞬间，值了。