fastapi中使用tortoise-orm(入门)_tortoise-orm的in_transaction

Tortoise-orm的使用

该数据模型参照Django的数据模型，因此使用上与Django的数据模型高度相似。
!!! 注意，tortoise-orm是异步框架。

官方文档：
https://tortoise-orm.readthedocs.io/en/latest/index.html
1
2
3
4
5

安装tortoise

pip install tortoise
1

安装数据模型迁移工具

pip install aerich==0.6.3

我用的mysql，因此还需要安装aiomysql包：
pip install aiomysql

aerich的功能类似于django的migrate。
1
2
3
4
5
6

配置

TORTOISE_ORM = {
	"connections": {
    	"default": {
       		'engine': 'tortoise.backends.mysql',  # MySQL or Mariadb
       		'credentials': {
            	'host': '127.0.0.1',
            	'port': '3306',
                'user': 'root',
                'password': '*****',
                'database': 'test',
                'minsize': 1,
                'maxsize': 5,
                'charset': 'utf8mb4',
                'echo': True
       		}
    	}
	},
	"apps": {
    	"models": {
        	"models": ["your_models_path", "aerich.models"],  
        	#  your_models_path: 例如my_api.models; 
        	# aerich.models必须要有，但不用创建对应的models文件
        	"default_connection": "default",
    	}
	},
	"use_tz": False,  # 建议不要开启，不然存储日期时会有很多坑，时区转换在项目中手动处理更稳妥。
	"timezone": "Asia/Shanghai"
}
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

在APP中导入models配置

from fastapi import FastAPI
from tortoise.contrib.fastapi import register_tortoise


api_app = FastAPI()
api_app.include_router(routers.meta_router, prefix="/v1/meta")

register_tortoise(
    app=api_app,
    config=TORTOISE_ORM,
    # generate_schemas=True,  # 如果数据库为空，则自动生成对应表单，生产环境不要开
    # add_exception_handlers=True,  # 生产环境不要开，会泄露调试信息
)
1
2
3
4
5
6
7
8
9
10
11
12
13

建立数据模型

# 创建models.py文件，即TORTOISE_ORM中models参数中指定的models文件。

from tortoise import fields
from tortoise.models import Model


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 CoreUser(DateTimeModel):
    id = fields.IntField(pk=True, index=True)
    username = fields.CharField(max_length=255, description="用户姓名")  
    # 与django不同的是，这里的description可以直接在数据库表中看到哦。
    password = fields.CharField(max_length=255, description="用户密码")  # 密码不要存明文，存hash值。
    is_delete = fields.BooleanField(default=False, description="是否删除")

    class Meta:
        table = "core_user"  # 数据库中的表名称
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

ORM数据库迁移操作

1. 打开命令行，切换到项目根目录
2. 初始化配置项

	aerich init -t db.config.TORTOISE_ORM
	
	初始化完成后会在当前目录生成一个文件pyproject.toml和一个文件夹migrations
	· pyproject.toml: 保存配置文件路径
	. migrations：存放.sql迁移文件 
3. 初始化数据库，一般情况下只用一次
	
	aerich init-db
	
	· 如果TORTOISE_ORM配置文件中的models改了名，执行这条命令时需要增加--app参数，来指定你修改的名字。
	· 在migrations的指定app目录下生成sql文件（如果model不为空时），并在数据库中生成表。
4. 更新模型并进行迁移

	aerich migrate --name any_name
	
	· 每次修改model后执行此命令，将在migrations文件夹下生成.sql迁移文件；
	· --name参数为你的迁移文件添加备注，默认为update；
	· 迁移文件名的格式为{version_num}{datetime}{any_name|update}.sql；
	· 如果aerich识别到您正在重命名列，它会要求重命名{old_column}为{new_column} [True]，您可以选择True
	   重命名列而不删除列，或者选择False删除列然后创建，如果使用Mysql，只有8.0+支持重命名。
5. 更新最新修改到数据库

	aerich upgrade [xxx.sql]

6. 后续每次修改models文件内容（新增、删除或修改数据模型， 非models文件重命名），只需执行步骤4、5即可。
7. 其它操作
	a. 降级到指定版本
		aerich downgrade -v 指定版本 -d 降级的同时删除迁移文件 --yes 确认删除，不再交互式输入
	b. 显示当前可以迁移的版本
		aerich heads
	c. 显示迁移历史
		aerich history
8. 查看数据库，验证迁移操作。
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

增删改查

from fastapi import APIRouter

meta_router = APIRouter()  # 需要在app中进行注册
# api_app = FastAPI()
# api_app.include_router(routers.meta_router, prefix="/v1/meta")


@meta_router.post("/user/createUser")
async def create_user(user: CreateUser):
    core_suer = await CoreUser.create(**user.dict())  # 操作也与django保持高度一致，返回创建成功的CoreUser对象
    return core_suer.id


@meta_router.put("/user/updateUser")
async def create_user(user: UpdateUser):
    num = await CoreUser.filter(id=user.user_id).update(
        username=user.username, password=user.password
    )
    return num  # 受影响的笔数，如果数据没有任何变化，则返回0条。注意这里返回值与django有差异，更实用。


@meta_router.get("/user/list")
async def get_user_list():
    query_set = await CoreUser.all()
    # 注意这里返回的query_set实际是个列表。List[CoreUser]
    records = []
    count = len(query_set）
    for obj in query_set:
        records.append(dict(
            username=obj.username,
            password=obj.password,
        ))
    return dict(count=count, records=records)


@meta_router.get("/user/info")
async def get_user(user_id):
    obj = await CoreUser.filter(id=user_id).first()
    return dict(username=obj.username, password=obj.password)
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

事务

使用下面两种方式，都可以保证所属的数据库操作都在同一个事务中执行，如果中间有代码执行失败，已经执行的数据库操作会被回滚。

第一种方式：装饰器

# 装饰器针对整个被修饰的方法
from tortoise import transactions


@transactions.atomic()
async def create_user():
	pass
1
2
3
4
5
6
7

第二种方式：使用in_transaction

# in_transaction针对with语句下的所有操作
from tortoise import transactions


async def create_user():
	async with transactions.in_transaction():
		pass
1
2
3
4
5
6
7