返回随笔AI & Python
完成ENGINEERING NOTE

Python-FastAPI

从异步编程、回调与 async/await 开始，整理 FastAPI 的请求处理、数据模型、依赖注入和异步接口开发基础。

后端Python

相关知识

异步编程

异步编程是一种处理并发操作的方法，使得程序可以在等待耗时操作（如文件读写、网络请求或数据库查询），完成的同时继续执行其他任务。与同步编程不同，异步编程不要求程序按顺序等待每个操作完成，而是通过回调、Promise或者async/await等机制，在操作完成时再执行特定的逻辑。 通俗的讲，异步编程即是允许我们在执行一个长时间的任务时，程序不需要等待，而是继续执行之后的代码，知道这些任务完成之后再回来通知你，通常是以回调函数(callback)的方式。这种编程模式避免了程序的堵塞，大大提高了cpu的执行效率。尤其适合一些io密集的操作。例如需要经常进行网络操作，数据库访问的应用。

回调函数（Callback Function）

回调函数是完成一个异步操作后执行的操作。例如在网络读写成功后，将会调用一个函数用于处理响应数据，这个函数称为回调函数。

Promise

Promise是一个对象，用来表示一个异步操作完成或失败后的结果值。Promise具有三种状态：pending(进行中)，fulfilled(已成功)和rejected(已失败)。Promise通过then、 catch和finally方法来处理异步操作的结果。 例如，我们在编写前端操作时，通常会使用Fetch库，这是一个使用Promise接口的库：

fetch("jskring.top").then((response) => {
    response.json();
});

fetch函数的返回值是一个Promise对象，它记录了网络操作的结果，如果请求成功完成，那么回调函数会被调起。请求的结果也会以参数的形式传递进来。而且，调用then函数后的返回值依然是一个Promise，也就是说可以这样将多个then函数或者catch和finally函数串联起来：

fetch("jskring.top").then((response) => {
    response.json();
}).then((json) => {
    console.log(json);
}).catch((err) => {
    console.log(err);
}).finally(() => {
    console.log("请求完成");
})

async/await

async/await是Promise的语法糖，它可以让异步操作写起来像同步操作一样。 async/await函数总是返回一个Promise，await用于等待Promise完成，并 返回Promise的结果。它使得代码更易读，更易维护。但是需要注意的是：await看上去是需要暂停函数的执行，但是在等待过程中，js同样可以处理其他任务。比如更新界面，运行其他程序代码等等。因为await底层也是基于Promise和事件循环机制实现的。 使用：

  • 首先我们需要将将async关键字将函数标记为异步函数（异步函数指的是返回值为Promise对象的函数），比如之前用到的fetch()就是一个异步函数。
  • 在异步函数中，我们可以调用其他异步函数，不过我们不需要使用then()，而是使用一个更加简洁的await语法，await会等待Promise完成之后直接返回最终的结果。

安装并进入.venv虚拟Python环境

python的环境配置一直是一个头疼的问题。两个不同的项目可能需要不同的依赖版本，这可能会引起冲突。所以，为每一个项目单独构建一虚拟的python环境是非常有必要的。

创建venv

你可以运行

python -m venv [venv_name]

来创建一个虚拟环境，但是也可以在vscode中按下CTRL+SHIFT+P后输入python create environment找到创建python环境，用vscode自动创建一个名为.venv的环境。这样会更加规范。

进入venv

在项目根目录下运行

.venv/Scripts/activate

即可进入虚拟环境，成功的标志是命令行前出现(.venv) 。然后你就可以在这个环境下安装依赖（例如FastAPI）了。

FastAPI

FastAPI 是一个轻量级、用于构建 API 的现代、快速（高性能）的 web 框架，使用 Python 并基于标准的 Python 类型提示。

多文件项目结构

fastapi项目以app.py为入口文件，只能在这里面使用app注解，可以通过引入其他文件夹下的包来区分功能。 项目格式如下：

.
├── app                  # 「app」是一个 Python 包
│   ├── __init__.py      # 这个文件使「app」成为一个 Python 包
│   ├── main.py          # 「main」模块，例如 import app.main
│   ├── dependencies.py  # 「dependencies」模块，例如 import app.dependencies
│   └── routers          # 「routers」是一个「Python 子包」
│   │   ├── __init__.py  # 使「routers」成为一个「Python 子包」
│   │   ├── items.py     # 「items」子模块，例如 import app.routers.items
│   │   └── users.py     # 「users」子模块，例如 import app.routers.users
│   └── internal         # 「internal」是一个「Python 子包」
│       ├── __init__.py  # 使「internal」成为一个「Python 子包」
│       └── admin.py     # 「admin」子模块，例如 import app.internal.admin

init.py作为一个包的标示，在第一次运行时需要创建，之后就不需要了（存疑）

