FastApi的官方全栈项目后端架构你看懂了吗?_fastapi项目结构

先看这张App目录下结构图, 我们一步步说起

api目录

api:  # 存放和接口有关的目录
  api_v1:  # 版本号为v1的接口目录
    endpoints:  # 存放一个个端点(也就是某一类接口)
      login.py   # 登陆业务相关的接口
      user.py  # 和用户有关业务的接口具体实现 类似于View
    api.py  # 用于管理每个端点, 也就是把每个业务模块的逻辑汇总
  deps.py  # 存放接口需要的依赖项, 例如(生成数据库会话, 获取当前用户等)
1
2
3
4
5
6
7

core目录

core:  # 存放一些配置有关的
  config.py  # 存放了一个配置类
  security.py  # 封装了某些和账户安全校验等有关的函数
1
2
3

crud目录, 详细解读

crud:  # 封装了公用的增删改查逻辑, 这样也可以解耦View中的逻辑
  base.py  # 公用增删改查基类的定义和封装
  crud_user.py  # 有关user业务的增删改查的封装
  ...
1
2
3
4

# base.py
from typing import TypeVar

ModelType = TypeVar("ModelType", bound=Base)
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)
1
2
3
4
5
6

首先来解读一下TypeVar, 这是定义泛型的一种方式.我们就用当前代码来讲解
ModelType这个类型, 通过bound指定了顶层类型为Base.
来源于class Base(AsyncAttrs, DeclarativeBase): pass

接下来看CreateSchemaType这个类型顶层是BaseModel, 这个约束很宽泛, 无论有多少层继承, 只要最终顶层的类是BaseModel就足够了.这里的用途请看下方代码.

    Any, Dict,
    Generic, List,
    Optional, Type,
    TypeVar, Union, Tuple, Sequence )

from fastapi.encoders import jsonable_encoder from pydantic import
BaseModel from sqlalchemy import select, Row, insert, RowMapping,
ChunkedIteratorResult from sqlalchemy.ext.asyncio import AsyncSession,
async_sessionmaker

from app.models.base_class import Base

ModelType = TypeVar("ModelType", bound=Base) CreateSchemaType =
TypeVar("CreateSchemaType", bound=BaseModel) UpdateSchemaType =
TypeVar("UpdateSchemaType", bound=BaseModel)

