Python
菜单
本页目录

Pydantic serialisation 结合 tortoise-orm 快速入门

模型序列化

  1. 创建 tortoise-orm 模型
from tortoise import fields
from tortoise.models import Model

class Tournament(Model):
    """
    This references a Tournament
    """
    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=100)
    #: The date-time the Tournament record was created at
    created_at = fields.DatetimeField(auto_now_add=True)
  1. 创建 pydantic 模型
from tortoise.contrib.pydantic import pydantic_model_creator
# 调用pydantic_model_creator()函数，传入模型类，生成pydantic模型类
Tournament_Pydantic = pydantic_model_creator(Tournament)
  1. 查看 Pydantic 模型的 JSON-Schema
# 调用schema()方法，查看Pydantic模型的JSON-Schema
# 输出结果为JSON-Schema格式的字符串,包含类和文档字符串和文档注释
>>> print(Tournament_Pydantic.schema())
{
    'title': 'Tournament',
    'description': 'This references a Tournament',
    'type': 'object',
    'properties': {
        'id': {
            'title': 'Id',
            'type': 'integer'
        },
        'name': {
            'title': 'Name',
            'type': 'string'
        },
        'created_at': {
            'title': 'Created At',
            'description': 'The date-time the Tournament record was created at',
            'type': 'string',
            'format': 'date-time'
        }
    }
}
  1. 序列化对象
# 在Tournament模型上调用create()方法，创建新的数据库记录，返回一个tournament
tournament = await Tournament.create(name="New Tournament")
tourpy = await Tournament_Pydantic.from_tortoise_orm(tournament)

序列化：

  • 把内存中的各种数据类型的数据通过网络传送给其他机器或者客户端
  • 把内存中的各种数据类型的数据保存到本地磁盘持久化
  • 反序列化则相反，把网络传输或者本地磁盘持久化的数据转换为内存中的各种数据类型
  1. 查看 Pydantic 模型序列化后的对象
# 调用model_dump()方法，将pydantic模型对象序列化为python字典
>>> print(tourpy.model_dump())
{
    'id': 1,
    'name': 'New Tournament',
    'created_at': datetime.datetime(2020, 3, 1, 20, 28, 9, 346808)
}
# 调用model_dump_json()方法，将pydantic模型对象序列化为JSON字符串
>>> print(tourpy.model_dump_json())
{
    "id": 1,
    "name": "New Tournament",
    "created_at": "2020-03-01T20:28:09.346808"
}

列表模型序列化为查询集

  1. 创建 tortoise-orm 模型
from tortoise import fields
from tortoise.models import Model

class Tournament(Model):
    """
    This references a Tournament
    """
    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=100)
    #: The date-time the Tournament record was created at
    created_at = fields.DatetimeField(auto_now_add=True)

    class Meta:
        # Define the default ordering
        #  the pydantic serialiser will use this to order the results
        ordering = ["name"]
  1. 创建 pydantic 列表模型
# 调用pydantic_queryset_creator()函数，传入模型类，生成pydantic列表模型类
from tortoise.contrib.pydantic import pydantic_queryset_creator

Tournament_Pydantic_List = pydantic_queryset_creator(Tournament)
  1. 查看 Tournament_Pydantic_List 模型的JSON-Schema
# Schema()方法，查看Pydantic列表模型的JSON-Schema
# 注意现在 Tournament 不再是根节点了。取而代之的是一个简单的列表，即 Tournament_Pydantic_List。
>>> print(Tournament_Pydantic_List.schema())
{
    'title': 'Tournaments',
    'description': 'This references a Tournament',
    'type': 'array',
    'items': {
        '$ref': '#/definitions/Tournament'
    },
    'definitions': {
        'Tournament': {
            'title': 'Tournament',
            'description': 'This references a Tournament',
            'type': 'object',
            'properties': {
                'id': {
                    'title': 'Id',
                    'type': 'integer'
                },
                'name': {
                    'title': 'Name',
                    'type': 'string'
                },
                'created_at': {
                    'title': 'Created At',
                    'description': 'The date-time the Tournament record was created at',
                    'type': 'string',
                    'format': 'date-time'
                }
            }
        }
    }
}
  1. 序列化查询集
# Create objects
await Tournament.create(name="New Tournament")
await Tournament.create(name="Another")
await Tournament.create(name="Last Tournament")

tourpy = await Tournament_Pydantic_List.from_queryset(Tournament.all())
  1. 查看 Tournament_Pydantic_List 模型序列化后的对象
# 调用model_dump()方法，将pydantic模型对象序列化为python字典
>>> print(tourpy.model_dump())
{
    'root': [
        {
            'id': 2,
            'name': 'Another',
            'created_at': datetime.datetime(2020, 3, 2, 6, 53, 39, 776504)
        },
        {
            'id': 3,
            'name': 'Last Tournament',
            'created_at': datetime.datetime(2020, 3, 2, 6, 53, 39, 776848)
        },
        {
            'id': 1,
            'name': 'New Tournament',
            'created_at': datetime.datetime(2020, 3, 2, 6, 53, 39, 776211)
        }
    ]
}
# 调用model_dump_json()方法，将pydantic模型对象序列化为JSON字符串
>>> print(tourpy.model_dump_json())
[
    {
        "id": 2,
        "name": "Another",
        "created_at": "2020-03-02T06:53:39.776504"
    },
    {
        "id": 3,
        "name": "Last Tournament",
        "created_at": "2020-03-02T06:53:39.776848"
    },
    {
        "id": 1,
        "name": "New Tournament",
        "created_at": "2020-03-02T06:53:39.776211"
    }
]