连接MySql

fastapi连接MySql真是个麻烦的事情，就这方面比mybatis麻烦很多。但是把准备工作做好了写起来很舒服，和mybatis-plus类似，不需要手动写sql了。

安装SQLModel和sql-connector-python

sql-connector-python是python连接数据库的驱动，SQLModel是FastAPI官方的库，可以很方便地创建数据库映射类和数据库连接。使用pip install即可安装。

创建引擎（Engine）

SQLModel的Engine是一个用来与数据库保持连接的对象。 使用这两行代码创建引擎：

DATABASE_URL = "mysql+mysqlconnector://username:password@localhost:3306/database_name"
engine = create_engine(DATABASE_URL)

创建会话（Session）依赖项

会话是程序与数据库交互的对象。它可以暂存程序中的数据对象，然后更新数据库；亦可以查询数据库，然后将结果储存在程序中的数据对象了里。 我们使用yield为每一个请求提供一个新的Session，这样确保每一个请求都使用一个单独的Session。

def get_session():
    with Session(engine) as session:
        yield session

SessionDep = Annotated[Session, Depends(get_session)]

with-yield块可以使得函数传出一个资源，然后在这个资源执行完后对其销毁。 Annotated是python自带的一个Typing，它可以向一个类型的注解中增加元数据 我们将上面的两个代码块放在同一个文件当中，命名为db.py，这个文件用来连接数据库。

创建模型

模型就像Springboot中的实体类，但是在这里需要指明字段的一些属性，比如默认值，是否是主键、外键、索引等等。这些指定通过sqlmodel的Field函数进行。

class Drink(SQLModel, table=True):
		__tablename__ = "drink"
    id: int | None = Field(default=None, primary_key=True)
    name: str
    description: str | None = None
    price: float

default字段指定了为空时的默认值，primary_key说明这是一个主键。table=True表明这是一个直接对接数据库的实体类。它应该表示 SQL 数据库中的一个表，而不仅仅是一个数据模型。 默认情况下，表名应该是模型名字的小写。但是你可以通过__tablename__字段手动指定绑定的表名。

实现增删改查操作

如果需要实现数据库操作，需要在router后面定义的函数参数中加上session参量，而且使用SessionDep构造。

  1. 查

@router.get(“/meal/listAllMeal”) async def listAllMeal(session: SessionDep) -> list[Meal]:

select(Meal)使用 select 函数构建查询语句，这里表示查询所有的 Meal 对象

使用 session.exec() 执行查询语句，返回一个结果生成器

使用 all() 方法将结果生成器中的所有结果转换为列表

meals = session.exec(select(Meal)).all()
return meals

@router.get(“/meal/{mealid}”) async def getMealById(mealid: int, session: SessionDep) -> Meal:

表示查询id为mealid的Meal对象

meal = session.get(Meal, mealid)
if not meal:
    raise HTTPException(status_code=404, detail="Meal not found")
return meal
```
上面两个函数分别实现了列表和id查询的操作。

2. 增 python @router.post("/meal/addMeal") async def addMeal(meal: Meal, session: SessionDep): session.add(meal) session.commit() session.refresh(meal) return f"add success:{meal}" 这个很简单，只需要调用session.add()函数就可以了。commit()函数用来将更改推送至数据库，refresh()函数用来同步数据库与程序中的更改。 3. 删 python @router.delete("/meal/{mealid}") async def deleteMealById(mealid: int, session: SessionDep): meal = session.get(Meal, mealid) if not meal: raise HTTPException(status_code=404, detail="Meal not found") session.delete(meal) session.commit() return "delete success" 进行删除操作时，需要先使用一次查操作获得需要删除的对象，并检验改对象是否存在。如果该对象存在，则将它传给delete()函数，并commit更改。 4. 改 python @router.post("/meal/update") async def update(updateMeal: Meal, session: SessionDep): meal = session.get(Meal, updateMeal.id) if not meal: raise HTTPException(status_code=404, detail="Meal not found") # 替换掉数据库对象的字段数据 mealData = updateMeal.dict(exclude_unset=True) for key, value in mealData.items(): setattr(meal, key, value) session.add(meal) session.commit() session.refresh(meal) return "update success" 改部分的逻辑比较复杂。 第一步：通过前端传来的对象获取需要更新的id，通过id获得需要更新的行的引用。需要注意，session.get()获得的是数据库对象的引用，也就是说，对这个对象的修改在commit后会直接作用与这个数据库对象。 第二步：替换掉meal中需要更新的字段。 第三步：将对meal引用的更改commit到数据库中。 如果这里不获取meal，直接把updateMeal add到数据库，会报错，因为这相当于增加了一个新的行。必须获取数据库对象后，对该对象修改并commit。