中间件

中间件(middleware)是一个每次请求进入FastAPI时都会被执行的函数，它会在请求到达路径参数之前运行一次，返回响应再运行一次

中间件函数顶部使用装饰器@app.middleware("http")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from fastapi import FastAPI

app = FastAPI()


@app.middleware("http")
async def middleware(request, call_next):
    print("middleware start")
    response = await call_next(request)
    print("middleware end")
    return response


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

发送任意请求时，你将能在控制台看到：

1
2
middleware start
middleware end

其中call_next函数用于将请求传递到下一步，请求传输路径为
请求 → 中间件2 → 中间件1 → 路由函数 → 中间件1返回 → 中间件2返回 → 客户端

显然，中间件具有洋葱结构，第一次请求时反向执行，返回时正向执行，例如：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from fastapi import FastAPI

app = FastAPI()

@app.middleware("http")
async def middleware1(request, call_next):
    print("middleware1 start")
    response = await call_next(request)
    print("middleware1 end")
    return response

@app.middleware("http")
async def middleware2(request, call_next):
    print("middleware2 start")
    response = await call_next(request)
    print("middleware2 end")
    return response

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

发送任意请求，你将在控制台看到：

1
2
3
4
middleware2 start
middleware1 start
middleware1 end
middleware2 end

依赖注入

依赖注入也可用于共享通用逻辑，与中间件不同的是，中间件会在接收到任意请求时生效，而依赖注入可以自行指定生效范围
查询参数可以写在依赖函数里供使用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from fastapi import FastAPI,Depends

app = FastAPI()

async def deps(
        id : int | None = None,
        uid : str | None = None,
):
    print("I am depend")
    return {"message": f"depend is successful with id:{id},uid:{uid}"}

@app.get("/")
async def root(comments = Depends(deps)):
    return comments

ORM建表

ORM(Object Relationship Mapping)是用Python对象操作数据库的框架，能有效减少重复的SQL语句，自动防止SQL注入攻击等

创建数据库引擎

创建数据库异步引擎用sqlalchemy.ext.asyncio.create_async_engine方法

1
2
3
4
5
6
7
8
9
10
from sqlalchemy.ext.asyncio import create_async_engine

async_database_url = "mysql+aiomysql://root:123456@localhost:3306/fastapi?charset=utf8"

async_engine = create_async_engine(
    async_database_url,
    echo = True,  #是否输出日志
    pool_size = 5,  #连接池最大连接数
    max_overflow = 10   #连接池临时扩容上限
)

数据类型

数据库中的数据类型与Python不同，sqlalchemy.orm.Mapped可以将Python类型映射为数据库类型

数据库中的类型从sqlalchemy 或 sqlalchemy.sql.sqltypes直接导入

定义模型类

所有数据库的表都继承基础父类Base，Base继承sqlalchemy.orm.DeclaretiveBase类

Base的一个子类对应数据库的一个表，一个属性对应一个字段

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
from fastapi import FastAPI
from sqlalchemy import func,DateTime, String
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from datetime import datetime

app = FastAPI()

async_database_url = "mysql+aiomysql://root:123456@localhost:3306/fastapi?charset=utf8"

async_engine = create_async_engine(
    async_database_url,
    echo = True,
    pool_size = 5,
    max_overflow = 10
)

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 User(Base):
    __tablename__ = 'user'

    id :Mapped[int] = mapped_column(primary_key=True)
    username : Mapped[str] = mapped_column(String(255),comment="用户名")
    password : Mapped[str] = mapped_column(String(255),comment="密码")

async def create_tables():
    async with async_engine.connect() as conn:
        await conn.run_sync(Base.metadata.create_all)