FastAPI学习参考文档

· · · FastAPI ·

### 安装FastAPI

需要安装所有的可选依赖及对应功能，包括了 uvicorn ，你可以将其⽤作运⾏代码的服务器。

`pip install fastapi[all] -i https://mirrors.aliyun.com/pypi/simple/`

你也可以分开来安装：并且安装 uvicorn 来作为服务器

`pip install fastapi`,     `pip install "uvicorn[standard]"`

#### PyCharm中配置项目启动

1. 点击右上角：Current File   ->   Edit Configurations -> +  -> FastAPI 
2. 配置项目启动：
   - Name: main
   - Applicatiojn file:  选择当前项目的启动文件：`D:\PycharmProjects\fastapi-study\main.py`
   - Uvicorn options:  `--relaod`

创建启动文件main.py

```python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def root():
    return 'hello'

@app.get("/hello/{name}")
def say_hello(name: str):
    return {"message": f"Hello {name}"}

# 可以不要
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)
```

启动方式

1. 通过pycharm启动按钮（开发模式），上面配置项目启动方式

2. 通过uvicorn命令启动（⽣产模式）：`uvicorn main:app --reload`
   - 命令含义如下:
     1. main ： main.py ⽂件（⼀个 Python「模块」）。
     2. app ：在 main.py ⽂件中通过 app = FastAPI() 创建的对象。
     3. --reload ：让服务器在更新代码后重新启动。仅在开发时使⽤该选项。

3. 在main.py中定义main函数（备⽤）

#### swagger访问不到问题处理

此时访问接口和文档：http://127.0.0.1:8000/,http://127.0.0.1:8000/docs，但是文档打不开，因为依赖的swagger的css/js使用的是国外的地址，需要下载到本地来依赖。通过源码`fastapi/docs.py`中找到了相关的地址：

```
https://cdn.jsdelivr.net/npm/swagger-ui-dist@5/swagger-ui-bundle.js
https://cdn.jsdelivr.net/npm/swagger-ui-dist@5/swagger-ui.css
https://cdn.jsdelivr.net/npm/swagger-ui-dist@5/favicon.png
```

将其下载，放到根目录下的static目录下，然后对源码`fastapi/docs.py`中找到了相关的地址替换

```
/static/swagger-ui-bundle.js
/static/swagger-ui.css"
/static/favicon.png
```

在app中注册static⽬录，修改main.py中

```python
from starlette.staticfiles import StaticFiles

app = FastAPI()
app.mount("/static", StaticFiles(directory="static"), name="static")
```

### FastAPI的路由和请求参数

#### FastAPI的路由

Route路由, 是⼀种映射关系！路由是把客⼾端请求的url路径与视图函数进⾏绑定映射的⼀种关系。

- 路径 ：是 /
- 请求⽅法：是 get
- 函数 ：是位于「装饰器」下⽅的函数（位于 @app.get("/") 下⽅）

```python
# 路径操作装饰器参数
# tags ⽂档标题
# summary ⽂档总结
# description ⽂档描述
# response_description 响应详情内容
# deprecated 接⼝⽂档是否废弃
@app.post("/items", tags=["这是items测试接⼝"],
          summary="this is items测试 summary",
          description="this is items测试 description...",
          response_description="this is items测试 response_description",
          deprecated=False)
def test():
    return {"items": "items数据"}
```

#### 路由分发

在main.py中的app作为主路由

```python
from apps.app01.urls import shop
from apps.app02.urls import user

app = FastAPI()
app.mount("/static", StaticFiles(directory="static"), name="static")

# 路由分发include_router
app.include_router(shop, prefix="/shop", tags=["购物中⼼接⼝"])
app.include_router(user, prefix="/user", tags=["⽤⼾中⼼接⼝"])
```

项目目录：

```
apps
--app01
  --urls.py
--app02
  --urls.py
main.py
static
```

其他的⼦路由`app01/urls.py`：

```python
from fastapi import APIRouter
shop = APIRouter()
@shop.get("/food")
def shop_food():
    return {"shop": "food"}
@shop.get("/bed")
def shop_bed():
    return {"shop": "bed"}
```

#### 路由参数

请求参数，通过路由路径携带的⽅式；我们称之为路由传参。参数也叫： 路由参数。你可以使⽤与Python 格式化字符串相同的语法来声明路径"参数"或"变量"：

```python
@app.get("/items/{item_id}")
def read_item(item_id):
    return {"item_id": item_id}
```

