FastApi中使用ORM

1、介绍

在应用的开发过程中肯定会使用到数据库, FastApi中是一个异步的web框架配合异步的ORM Tortoise能让FastAPI的并发性能，而且TortoiseORM是受Django ORM框架启发的，从Django ORM 移动TortoiseORM 就很平滑。

# 安装
pip install fastapi
pip install tortoise-orm

2、项目代码地址

链接：https://pan.baidu.com/s/1HFQ8ZB64Du9Dg8a66ftGGw?pwd=wfo7 
提取码：wfo7

3、数据库结构

• 用户和组是多对多的关系，一个用户可以添加多个组，一个组可以被多个用户添加。

• 用户组和权限是多对多的关系，一个用户组可以有多个权限，一个权限可以被多个用户库添加，用户不直接享有权限通过加入不同的用户组拥有不同的权限。

• 权限表表示不同应用的权限，appname+action是唯一的，appname和应用对应，每个应用中有增删改的权限。

• 操作记录表记录用户的操作。

# models.py
from enum import Enum
from tortoise.models import Model
from tortoise import fields
# pip install tortoise-orm
​
​
class DateTimeModel(Model):
    created_at = fields.DatetimeField(auto_now_add=True, description="创建时间")
    updated_at = fields.DatetimeField(auto_now=True, description="更新时间")
​
    class Meta:
        abstract = True  # 抽象模型，将不会在数据库中创建表
​
​
class User(DateTimeModel):
    """用户表"""
    id = fields.IntField(pk=True, description="用户id")
    username = fields.CharField(max_length=255, description="姓名")
    account = fields.CharField(max_length=255, description="账号")
    pwd = fields.CharField(max_length=255, description="密码")
    is_delete = fields.BooleanField(default=False, description="是否删除")
    last_login = fields.DatetimeField(description="最后登录时间", auto_now_add=True)
    is_superuser = fields.BooleanField(default=False, description="是否是超级管理员")
    is_active = fields.BooleanField(default=False, description="是否活动状态")
    userGroup = fields.ManyToManyField("models.Group", related_name="user", description="用户组")
​
    class Meta:
        table = "user"  # 数据库中的表名称
        table_description = '用户表'
​
​
class Group(Model):
    """用户组表"""
    id = fields.IntField(pk=True, description="用户组id")
    username = fields.CharField(max_length=255, description="组名")
    groupPermission = fields.ManyToManyField("models.Permission", related_name="group", description="用户组权限")
​
    class Meta:
        table = "group"  # 数据库中的表名称
        table_description = '用户组表'
​
​
class Operate(Enum):
    default = "default"
    add = 'add'
    delete = 'delete'
    update = 'update'
    query = "query"
​
​
class Permission(Model):
    """权限表"""
    id = fields.IntField(pk=True, description="权限id")
    appname = fields.CharField(max_length=255, description="应用名称")
    name = fields.CharField(max_length=255, description="权限名")
    action = fields.CharEnumField(Operate, description="权限", default=Operate.default)
​
    class Meta:
        table = "permission"  # 数据库中的表名称
        table_description = '权限表'
​
​
class OperationRecords(Model):
    """操作记录表"""
    id = fields.IntField(pk=True, description="权限id")
    operate = fields.CharEnumField(Operate, description="操作", default=Operate.query)
    user = fields.ForeignKeyField("models.User", related_name="operation")
    operate_time = fields.DatetimeField(auto_now_add=True, description="操作时间")
​
    class Meta:
        table = "operationRecord"  # 数据库中的表名称
        table_description = '操作记录表'

​

4、配置数据库

# settings.py
TORTOISE_ORM = {
    'connections': {
        'default': {
            # 'engine': 'tortoise.backends.asyncpg',  PostgreSQL
            'engine': 'tortoise.backends.mysql',  # MySQL or Mariadb
            'credentials': {
                'host': '192.168.1.200',
                'port': '3306',
                'user': 'root',
                'password': '123456',
                'database': 'test',
                'minsize': 1,
                'maxsize': 5,
                'charset': 'utf8mb4',
                "echo": True
            }
        },
    },
    'apps': {
        'models': {
            'models': ['apps.models', "aerich.models"],
            'default_connection': 'default',
​
        }
    },
    'use_tz': False,
    'timezone': 'Asia/Shanghai'
}