注意：

  • .model_dump() 方法具有一个包含列表的根元素，而 .model_dump_json() 方法将列表作为根节点。
  • 同时还要注意结果 (名称) 按字母顺序排序。

实战演练

步骤：

  1. 创建 Tortoise-ORM 模型
  2. 创建 Pydantic 模型
  3. 初始化 Tortoise-ORM
  4. 创建和操作数据
  5. 创建API接口
  6. 启动fastapi服务器

安装依赖: 确保你已经安装了 FastAPI, Tortoise-ORM, pydantic, 和 uvicorn。你可以使用 pip 来安装这些依赖：

pip install fastapi tortoise-orm pydantic uvicorn

创建 Tortoise-ORM 模型

from tortoise.models import Model
from tortoise import fields

class Tournament(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=255)
    created_at = fields.DatetimeField(auto_now_add=True)
    updated_at = fields.DatetimeField(auto_now=True)

    class Meta:
        table = "tournament"

创建 Pydantic 模型

from pydantic import BaseModel
from datetime import datetime

class Tournament_Pydantic(BaseModel):
    id: int
    name: str
    created_at: datetime
    updated_at: datetime = None

    class Config:
        orm_mode = True

初始化 Tortoise-ORM

在项目启动时，你需要初始化 Tortoise-ORM 并连接到数据库。通常，这些操作会在项目的启动脚本或配置模块中完成。

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

# 配置 MySQL 数据库
TORTOISE_ORM = {
    "connections": {
        "default": {
            # "engine": "tortoise.backends.asyncpg",   # 数据库引擎 PostgresQL
            "engine": "tortoise.backends.mysql",  # 数据库引擎 Mysql or Mariadb
            "credentials": {
                "host": "127.0.0.1",    # 地址
                "port": "3306",         # 端口
                "user": "root",         # 用户名
                "password": "root",     # 密码
                "database": "fastapi",  # 数据库名称（需要提前创建数据库）
                "minsize": 1,           # 最少连接
                "maxsize": 5,           # 最大连接
                "charset": "utf8mb4",   # 编码
                "echo": True            # 是否反馈SQL语句
            }
        }
    },
    "apps": {
        "models": {
            "models": ["models", "aerich.models"],   # models数据模型迁移
            "default_connection": "default"
        }
    },
    "use_tz": False,
    "timezone": "Asia/Shanghai"
}

# 创建实例
app = FastAPI()

# 注册Tortoise-ORM, 并生成数据库表结构
register_tortoise(
    app,
    config=TORTOISE_ORM,
    generate_schemas=True,
)

创建和操作数据

创建数据记录并将其转换为 Pydantic 模型的过程通常会在业务逻辑层（例如，服务层）中完成。

async def create_tournament(name: str):
    # 创建新的比赛记录
    tournament = await Tournament.create(name=name)
    # 使用 Pydantic 模型进行序列化
    tournament_pydantic = await Tournament_Pydantic.from_tortoise_orm(tournament)
    return tournament_pydantic

创建API接口

在这一步中，我们将使用 FastAPI 创建一个 API 端点，使客户端能够通过 HTTP 请求与数据库进行交互。我们会使用我们之前定义的 create_tournament 函数来处理请求，并将 Tortoise-ORM 和 Pydantic 模型结合起来进行数据处理。

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

# 创建 FastAPI 应用实例
app = FastAPI()

class TournamentCreateRequest(BaseModel):
    name: str # 请求体中的字段：比赛名称

@app.post("/tournaments/", response_model=Tournament_Pydantic)
async def create_tournament_endpoint(request: TournamentCreateRequest):
    """
    创建一个新的比赛记录。
    
    请求体（body）：TournamentCreateRequest
    响应体（body）：Tournament_Pydantic
    """
    try:
       # 调用先前定义的函数，创建比赛记录并序列化为 Pydantic 模型
        tournament_pydantic = await create_tournament(request.name)
        return tournament_pydantic # 返回创建的比赛记录
    except IntegrityError as e:
        # 捕获数据库完整性错误，例如唯一性约束失败
        raise HTTPException(status_code=400, detail="Integrity error")
    except Exception as e:
        # 捕获其他可能的异常
        raise HTTPException(status_code=500, detail=str(e))

启动fastapi服务器

# 启动 FastAPI 服务器
if __name__ == "__main__":
    import uvicorn
    # 启动服务器，监听 0.0.0.0 的 8000 端口
    uvicorn.run(app, host="0.0.0.0", port=8000，reload=True)
上一篇 Tortoise-ORM FastAPI integration 中文文档完整版
下一篇 Pydantic-serialisation 结合 tortoise-orm 使用文档（完整版）