路径参数 item_id 的值将作为参数 item_id 传递给你的函数。所以，如果你运⾏⽰例并访问[http://127.0.0.1:8000/items/foo]，将会看到如下响应：`{"item_id":"foo"}`

##### 定义参数的类型

你可以使⽤标准的 Python 类型标注为函数中的路径参数声明类型。

```
@app.get("/items/{item_id}")
def read_item(item_id: int):
    return {"item_id": item_id}
```

在这个例⼦中， item_id 被声明为 int 类型。FastAPI 通过上⾯的类型声明提供了对请求的⾃动"解析"。

同时还提供数据校验功能：

如果你通过浏览器访问 [http://127.0.0.1:8000/items/foo]，你会看到⼀个清晰可读的 HTTP 错误：

```json
{
    "detail": [
        {
            "type": "int_parsing",
            "loc": [
                "path",
                "item_id"
            ],
            "msg": "Input should be a valid integer, unable to parse string as an integer",
            "input": "we"
        }
    ]
}
```

因为路径参数 item_id 传⼊的值为 "foo" ，它不是⼀个 int 。你可以使⽤同样的类型声明来声明 str 、 float 、 bool 以及许多其他的复合数据类型。

##### 路由匹配的顺序

**由于路由匹配操作是按顺序依次运⾏的**，你需要确保路径` /users/me` 声明在路径`/users/{user_id} `之前：

```python
@user.get("/users/me")
def read_user_me():
    return {"user_id": "the current user"}

@user.get("/users/{user_id}")
def read_user(user_id: str):
    return {"user_id": user_id}
```

否则， /users/{user_id} 的路径还将与 /users/me 相匹配，"认为"⾃⼰正在接收⼀个值为"me" 的 user_id 参数。

##### 预设值参数

如果你有⼀个接收路径参数的路径操作，但你希望预先设定可能的有效参数值，则可以使⽤标准的Python Enum 类型。

Python中的枚举数据类型：是指列出有穷集合中的所有元素，即⼀⼀列举的意思。在Python中，枚举可以视为是⼀种数据类型，当⼀个变量的取值只有⼏种有限的情况时，我们可以将其声明为枚举类型。

```python
from enum import Enum

class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"

@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
    if model_name is ModelName.alexnet:
        return {"model_name": model_name, "message": "Deep Learning FTW!"}
    if model_name.value == "lenet":
        return {"model_name": model_name, "message": "LeCNN all the images"}
    return {"model_name": model_name, "message": "Have some residuals"}
```

#### 请求URL传参

##### URL传参

url请求参数是通过url请求地址携带的，例如，在以下 url 中：http://127.0.0.1:8000/items/?skip=0&limit=10

这些请求参数是键值对的集合，这些键值对位于 URL 的 ?之后，并以 & 符号分隔。

请求参数为：

- skip ：对应的值为 0
- limit ：对应的值为 10

当你为它们声明了 Python 类型（在上⾯的⽰例中为 int ）时，它们将转换为该类型并针对该类型进⾏校验。

```python
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]

@app.get("/items/")
def read_item(skip: int = 0, limit: int = 10):
    return fake_items_db[skip: skip + limit]
```

URL请求参数不是路径的固定部分，因此它们可以是可选的，并且可以有默认值。

在上⾯的⽰例中，它们具有 skip=0 和 limit=10 的默认值。你可以将它们的默认值设置为None 。

```python
@app.get("/items/{item_id}")
async def read_item(item_id: str, q: Union[str, None] = None, short: bool = False):
    item = {"item_id": item_id}
    if q:
        item.update({"q": q})
    if not short:
        item.update({"description": "This is an amazing item that has a long description"})
    return item
```

Union类型是⼀种⽤于表⽰⼀个变量可以是多个不同类型之⼀的类型注解。它是typing模块中的⼀个类，可以与其他类型⼀起使⽤，以指定⼀个变量可以接受的多个类型。

**Union类型的语法为：Union[type1, type2, ...]**

这⾥的type1、type2等代表要包含在Union类型中的类型。

**注意：当你想让⼀个查询参数成为必需的，不声明任何默认值就可以**：

```python
@app.get("/items/{item_id}")
def read_user_item(
        item_id: str, needy: str, skip: int = 0, limit: Union[int, None] = None):
    item = {"item_id": item_id, "needy": needy, "skip": skip, "limit": limit}
    return item
```

在这个例⼦中，有3个查询参数：

- needy ，⼀个必需的 str 类型参数。
- skip ，⼀个默认值为 0 的 int 类型参数，也是可选的。
- limit ，⼀个可选的 int 类型参数。

##### ⼀个参数名，多个值

我们还可以使⽤ Query 去接收⼀组值。使⽤ List[str] 来接收⼀组值，⾥⾯的str可以根据需求随意调换。

```python
@app.get("/test5/")
async def read_items(q: Union[List[str], None] = None):
    query_items = {"q": q}
    return query_items
```

http://127.0.0.1:8000/test5?q=wed&q=345,     这是因为参数 q中以⼀个 Python list 的形式接收到查询参数 q 的多个值。

#### 参数校验

Query是 FastAPI专⻔⽤来装饰URL请求参数的类，也可以提供校验。

- 默认值设置，和参数接⼝描述。description是 Query中的⼀个字段，⽤来描述该参数；

- 字符串⻓度校验。注意：max_length和 min_length仅可⽤于 str 类型的参数。
- 正则表达式校验。也就是说该接⼝的查询参数只能是" laoxiao"，不然就会报错，当然其他的正则表达式都是可以拿来⽤的！
- 数值⼤⼩校验。如果参数的类型是int，float就可以采⽤Query进⾏⼤⼩校验，或者范围校验。

```python
@app.get("/search")
def search_items(
    q: str = Query(..., min_length=2, max_length=50, description="搜索关键词"),
    q2: Union[str, None] = Query(default=None, description="参数q2"),
    page: int = Query(1, ge=1, le=100, description="页码"),
    size: int = Query(10, gt=0, le=100, description="每页条数"),
    sort: str = Query("name", regex="^(name|price)$", description="排序字段")
):
    return {
        "q": q,
        "page": page,
        "size": size,
        "sort": sort
    }
```

| 参数名        | 类型    | 默认值        | 说明                                        |
| ------------- | ------- | ------------- | ------------------------------------------- |
| `default`     | `Any`   | `...`（必填） | 默认值，如果设置为 `...` 表示该参数为必填。 |
| `alias`       | `str`   | `None`        | 参数的别名（URL 中的名字）。                |
| `title`       | `str`   | `None`        | 参数的标题（用于 OpenAPI 文档）。           |
| `description` | `str`   | `None`        | 参数描述（用于 OpenAPI 文档）。             |
| `min_length`  | `int`   | `None`        | 字符串最小长度。                            |
| `max_length`  | `int`   | `None`        | 字符串最大长度。                            |
| `ge`          | `float` | `None`        | 数值最小值（greater than or equal）。       |
| `le`          | `float` | `None`        | 数值最大值（less than or equal）。          |
| `gt`          | `float` | `None`        | 数值必须大于（greater than）。              |
| `lt`          | `float` | `None`        | 数值必须小于（less than）。                 |
| `regex`       | `str`   | `None`        | 正则表达式验证。                            |
| `deprecated`  | `bool`  | `False`       | 是否标记为已弃用。                          |

#### 请求体传参

当你需要将数据从客⼾端(例如浏览器)发送给API时,你将其作为[请求体] （request body）发送,请求体是客⼾端发送给API的数据.响应体是API发送给客⼾端的数据。

```python
from datetime import date
from typing import Union, List, Optional

from pydantic import BaseModel, Field, field_validator

class Addr(BaseModel):
    province: str
    city: str

class User(BaseModel):
    # name: str = Field(regex="^a") # name参数值必须是a开头
    name: str
    age: int = Field(default=0, ge=0, lt=100)  # age参数给默认值0,数值区间范围约束为
    birth: Union[date, None] = None  # birth约束类型为date或者None,默认值为None
    friends: List[int] = []  # friends给默认值为[]
    description: Optional[str] = None  # description约束类型为str或者None,默认值为N
    addr: Addr  # 嵌套类型

@field_validator("name")  # 校验name字段
    def name_must_alpha(cls, value):
        assert value.isalpha(), 'name must be alpha'
        # 断⾔语句,检查value是否全部由字⺟组成,如果value的值不满⾜条件则引发AssertionEr
        # 并且异常错误消息是:"name must be alpha"
        return value

@app.post("/data")
async def data(user: User):
    print(user.name, user.age)
    return user
```

注意： Field 的⼯作⽅式和 Query 、 Path 和 Body 相同，包括它们的参数等等也完全相同。

### SQLAlchemy-2.0中模型定义

#### SQLAlchemy的介绍

SQLAlchemy 是 Python ⽣态系统中最流⾏的 ORM。SQLAlchemy 设计⾮常优雅，分为了两部分⸺底层的 Core 和上层的传统 ORM。在 Python 乃⾄其他语⾔的⼤多数 ORM 中，都没有实现很好的分层设计， ⽐如 django 的 ORM，数据库链接和 ORM 本⾝完全混在⼀起。

**SQLAlchemy 是 Python 中⼀个通过 ORM 操作数据库的框架**。

SQLAlchemy对象关系映射器提供了⼀种⽅法，⽤于将⽤⼾定义的Python类与数据库表相关联，并将这些类（对象）的实例与其对应表中的⾏相关联。它包括⼀个透明地同步对象及其相关⾏之间状态的所有变化的系统，称为⼯作单元，以及根据⽤⼾定义的类及其定义的彼此之间的关系表达数据库查询的系统。

可以让我们使⽤类和对象的⽅式操作数据库，从⽽从繁琐的 sql 语句中解脱出来。

ORM 就是 Object Relational Mapper 的简写，就是关系对象映射器的意思。

#### 数据库引擎

任何SQLAlchemy应⽤程序的开始都是⼀个名为 Engine . 此对象充当连接到特定数据库的中⼼源，提供⼯⼚和称为 connection pool 对于这些数据库连接。引擎通常是⼀个只为特定数据库服务器创建

⼀次的全局对象，并使⽤⼀个URL字符串进⾏配置，该字符串将描述如何连接到数据库主机或后端。

sqlalchemy使⽤ create_engine() 函数从URL⽣成⼀个数据库引擎对象。⽐如:

```py
engine = create_engine(r'sqlite:///C:\path\to\foo.db')
```

##### ⽀持的数据库

URL通常可以包括⽤⼾名、密码、主机名、数据库名以及⽤于其他配置的可选关键字参数。主题格式为：

```python
dialect+driver://username:password@host:port/database
```

注意：你必须安装，你想要的数据库驱动库

1) SQLite数据库

sqlite使⽤python内置模块连接到基于⽂件的数据库 sqlite3 默认情况下。

```python
# Unix/Mac - 4 initial slashes in total
engine = create_engine('sqlite:////absolute/path/to/foo.db')
# Windows
engine = create_engine('sqlite:///C:\\path\\to\\foo.db')
# Windows alternative using raw string
engine = create_engine(r'sqlite:///C:\path\to\foo.db')
# 当前⽬录下的test.db
engine = create_engine(r'sqlite:///test.db', echo=True, future=True)
```

2. MySQL数据库

mysql⽅⾔使⽤mysql python作为默认dbapi。mysql dbapis有很多，包括pymysql 和mysqlclient：

```python
# default
engine = create_engine('mysql://scott:tiger@localhost/foo?charset=utf8mb4')
# mysqlclient (a maintained fork of MySQL-Python)
engine = create_engine('mysql+mysqldb://scott:tiger@localhost/foo?charset=utf8mb4')
# PyMySQL
engine = create_engine('mysql+pymysql://scott:tiger@localhost/foo?charset=utf8mb4')
```

3. Microsoft SQL数据库

SQL Server⽅⾔使⽤pyodbc作为默认dbapi。PYMSSQL也可⽤:

```python
# pyodbc
engine = create_engine('mssql+pyodbc://scott:tiger@mydsn')
# pymssql
engine = create_engine('mssql+pymssql://scott:tiger@hostname:port/dbname')
```

##### 数据库引擎的参数

```python
from sqlalchemy import create_engine

engine = create_engine(
    url,                    # 数据库连接字符串（必填）
    *,                      # 以下为关键字参数
    echo=False,             # 是否打印SQL日志
    pool_size=5,            # 连接池大小
    max_overflow=10,        # 超出pool_size后可再创建多少连接
    pool_recycle=-1,        # 多少秒后回收连接（防止数据库主动断开）
    pool_pre_ping=False,    # 每次取出连接时先ping一下
    pool_timeout=30,        # 获取连接的超时时间（秒）
    future=True,            # 使用SQLAlchemy 2.0风格
    isolation_level=None,   # 设置事务隔离级别
    connect_args=None,      # 额外传给DBAPI的参数
    execution_options=None, # 全局执行选项
    **dialect_kwargs        # 其他方言特定参数
)

engine = create_engine('mysql+pymysql://root:123123@localhost/test_db2?charset=utf8mb4', echo=True, future=True, pool_size=10)

from sqlalchemy.orm import Session
def get_session():
    session = Session(bind=engine)
    try:
        yield session
    finally:
        session.close()  # 每次用完session，会自动关闭
```

1. echo=False   -- 如果为真，引擎将记录所有语句以及 repr() 其参数列表的默认⽇志处理程序
2. future   -- 使⽤2.0样式 Engine 和 Connection API。
3. logging_name   -- 将在“sqlalChemy.engine”记录器中⽣成的⽇志记录的“name”字段中使⽤的

字符串标识符。

4. pool_size=5   # 连接池的⼤⼩默认为 5 个，设置为 0 时表⽰连接⽆限制
5. pool_recycle=3600,   # 设置时间以限制数据库⾃动断开
6. pool_timeout：  连接超时时间，默认为30秒，超过时间的连接都会连接失败。

| 参数名                 | 类型 | 默认值 | 说明                                                         |
| ---------------------- | ---- | ------ | ------------------------------------------------------------ |
| **url**                | str  | 必填   | 数据库URL，如：`postgresql+psycopg2://user:pass@host:5432/db` |
| **echo**               | bool | False  | 是否输出所有SQL语句（调试时用）                              |
| **pool\_size**         | int  | 5      | 连接池中保持的连接数                                         |
| **max\_overflow**      | int  | 10     | 超出 `pool_size` 后最多再创建多少连接                        |
| **pool\_recycle**      | int  | -1     | 多少秒后强制回收连接（MySQL 8小时超时用 `3600`）             |
| **pool\_pre\_ping**    | bool | False  | 每次取连接前先发 `SELECT 1` 检测是否存活（推荐True）         |
| **pool\_timeout**      | int  | 30     | 获取连接的超时时间（秒）                                     |
| **future**             | bool | True   | 启用 SQLAlchemy 2.0 风格（推荐True）                         |
| **isolation\_level**   | str  | None   | 设置事务隔离级别，如 `READ COMMITTED`                        |
| **connect\_args**      | dict | None   | 额外传给底层DBAPI的参数，如 SSL、charset 等                  |
| **execution\_options** | dict | None   | 全局执行选项，如 `stream_results=True`                       |

包含数据库的依赖：`pip install sqlalchemy pymysql`

#### 定义模型类

这种模型类结构称为声明性映射，它同时定义了 Python 对象模型，以及描述的数据库元数据 在特定数据库中存在或将要存在的真实 数据库 表。

映射从⼀个基类开始，并且是 通过对类的继承来创建⼀个简单的⼦类。这⾥的⽗类是：Base 模型类。

##### 定义模型

```python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import enum
from datetime import datetime, date
from decimal import Decimal
from typing import Optional

from sqlalchemy import String, DECIMAL, Boolean, func, ForeignKey, DateTime
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

# 定义一个模型类的基类
class Base(DeclarativeBase):
    # 所有的模型类，都有的属性和字段映射
    create_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), comment='记录的创建时间')
    update_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), onupdate=func.now(), comment='记录的最后一次修改时间')

class SexValue(enum.Enum):
    """通过枚举，可以给一些属性（字段）设置预设值"""
    MALE = '男'
    FEMALE = '女'

class Employee(Base):
    """员工的模型类"""

__tablename__ = 't_emp'
    __table_args__ = {"comment": "员工表"}

# id属性， 字段。主键，⾃增
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(40), name='emp_name', unique=True, nullable=False)

# DECIMAL 其中10代表总位数，2代表小数点后的位数
    sal: Mapped[Decimal] = mapped_column(DECIMAL(10, 2), nullable=True, comment='员工的基本工资')
    bonus: Mapped[int] = mapped_column(default=0, comment='员工的津贴和奖金')
    is_leave: Mapped[bool] = mapped_column(Boolean, default=False, comment='员工是否离职，True代表已经离职，False代表在职')

gender: Mapped[SexValue]
    entry_date: Mapped[date] = mapped_column(insert_default=func.now(), nullable=False, comment='入职时间')

# 和部门表关联的外键
    dept_id: Mapped[Optional[int]] = mapped_column(ForeignKey('t_dept.id'), nullable=True)
    # 定义一个关联属性： 该员工所属的部门
    dept: Mapped[Optional['Dept']] = relationship(back_populates='emp_list', cascade='save-update')

# 定义一个和身份证关联的属性：idc
    idc: Mapped[Optional['IdCard']] = relationship(back_populates='emp')
    
    
    # 需要在Employee模型类中增加一个__str__函数
    def __str__(self):
        return f'{self.name}, {self.gender.value}, {self.sal}, {self.entry_date}, {self.bonus}'
```

| 关键字               | 类型/取值               | 说明            | 示例                                        |
| -------------------- | ----------------------- | --------------- | ------------------------------------------- |
| **primary\_key**     | bool                    | 设为主键        | `mapped_column(primary_key=True)`           |
| **nullable**         | bool                    | 是否允许 NULL   | `mapped_column(nullable=False)`             |
| **default**          | 任意值 / Python 函数    | Python 端默认值 | `mapped_column(default=0)`                  |
| **default\_factory** | Callable                | 动态默认值      | `mapped_column(default_factory=uuid4)`      |
| **server\_default**  | str / `text()` / `func` | 数据库端默认值  | `mapped_column(server_default=func.now())`  |
| **server\_onupdate** | str / `func`            | 更新时触发      | `mapped_column(server_onupdate=func.now())` |
| **unique**           | bool                    | 唯一约束        | `mapped_column(unique=True)`                |
| **index**            | bool                    | 创建索引        | `mapped_column(index=True)`                 |
| **autoincrement**    | bool / str              | 自增（仅整数）  | `mapped_column(autoincrement=True)`         |
| **comment**          | str                     | 数据库注释      | `mapped_column(comment="创建时间")`         |

##### engine负责数据库迁移

```python
# 所有的表都重新创建
Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)

# 单独把某个表创建⼀下
Employee.__table__.drop(engine)
Employee.__table__.create(engine)
```

#### alembic数据库迁移⼯具

Alembic 使⽤ SQLAlchemy 作为底层引擎，为关系数据库提供*变更管理*脚本的创建、管理和调⽤。

安装alembic: `pip install alembic`

初始化alembic环境:   `alembic init alembic`,此时在根目录下生成**alembic**目录和**alembic.ini**：

```python
alembic
    README   
    env.py   # 环境配置
    script.py.mako
    versions  # 版本
alembic.ini  #配置文件
```

修改配置`alembic.ini`和环境`env.py `

```python
# alembic.ini
sqlalchemy.url = mysql+pymysql://root:123123@localhost/test_db2?charset=utf8mb4

# env.py
form db.emp.models import Employee
target_metadata = [Employee.metadata]
```

##### 执⾏命令

```python
# ⾃动⽣成迁移脚本
alembic revision --autogenerate -m "init commit" # 注意修改了orm之后，修改-m后迁移脚

# 数据库迁移命令
alembic upgrade head
```

- alembic upgrade head ：将数据库升级到最新版本。
- alembic downgrade base ：将数据库降级到最初版本。
- `alembic upgrade <version>` ：将数据库升级到指定版本。
- `alembic downgrade <version> `：将数据库降级到指定版本。

### Session的相关操作

#### Session对象

session⽤于创建程序和数据库之间的会话，所有对象的载⼊和保存都需通过session对象 。在Web项⽬中，⼀个请求共⽤⼀个session对象。

```python
# 第⼀种，需要⾃⼰提交事务
with Session(bind=engine) as session:
    session.begin()
    try:
        session.add(some_object)
        session.add(some_other_object)
    except:
        session.rollback()
        raise
    else:
        session.commit()

# 第⼆种， 不需要⾃⼰提交事务
with sessionmaker(bind=engine).begin() as session:
    sess.execute()
```

#### 新增模型对象操作

对模型对象进⾏新增，有两种⽅式：

```python

insert_stmt = insert(User).values(name='name1')
with Session() as sess:
    sess.execute(insert_stmt)
    sess.commit()
    

insert_stmt2 = insert(User)
with Session() as sess:
    sess.execute(insert_stmt2,{'name':'name1'})
    sess.commit()
    

with Session() as sess:
    sess.execute(insert_stmt2,[{'name':'name1'},{'name':'name2'}])
    sess.commit()

obj=User(name='name2')
with Session() as sess:
    sess.add(obj)
    sess.commit()
    

obj=User(name='name2')
obj2=User(name='name2')
with Session() as sess:
    sess.add(obj)
    sess.add(obj2)
    # 或者 s.add_all([obj,obj2])
    sess.commit()
    
# 批量添加对象
with sessionmaker(engine).begin() as session:
    emp1 = Employee(name='zs', sal=2000, bonus=500, gender=SexValue.MALE)
    emp2 = Employee(name='ls', sal=3000, bonus=400, gender=SexValue.MALE)
    session.add_all((emp1, emp2))
```

#### 简单查询操作

根据主键查询:  `emp = session.get(Emp, 1)`

查询整张表的数据:

- 返回模型对象

```python
  statement = select(Employee)
  list_emp = session.scalars(statement).all()
  for o in list_emp:
      print(o)
  # 需要在Employee模型类中增加⼀个__str__函数
  def __str__(self):
      return f'{self.name}, {self.gender.value}, {self.sal}, {self.entry_date}'
  ```

- 返回row对象，⼀般⽤于指定返回的字段

```python
  statement = select(Employee.name, Employee.sal, Employee.gender)
  # list_emp中的元素不是模型对象， ⽽是包含多个字段只的row对象
  list_emp = session.execute(statement).all()
  for row in list_emp:
      print(row.name, row.sal, row.gender.value)
  ```

- 执⾏原⽣的SQL，并返回row对象

```python
  # 执⾏原⽣sql
  sql_text = text('select id, name, sal, gender from t_emp')
  # list_emp中的元素不是模型对象， ⽽是包含多个字段只的row对象
  list_emp = session.execute(sql_text).all()
  for row in list_emp:
      print(row)
      print(row.name, row.sal, row.gender)
  ```

- 执⾏原⽣的sql，并返回模型对象

```python
  # 执⾏原⽣sql
  sql_text = text('select id, name, sal, gender from t_emp')
  # ⼿动建⽴映射关系
  new_sql = sql_text.columns(Employee.id, Employee.name, Employee.sal,
  Employee.gender)
  # 得到orm的查询语句
  orm_sql = select(Employee).from_statement(new_sql)
  # list_emp中的元素是模型对象
  list_emp = session.execute(orm_sql).scalars()
  for o in list_emp:
      print(type(o))
      print(o)
  ```

#### 修改操作

先查询出来，再修改属性

```python
# 查询，再修改
old_emp = session.get(Emp, 1)
print(old_emp.name)
old_emp.name = '李四'
```

直接根据主键修改

```python
# 注意where语句就是条件的意思，其中id前⾯的Employee必须有
session.execute(update(Employee).where(Employee.id ==1).values(sal=Decimal(6000), bonus=Decimal(500)))
```

直接根据主键批量修改

```python
# 批量修改，必须根据主键来修改
session.execute(update(Employee), [
    {'id': 1, 'name': '张三三'},
    {'id': 2, 'name': '李四四'},
])
```

#### 删除操作

先查询，再删除

```python
# 先查询出来，再删除
emp = session.get(Employee, 2)
session.delete(emp)
```

直接删除

```python
# 删除的数据，由where条件决定
session.execute(delete(Employee).where(Employee.id == 6))
```

#### 查询条件

过滤是数据提取的⼀个很重要的功能，以下对⼀些常⽤的过滤条件进⾏解释，并且这些过滤条件都是只能通过where⽅法实现的：

1. equals ： ==，或者.is_ 函数

2. not equals : != 或者 isnot函数

3. like & ilike [不区分⼤⼩写]：

4. 在某个集合中存在，in_函数， 或者notin_函数（不存在）

5. And 多条件组合

```python
   # 查询员⼯的名字中包含”四“ ，并且基本薪资⼤于3000的所有员⼯
   where(Employee.name.like('%四%'), Employee.sal > 3000))
   where(and_(Employee.name.like('%四%'), Employee.sal > 3000))
   ```

6. or多条件组合

```python
   where(or_(Employee.name.like('%四%'), Employee.sal > 3000))
   ```

7. 聚合函数

- func.count：统计⾏的数量。
   - func.avg：求平均值。
   - func.max：求最⼤值。
   - func.min：求最⼩值。
   - func.sum：求和。

```python
   result = session.execute(select(func.count(Employee.id)))
   ```

8. 分⻚查询

a. limit函数：可以限制查询的时候只查询前⼏条数据。 属top-N查询

b. offset：可以限制查找数据的时候过滤掉前⾯多少条。可指定开始查询时的偏移量。

```python
   result = session.scalars(select(Employee).offset(0).limit(2))
   ```

9. 排序： order_by函数

```python
   result =session.scalars(select(Employee).order_by(Employee.sal.desc()).offset(0).limit(2))
   ```

10. 分组查询：group_by 和过滤函数 having

```python
    # 查询每个性别，各有多少员⼯
    result = session.execute(select(Employee.gender,
    func.count(Employee.id)).group_by(Employee.gender))
    for o in result:
        # print(type(o))
        print(o)
    ```

### 模型类之间的关联关系

#### ⼀对多和多对⼀关联

⽐如：作者和⽂章之间， 部⻔和员⼯之间都是⼀对多的关联关系。反过来就是：多对⼀的关联关系。

##### 定义外键约束

定义关系的第⼀步是创建外键。外键是（foreign key）⽤来在 A 表存储 B 表的主键值以便和 B 表建⽴联系的关系字段。

因为外键只能存储单⼀数据（标量），所以外键总是在 “多” 这⼀侧定义，多篇⽂章属于同 ⼀个作者，所以我们需要为每篇⽂章添加外键存储作者的主键值以指向 对应的作者。在 Article 模型中，我们定义⼀个 author_id 字段作为外键：

**注意:ForeginKey的参数是<表名>.<键名>，⽽不是<类名>.<字段名**

```python
# 多对⼀关联的外键
dept_id: Mapped[Optional[int]] = mapped_column(ForeignKey('t_dept.id'))
```

##### 定义关联属性和双向关联

我们在 Author 类中定义了集合关系属性 articles ，⽤来获取某个作者拥有的多篇⽂章记录。在某些情况下，你也许希望能在 Article 类中定义 ⼀个类似的 author 关系属性，当被调⽤时返回对应的作者记录，⽽这种两侧都添加关系属性获取对⽅记录的关系我们称之为 双向关系。双向关系并不是必须的，但在某些情况下会⾮常⽅便。

```python
# 多对⼀关联的属性， back_populates写对⽅模型类中的关联属性名字
dept: Mapped[Optional['Dept']] = relationship(back_populates='emp_list')

# ⼀对多的关联属性
emp_list: Mapped[List['Employee']] = relationship(back_populates='dept',cascade='save-update')
```

##### 级联操作

**cascade，默认选项为save-update**：

- ⼀：save-update：默认选项，在添加⼀条数据的时候，会把其他和次数据关联的数据都添加到数据库中，这种⾏为就是save-update属性决定的
- ⼆：delete：表⽰当删除某⼀个模型中的数据的时候，也删除掉使⽤relationship和此数据关联的数据
- 三：delete-orphan：表⽰当对⼀个ORM对象解除了⽗表中的关联对象的时候，⾃⼰便会被删除，如果⽗表的数据被删除，同样⾃⼰也会被删除，这个选项只能⽤在⼀对多上，不能⽤在多对多和多对⼀上，并且使⽤的时候还需要在⼦模型的relationship中增加参数：single_parent=True
- 四：merge(合并)：默认选项，当在使⽤session.merge合并⼀个对象的时候，会将使⽤了relationship相关联的对象也进⾏merge操作
- 五：expunge：移除操作的时候，会将相关联的对象也进⾏移除，这个操作只是从session中移除，并不会正则从数据库删除
- 六：all：对 save-update、merge、refresh-expire、expunge、delete 这⼏种的缩写

##### 树形结构的⾃关联

```py
class Dept(Base):
    """部⻔的模型类"""
    __tablename__ = 't_dept'
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(20), unique=True, nullable=False)
    address: Mapped[Optional[str]] = mapped_column(String(50))
    # ⼀对多的关联属性
    emp_list: Mapped[List['Employee']] = relationship(back_populates='dept', cascade='save-update')
    # ⾃⼰和⾃⼰外键
    pid: Mapped[Optional[int]] = mapped_column(ForeignKey('t_dept.id'))
    # ⾃⼰和⾃⼰⼀对多的关联的属性
    children: Mapped[List['Dept']] = relationship(back_populates='parent')
    # ⾃⼰和⾃⼰多对⼀的关联的属性 remote_side = [id]， 写到多的⼀端。（⾮列表）
    parent: Mapped[Optional['Dept']] = relationship(back_populates='children', remote_side=[id])
```

**维护⼀对多⾃关联关系的时候（relationship），必须加⼊：remote_side= [id]**

#### ⼀对⼀关联

⼀对⼀关系实际上是通过建⽴双向关系的⼀对多关系的基础上转化⽽来。

⽐如：⼀个⽤⼾对应⼀张⾝份证，⼀张⾝份证属于⼀个⽤⼾。

```python
class IdCard(Base):
    """省份证的模型类， 它和员⼯之间是⼀对⼀的关联关系"""
    __tablename__ = 't_id_card'
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    card_number: Mapped[str] = mapped_column(String(18), unique=True, nullable=False, comment='省份证号码')
    origin: Mapped[Optional[str]] = mapped_column(String(50), comment='籍贯')
    
    # 外键
    emp_id: Mapped[int] = mapped_column(ForeignKey('t_emp.id'))
    # 和员⼯的关联属性
    emp: Mapped['Employee'] = relationship(single_parent=True,back_populates='idc')
```

#### 多对多关联

在 SQLAlchemy 中，要想表⽰多对多关系，除了关系两侧的模型外，我们还需要创建⼀个关联表（middle_table）。关联表不存储数据，只⽤来存储关系两侧模型的外键对应关系。

```python
# 多对多关联，先定义中间表（没有对应的模型类）
middle_table = Table(
    't_user_role',
    Base.metadata,
    Column('user_id', ForeignKey('t_user.id'), primary_key=True), # 联合主键
    Column('role_id', ForeignKey('t_role.id'), primary_key=True),
    )

class User(Base):
    # ⽤⼾和⻆⾊之间是多对多关系， ⼀个⽤⼾可以拥有多个⻆⾊，⼀个⻆⾊可以所属多个⽤⼾
    """⽤⼾的模型类"""
    __tablename__ = 't_user'
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    username: Mapped[str] = mapped_column(String(20), unique=True,
    nullable=False)
    password: Mapped[str] = mapped_column(String(20), nullable=False)
    # ⼀个⽤⼾有多个⻆⾊
    roles: Mapped[Optional[List['Role']]] = relationship(secondary=middle_table, back_populates='users')

class Role(Base):
    """⻆⾊的模型类"""
    __tablename__ = 't_role'
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(20), unique=True, nullable=False,comment='⻆⾊的名字')
    
    # ⼀个⻆⾊被多有⽤⼾所拥有
    users: Mapped[Optional[List['User']]] =relationship(secondary=middle_table, back_populates='roles')
```

#### 关联查询

```python
# 关联查询
# 1、查询2020年⼊职的员⼯姓名以及该员⼯的所在部⻔名称
result = session.execute(select(Employee.name, Dept.name).join(Dept,isouter=True).where(extract('year', Employee.entry_date) == 2018))
# 2、 查询省份号码是：11111111111。 的员⼯的姓名，以及该员⼯所属的部⻔
result = session.execute(select(Employee.name, Dept.name).join(Dept,isouter=True).join(IdCard).where(IdCard.card_number == '111111111111'))
# 3、 查询，每个部⻔名字、city以及该部⻔下⾯的员⼯个数
result = session.execute(select(Dept,func.count(Employee.dept_id)).join(Employee, isouter=True).group_by(Dept.id))
# 4、查询拥有⻆⾊数量超过1的 ⽤⼾名以及他所拥有的⻆⾊数量。
sub_stmt = select(User.username.label('username'),func.count(Role.id).label('role_count')).join(User.roles).group_by(User.id).subquery()
result = session.execute(select(sub_stmt.c.username,sub_stmt.c.role_count).where(sub_stmt.c.role_count > 1))
# 5、 查询每个部⻔的名字以及它⾥⾯的员⼯数量，并且按照员⼯数量的个数降序排序。
result = session.execute(select(Dept,func.count(Employee.dept_id).label('ec')).join(Employee,isouter=True).group_by(Dept.id).order_by(desc('ec')))
```

### FastAPI和SQLAlchemy的整合

#### 定义Pydantic模型

⼀般情况下，每⼀个功能模块都需要⾃⼰的数据模式（schemas）。数据模式（schemas）是FastAPI模块⽤于数据传递的对象，通过继承pydantic中的类建⽴。多个pydantic模型类组成了完整的schemas。

⼀旦我们定义了Pydantic模型，我们可以使⽤ orm_mode 参数将ORM模型转换为Pydantic模型。orm_mode 参数告诉Pydantic将ORM模型视为数据库记录（dict）⽽不是普通的Python对象。

我们可以将Pydantic模型应⽤到FastAPI中的路径参数、请求体和响应模型上。FastAPI会⾃动根据Pydantic模型⽣成⽂档，并进⾏输⼊数据的验证和输出数据的编码。

```python
class EmpBase(BaseModel):
    id: Optional[int] = Field(description='员⼯的ID,可以不传', default=None)
    name: str = Field(description='员⼯的名字')
    gender: SexValue = Field(description='员⼯的性别', default=SexValue.MALE)
    sal: Optional[Decimal] = Field(description='基本薪资', default=0)
    bonus: Optional[int] = Field(description='奖⾦和津贴', default=0)
    is_leave: Optional[bool] = Field(description='是否离职', default=False)
    entry_date: Optional[date] = Field(description='⼊职时间')
    dept_id: Optional[int] = Field(description='员⼯所属的部⻔ID,可以不传',default=None)

class Config:
        from_attributes = True

class DeptBase(BaseModel):
    id: Optional[int] = Field(description='部⻔的ID,可以不传', default=None)
    name: str = Field(description='部⻔的名字')
    city: str = Field(description='部⻔所在的城市', default=None)

class EmpBo(EmpBase):
    dept: Optional[DeptBase] = Field(default=None)
    class Config:
        from_attributes = True
```

#### Config 类

此类 Config ⽤于为 Pydantic 提供配置:`from_attributes = True`.将告诉 Pydantic 模型读取数据，即它不是⼀个 dict，⽽是⼀个 ORM 模型。这样该 Pydantic 模型就

会尝试从属性中获取它，如 id = db_model.id。

#### FastAPI和SQLAlchemy整合

```python
@app.get("/test", response_class=HTMLResponse)
def test(request: Request, name: Union[str, None], session: Session = Depends(get_session)):
    all_list = session.query(Employee).all()
    return templates.TemplateResponse("result.html", {"request": request,'emp_list': all_list})

def get_session():
    session = Session(bind=engine)
    try:
        yield session
    finally:
        session.close()
```
- tortoise项目案例：wzd_data_union_platform

- SQLAlchemy项目案例：wzd_alliance_sync_platform，Session要使用get_db不能是async_db_session

FastAPI学习参考文档

Mybtais核心参考文档资源

TypeScript学习

FastAPI学习参考文档

JDK 源码解析 —— 调试环境搭建（二）进阶

JDK 源码解析 —— 集合（一）数组 ArrayList

Dubbo 源码分析 —— 调试环境搭建