FastAPI

Python类型提示

函数定义处，对于添加类型提示，使得函数内部可以辨别该变量的类型，进而为自动补全提供了便利：

1
2
3
4
5
6
7
8
    def get_full_name(first_name: str, last_name: str):
        full_name = first_name.title() # 此处打点后就会有相应多选的弹窗
```  
列表(list)、字典(dict)等类型的标注需要调用`typing`库：
```python
from typing import List, Dict, Set

x: List[List[int]] = [[1,2], [1,5]]

此外，还可以使用Optional和Any来实现可选和任意传入的方式：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from typing import Optional,Any, Sequence, Tuple, Callable, TypeVar
# Sequence限定一定是一个序列，列表或者元组
# Callable用于限定返回的是一个符合参数和返回值限制的函数
# TypeVar是一种泛型，限定了一种特定类型，:
T = TypeVar('T')
def get_item(lst: List[T], index: int) -> T:
    return lst[index]

def foo(output : [Optional]=False):
    pass

def foo2(output: Any):
    pass

# 由于tuple的特殊性，如果直接使用tuple来限制最终只会限制是否为一个元组，若要限制元组中每个位置上元素的类型则要用Tuple来一一限制：
x: tuple = (1,2,3,"HELLO")
y: Tuple[int, int, int] = (1, 1, 3)

虚拟环境

在开始一个Python工程时，创建一个虚拟环境:

1
2
3
4
5
6
# -m表示以脚本的方式调用一个模块，此后将告知使用哪个模块，venv即为相应模块，将在.venv目录下创建虚拟环境
python -m venv .venv
```   
通过如下方式激活虚拟环境：
```shell
.venv\Scripts\Activate.ps1

可用如下方法在PowerShell中检查是否激活成功了：

1
Get-Command python

并发 async/await

  1. 若正使用第三方库，可以使用await来调用:result = await some_lib() ，通过async def声明(只能在此内使用await)操作函数：@app.get('/') async def read_results(): results = await some_library() return results。
  2. 若进行通信的第三方库不支持使用await，则直接使用def定义即可。
  3. 若应用程序不用和其他任何东西通信来等待响应，则使用async def。
  4. 不清楚，则使用def即可。
    遵循上述可进行性能优化。

@decorator Info

装饰器，放置在函数的上一行，用来修饰其下的函数。

查询参数

直接添加到函数定义处的参数列表中：

1
2
3
4
5
6
7
8
9
10
from fastapi import FastAPI

app = FastAPI()

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


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

可以通过如下方式来进行相关访问：http://127.0.0.1:8000/items/?skip=0&limit=10.
此外还可以通过不添加默认值的方式来设定某些查询参数的必须属性，即非填不可。

HTTP报文

一般包含请求信息(请求信息，类型method、URL链接、…)、请求头headers、请求体body(表单数据、JSON数据、文件上传等)

HTTP与Web标准

从文件分离，到路由分类集中

    app = FastAPI()

    router = APIRouter()
    @router.get("/items/{item_id}")
async def read_item(
        item_id: int, needy: str, skip: int = 0, limit: Union[int, None] = None
):
    pass

@router.get("/items/{items_id}")
async def read_item(items_id: int, needy: str, skip: int = 0, limit: Union[int, None] = None):
    pass

app.include_router(router, prefix="/api")

中间件

在路由动作前(后)统一加一个操作，加在路由前的操作为中间件。

数据存储：从内存到数据库

使用数据库存数据

SQL(关系型)数据库

from sqlmodel import Field, Session, SQLModel, create_engine, select

### 创建模型
class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: int | None = None



### 创建引擎
'''
    使用 check_same_thread=False 可以让 FastAPI 在不同线程中使用同一个 SQLite 数据库。这很有必要，因为单个请求可能会使用多个线程（例如在依赖项中）。
    不用担心，我们会按照代码结构确保每个请求使用一个单独的 SQLModel 会话，这实际上就是 check_same_thread 想要实现的。
'''
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, connect_args=connect_args)

### 创建表
def create_db_and_tables():
    # 为所有表模型创建表
    SQLModel.metadata.create_all(engine)

# 创建会话依赖项，
'''
    Session 会存储内存中的对象并跟踪数据中所需更改的内容，然后它使用 engine 与数据库进行通信。

    我们会使用 yield 创建一个 FastAPI 依赖项，为每个请求提供一个新的 Session 。这确保我们每个请求使用一个单独的会话。

    然后我们创建一个 Annotated 的依赖项 SessionDep 来简化其他也会用到此依赖的代码。
'''
def get_session():
    with Session(engine) as session:
        yield session

SessionDep = Annotated[Session, Depends(get_session)]

### 启动时创建数据库表
@app.on_event("startup")
def on_startup():
    create_db_and_tables()

### 创建Hero类
'''
    每个SQLModel模型也是一个Pydantic模型，您可在与其相同的类型标注中使用，如：
    若声明一个类型为`hero`的参数，其将从JSON主体中读取数据，同样也可以声明为函数的返回类型
'''
@app.post("/heroes/")
def create_hero(hero: Hero, session: SessionDep) -> Hero:
    session.add(hero)
    session.commit()
    session.refresh(hero)
    return hero
with Session(engine) as session:
    # 可以使用如下方式获取数据库中相关数据，但是请注意对于MySQL来说，`session.exec(statement).first()`获取的是一个元组，因此此处仍需要通过[]来再取相应索引的元素
    statement = select(Hero).where(Hero.name == "Spider-Boy")
    hero = session.exec(statement).first()[0]
    print(hero)