​

5、迁移数据库

aerich是一种ORM迁移工具，需要结合tortoise异步orm框架使用。安装`aerich

# 1、安装aerich工具
pip install aerich
# 2、初始化配置，初始化一次即可,会生成pyproject.toml文件
aerich init -t settings.TORTOISE_ORM
# 3、初始化数据库生成数据表和migrations文件
aerich init-db
# 4、修改models的数据之后重新生成迁移文件
aerich migrate
# 5、将迁移文件写入数据库中
aerich upgrade
# 6、回退到上一个版本
aerich downgrade
# 7、查看迁移记录
aerich history

5.4 添加逻辑代码

1、添加用户逻辑

"""
app01 为FastApi ORM 操作 User表
"""
​
from fastapi import APIRouter
from typing import List
from pydantic import BaseModel, Field, validator
from models import User, Group
from exception import MyException
app10 = APIRouter()
​
​
class UserIn(BaseModel):
    username: str
    account: str
    pwd: str
    is_superuser: bool
    userGroup: List[int] = Field(default=[1], description="未设置用户组时添加到默认组中")
​
    @validator("username")
    def check_username(cls, username):
        assert 2 <= len(username) <= 12, "用户名称的长度为2~12"
        return username
​
    @validator("account")
    def check_account(cls, account):
        assert 6 <= len(account) <= 12, "账号的长度为6~12"
        return account
​
    @validator("account")
    def check_pwd(cls, pwd):
        assert 6 <= len(pwd) <= 12, "密码的长度为6~12"
        return pwd
​
​
class UserOut(BaseModel):
    user_id: int
    username: str
    account: str
    pwd: str
    is_superuser: bool
    group_ids: List[int]
    permission_ids: List[int]
​
​
async def get_user_info(user: User) -> dict:
    group_ids = await user.userGroup.all().values_list("id", flat=True)
    permission_ids = await Group.filter(id__in=group_ids).values_list("groupPermission__id", flat=True)
    return {
        "user_id": user.id,
        "username": user.username,
        "account": user.account,
        "pwd": user.pwd,
        "is_superuser": user.is_superuser,
        "group_ids": group_ids,
        "permission_ids": permission_ids
    }
​
​
@app10.get("/", summary="获取所有用户", response_model=List[UserOut])
async def get_users():
    users = await User.filter(is_delete=False)
    user_infos = []
    for user in users:
        user_info = await get_user_info(user)
        user_infos.append(user_info)
    return user_infos
​
​
@app10.get('/{user_id}', summary="获取特定的用户", response_model=UserOut)
async def get_user(user_id: int):
    users = await User.filter(id=user_id, is_delete=False)
    if not users:
        raise MyException(f"不存在用户:{user_id}")
    user_info = await get_user_info(users[0])
    return user_info
​
​
@app10.post('/', summary="添加用户", response_model=UserOut)
async def add_user(user_in: UserIn):
    if await User.filter(account=user_in.account).exists():
        raise MyException("用户已经存在")
    user = await User.create(username=user_in.username, account=user_in.account, pwd=user_in.pwd,
                             is_superuser=user_in.is_superuser)
    groups = await Group.filter(id__in=user_in.userGroup)
    await user.userGroup.add(*groups)
    user_info = await get_user_info(user)
    return user_info
​
​
@app10.put("/{user_id}", summary="修改用户", response_model=UserOut)
async def update_user(user_id: int, user_in: UserIn):
    data = user_in.dict()
    user_group = data.pop("userGroup")
    await User.filter(id=user_id).update(**data)
    user = await User.get(id=user_id)
    groups = await Group.filter(id__in=user_group)
    await user.userGroup.clear()
    await user.userGroup.add(*groups)
    user_info = await get_user_info(user)
    return user_info
​
​
@app10.delete("/{user_id}", summary="删除用户")
async def delete_user(user_id: int):
    await User.filter(id=user_id).update(is_delete=True)
    return {"message": "ok"}

操作用户接口包含:

• GET /user: 查询所有的用户信息

• GET /user/1: 查询id=1的用户信息

• POST /user/添加用户，添加用户时可以绑定用户组也可以使用默认的用户组

• PUT /user/1 更新用户id为1的用户信息

• DETELE /user/1删除用户id为1的用户信息, 逻辑删除，将is_delete为True

注意: 外键字段不能为空是数据库强制的验证，而且ORM中连表查询外键为空也可能出现问题，为了数据命中索引时也尽量不设置为空。但是新添加的用户时，可能不会立即加入到某个组，所以设定为未分配组时将添加到默认(default)组中，后续的权限表也是如此先建好一个默认的用户组和默认的权限表。

2、添加用户组逻辑

"""
app01 为FastApi ORM 操作 Group
"""
​
from fastapi import APIRouter, Depends
from fastapi.responses import JSONResponse
from typing import List, Union
from pydantic import BaseModel, Field, validator
from models import User, Group, Permission
from exception import MyException
​
app11 = APIRouter()
​
​
class GroupIn(BaseModel):
    username: str
    groupPermission: List[int] = [1]
​
    @validator("username")
    def check_username(cls, username):
        assert 2 <= len(username) <= 12, "组名名称的长度为2~12"
        assert username != "default", "default组不能增加删除修改"
        return username
​
​
class GroupOut(BaseModel):
    group_id: int
    username: str
    permission_ids: List[int]
​
​
async def get_group_info(group: Group) -> dict:
    permission_ids = await group.groupPermission.all().values_list("id", flat=True)
    return {
        "group_id": group.id,
        "username": group.username,
        "permission_ids": permission_ids
    }
​
​
async def check_default_group(group_id: int):
    assert group_id !=1, "默认组不能修改和删除"
    return group_id
​
​
@app11.get("/", summary="获取所有组", response_model=List[GroupOut])
async def get_groups():
    groups = await Group.all()
    group_infos = []
    for group in groups:
        group_info = await get_group_info(group)
        group_infos.append(group_info)
    return group_infos
​
​
@app11.get('/{group_id}', summary="获取特定的组", response_model=GroupOut)
async def get_group(group_id: int):
    groups = await Group.filter(id=group_id)
    if not groups:
        raise MyException(f"不存在组:{group_id}")
    group_info = await get_group_info(groups[0])
    return group_info
​
​
@app11.post('/', summary="添加组", response_model=GroupOut)
async def add_group(group_in: GroupIn):
    if await Group.filter(username=group_in.username).exists():
        raise MyException(f"{group_in.username}组已经存在")
    group = await Group.create(username=group_in.username)
    permissions = await Permission.filter(id__in=group_in.groupPermission)
    await group.groupPermission.add(*permissions)
    group_info = await get_group_info(group)
    return group_info
​
​
@app11.put("/{group_id}", summary="修改组", response_model=GroupOut)
async def update_group(group_in: GroupIn, group_id: int = Depends(check_default_group)):
    await Group.filter(id=group_id).update(username=group_in.username)
    group = await Group.get(id=group_id)
    permissions = await Permission.filter(id__in=group_in.groupPermission)
    await group.groupPermission.clear()
    await group.groupPermission.add(*permissions)
    group_info = await get_group_info(group)
    return group_info
​
​
@app11.delete("/{group_id}", summary="删除组")
async def delete_group(group_id: int = Depends(check_default_group)):
    await Group.filter(id=group_id).delete()
    return {"message": "ok"}
​

用户组中的接口为：

• GET /group: 查询所有的用户组信息

• GET /group/1: 查询用户组1的信息

• POST /group: 添加用户组，添加用户组可以立即绑定权限，也可以后续更改用户组信息时绑定权限

• PUT /group/1: 修改用户组信息。

• DELETE /group/1: 删除用户组， 一般不会删除，所以直接数据库上删除

注意: 填加用户组信息也一样，如果添加时没有绑定权限表则绑定没有任何权限的表。

3、添加权限逻辑

"""
app01 为FastApi ORM 操作 Permission
"""
​
from fastapi import APIRouter, Depends
from fastapi.responses import JSONResponse
from typing import List, Union
from pydantic import BaseModel, Field, validator
from models import User, Permission, Operate
from exception import MyException
​
​
app12 = APIRouter()
​
​
class PermissionIn(BaseModel):
​
    appname: str
    name: str
    action: Operate = Operate.query
​
    @validator("appname")
    def check_appname(cls, appname):
        assert 2 <= len(appname) <= 12, "应用名称的长度为2~12"
        assert appname != "0", "应用名称为0的权限不能增加删除修改"
        return appname
​
    def check_name(cls, name):
        assert 2 <= len(name) <= 12, "权限名称的长度为2~12"
        return name
​
​
class PermissionOut(BaseModel):
    permission_id: int
    appname: str
    name: str
    action: Operate
​
​
async def get_permission_info(permission: Permission) -> dict:
    return {
        "permission_id": permission.id,
        "appname": permission.appname,
        "name": permission.name,
        "action": permission.action
    }
​
​
async def check_default_permission(permission_id: int):
    assert permission_id != 1, "默认权限不能修改和删除"
    return permission_id
​
​
@app12.get("/", summary="获取权限", response_model=List[PermissionOut])
async def get_permissions():
    permissions = await Permission.all()
    permission_infos = []
    for permission in permissions:
        permission_info = await get_permission_info(permission)
        permission_infos.append(permission_info)
    return permission_infos
​
​
@app12.get('/{permission_id}', summary="获取特定的组", response_model=PermissionOut)
async def get_permission(permission_id: int):
    permissions = await Permission.filter(id=permission_id)
    if not permissions:
        raise MyException(f"不存在组:{permission_id}")
    permission_info = await get_permission_info(permissions[0])
    return permission_info
​
​
@app12.post('/', summary="添加权限", response_model=PermissionOut)
async def add_permission(permission_in: PermissionIn):
    if await Permission.filter(appname=permission_in.appname, action=permission_in.action).exists():
        raise MyException(f"{permission_in.appname}中已经存在权限为{permission_in.action}")
    permission = await Permission.create(**permission_in.dict())
    permission_info = await get_permission_info(permission)
    return permission_info
​
​
@app12.put("/{permission_id}", summary="修改权限", response_model=PermissionOut)
async def update_permission(permission_in: PermissionIn, permission_id: int = Depends(check_default_permission)):
    await Permission.filter(id=permission_id).update(**permission_in.dict())
    permission = await Permission.get(id=permission_id)
    permission_info = await get_permission_info(permission)
    return permission_info
​
​
@app12.delete("/{permission_id}", summary="删除组")
async def delete_permission(permission_id: int = Depends(check_default_permission)):
    await Permission.filter(id=permission_id).delete()
    return {"message": "ok"}

​

权限的接口为：

• GET /permission: 查询所有的权限信息

• GET /permission/1: 查询id1的权限信息

• POST /permission: 添加权限表

• PUT /permission/1: 修改用权限信息。

• DELETE /permission/1: 删除id为1权限

注意: 权限控制只精细到应用的操作，不同的应用都有增删改查的权限，用户组中有不同应用的权限，用户通过加入不同的用户组中获取这个权限， 权限表中appname+action是唯一的。

4、操作记录

async def recode_user_operate(request: Request, call_next):
    # 1.从cookies中获取用户ID,未登录不记录
    # user_id = request.cookies.get("user_id")
    user_id = 1
    response = await call_next(request)
    if user_id and await User.filter(id=user_id, is_delete=False).exists():
        user = await User.get(id=user_id)
        url = request.url.path
        path = url.split("/")[1]
        if path in ["user", "group", "permission"]:
            operator_map = {"GET": Operate.query, "POST": Operate.add, "PUT": Operate.update, "DELETE": Operate.delete}
            await OperationRecords.create(user=user, operate=operator_map[request.method])
    return response

通过中间件的方式将用户的行为记录到数据库中，用户的行为包括查询数据、更改数据、增加数据和删除数据。用户登录时会将用户的信息保存一个到浏览器的cookies中，用户操作时从request中获取用户的id查询登录的用户，这个示例中未实现用户登录功能所以指定用户id为1。

5、测试接口

经过上面的初始化数据库之后，数据库中生成了user、group和permission的表，之后添加一条默认权限和默认组记录

# 添加权限
insert into test.permission(`appname`, `name`, `action`) values("0", "defalut", "default")
# 添加用户组
insert into test.group(username)  values("default" )
# 默认权限绑定默认的组
insert into test.group_permission(group_id, permission_id) values(1, 1)

打开http://127.0.0.1:8000/docs网站测试

添加用户

添加权限

添加用户组

修改用户权限

上面操作的操作记录都会记录在数据库中