# 这里我们定义了一个泛型类 class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]): 	""" 	这里我们定义了一个泛型类, 允许的类型 -> [
        ModelType, CreateSchemaType, UpdateSchemaType
    ]
    分别是一个用于数据库的sqlalchemy模型类 -> ModelType
    新建用户/更新用户 的pydantic模型类 -> ModelType / UpdateSchemaType
    """
    def __init__(self, model: Type[ModelType]):
        """
        默认方法: create - select - update - delete

        * `model`: 一个`sqlalchemy base` 类
        * `schema`: 一个 `pydantic` 模型 / 类
        """
        self.model = model

    async def get(self, async_session: AsyncSession, _id: Any) -> Optional[ModelType]:
        print(_id)
        result = await async_session.execute(select(self.model).where(self.model.id == _id))
        return result.scalars().first()

    async def get_multi(
            self, async_session: AsyncSession, *, skip: int = 0, limit: int = 100
    ) -> Sequence[Row | RowMapping | Any]:
        result = await async_session.execute(
            select(self.model).offset(skip).limit(limit)
        )
        return result.scalars().all()

    async def create(self, async_session: AsyncSession, *, obj_in: CreateSchemaType) -> ModelType:
        async with async_session.begin():
            obj_in_data = jsonable_encoder(obj_in)
            db_obj = self.model(**obj_in_data)
            async_session.add(db_obj)
            return db_obj

    async def update(
            self,
            async_session: AsyncSession,
            *,
            db_obj: ModelType,
            obj_in: Union[UpdateSchemaType, Dict[str, Any]]
    ) -> ModelType:
        """
        通过接收用户请求更新字段, 查找当前用户后按照用户提供字段做变更
        :param async_session: 会话
        :param db_obj: 当前用户old
        :param obj_in: 用户提供更新字段new
        :return:
        """
        result = await async_session.execute(select(self.model).where(self.model.id ==
db_obj.id))
        if isinstance(obj_in, dict):
            update_data = obj_in
        else:
            update_data = obj_in.model_dump(exclude_unset=True)
            print(update_data)
        user = result.scalars().first()

        for field, val in update_data.items():
            setattr(user, field, val)

        await async_session.commit()
        return user

    async def remove(self, async_session: AsyncSession, *, _id: int) -> ModelType:
        stmt = await async_session.execute(select(self.model).where(self.model.id == _id))
        result = stmt.scalar_one()
        user = await async_session.delete(
            result
        )
        await async_session.flush()
        await async_session.commit()
        return user ```
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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89

我们单拎出来一个方法create来看

    async def create(
    	self, 
    	async_session: AsyncSession, 
    	*, 
    	obj_in: CreateSchemaType
    ) -> ModelType:
    	"""
    	我们在此处定义的obj_in实际上就是一个新建用户所需模型,
    	也就是自定制的pydantic模型(schema).
    	"""
        async with async_session.begin():
            obj_in_data = jsonable_encoder(obj_in)
            db_obj = self.model(**obj_in_data)
            async_session.add(db_obj)
            return db_obj  
            # 注意此处返回的实际上是: <__main__.User object at 0x00000286F439B820> <=> User <=> ModelType <= Base =>
            # 这个对象的User就是我们用sqlalchemy定义的表模型
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

这里只有定义, 那么如何使用呢?

我们上面说到那只是个CRUD的基类, 定义一些偏低级的方法.但是我们可以通过这个基类去拓展, 且看如下代码:

class CRUDUser(CRUDBase[User, UserCreate, UserUpdate]):
	"""
	此处我们定制化了一个User业务的增删改查类和方法
	还记得这三个泛型类型吗, 我们这里的三个泛型刚好是其`实例`
	之前我们的表模型是个ModelType来泛指(并不具体化)而定制化之后就变成User
	ModelType = TypeVar("ModelType", bound=Base)
    CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
    UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)
	"""
    async def update(
            self, async_session: AsyncSession, 
            *, db_obj: User, obj_in: Union[UserUpdate, Dict[str, Any]]
    ) -> User:  # 此处返回依然是一个表的实例对象
        if isinstance(obj_in, dict):
            update_data = obj_in
        else:
            update_data = obj_in.model_dump(exclude_unset=True)
        print("update_data--->", update_data)

        # 检测密码是否与之前一致
        if update_data.get("password", None):
            # 密码需要做hash处理
            hashed_password = get_password_hash(update_data["password"])
            # 并删除原密码
            del update_data["password"]
            # 在更新内容中携带hashed password
            update_data["password"] = hashed_password
            # 以上是我们的定制化需求, 而后我们可以正常去调用我们(父类方法)底层方法
        return await super().update(async_session, db_obj=db_obj, obj_in=update_data)

# 当然在使用这个类时我们必须创建一个实例
user = CRUDUser(User)
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

在视图中如何使用?

我们可以直接导入上述创建的user实例然后user.update(...)

@router.put("/{user_id}", response_model=schemas.User)
async def update_user(
        *,
        db: AsyncSession = Depends(deps.get_async_session),
        user_id: int,
        user_in: schemas.UserUpdate,
        # current_user: models.User = Depends(deps.get_current_active_superuser),
) -> Any:
    """
    更新用户信息
    """
    user = await crud.user.get(db, _id=user_id)
    if not user:
        raise HTTPException(
            status_code=404,
            detail="The user with this username does not exist in the system",
        )
    user = await crud.user.update(db, db_obj=user, obj_in=user_in)
    return user
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

其实官方以上这种做法实现起来也并不是最优解, 我们当然可以采取自己更简便的方式, 但是从中我们可以学到一种模式和切身体会的去用泛型, 而不是纸上谈兵的去记忆那些概念

db目录

db:  # 和数据库配置相关
 session.py  # 一个定义数据库引擎和会话的脚本
 init_db.py  # 一个初始化数据库的脚本
1
2
3

models目录

定义了数据表模型

models:  # 用于定义数据表模型
  base_class.py  # 定义了表模型要继承的基类Base
  user.py  # 定义了继承Base类的表模型
  ...
1
2
3
4

shcemas目录

定义了pydantic模型, 就像下面这样

class UserBase(BaseModel):
    name: Optional[constr(max_length=50, strip_whitespace=True)] = Field(None, >example="Alice")
    email: Optional[EmailStr] = Field(None, max_length=20)
    create_date: datetime.datetime = Field(default_factory=datetime.datetime.now)
    activated: bool = False
1
2
3
4
5

未完待续…