数据库和ORMS：使用Tortoise ORM与数据库通信

Michael阿明

发布于 2022-11-27 17:22:22

1.9K0

发布于 2022-11-27 17:22:22

文章被收录于专栏：Michael阿明学习之路Michael阿明学习之路

文章目录
1. 安装环境
2. 创建数据库模型
3. 设置 `Tortoise` 引擎
4. create
5. 查询
6. 修改、删除
7. 添加关联
8. 用Aerich建立数据库迁移系统

learn from 《Building Data Science Applications with FastAPI》

Tortoise ORM 是一种现代异步 ORM，非常适合 FastAPI项目

1. 安装环境

pip install tortoise-orm

https://tortoise-orm.readthedocs.io/en/latest/getting_started.html#installation

本文目录结构

2. 创建数据库模型

# _*_ coding: utf-8 _*_
# @Time : 2022/3/18 9:30
# @Author : Michael
# @File : models.py
# @desc :

from datetime import datetime
from typing import Optional
from pydantic import BaseModel, Field
from tortoise.models import Model
from tortoise import fields

class PostBase(BaseModel):
    title: str
    content: str
    publication_date: datetime = Field(default_factory=datetime.now)
    class Config:
        orm_mode = True
        # 此选项将允许我们将ORM对象实例转换为Pydantic对象实例
        # 因为FastAPI设计用Pydantic模型，而不是ORM模型

class PostPartialUpdate(BaseModel):
    title: Optional[str] = None
    content: Optional[str] = None

class PostCreate(PostBase):
    pass

class PostDB(PostBase):
    id: int

class PostTortoise(Model):
    id = fields.IntField(pk=True, generated=True) 
    # pk=True 表示主键
    publication_date = fields.DatetimeField(null=False)
    title = fields.CharField(max_length=255, null=False)
    content = fields.TextField(null=False)
    class Meta:
        table = 'posts'

https://tortoise-orm.readthedocs.io/en/latest/fields.html

3. 设置 `Tortoise` 引擎

设置数据库位置、模型等
他可以自动启动和关闭连接，当你启动和关闭app时，之前的 SQLAlchemy 是需要手动编写的

# _*_ coding: utf-8 _*_
# @Time : 2022/3/18 9:57
# @Author : Michael
# @File : app.py
# @desc :

from tortoise.contrib.fastapi import register_tortoise
from typing import Optional, List, Tuple
from fastapi import FastAPI, Depends, Query, status
from web_python_dev.sql_tortoise_orm.models import PostDB, PostCreate, PostPartialUpdate, PostTortoise

app = FastAPI()

TORTOISE_ORM = {
    "connections": {"default": "sqlite://cp6_tortoise.db"},
    "apps": {
        "models": {
            "models": ["web_python_dev.sql_tortoise_orm.models"],
            "default_connection": "default",
        },
    },
}

register_tortoise(
    app,
    config=TORTOISE_ORM,
    generate_schemas=True,
    add_exception_handlers=True,
)

db_url 设置参考： https://tortoise-orm.readthedocs.io/en/latest/databases.html?highlight=db_url#db-url

4. create

async def pagination(skip: int = Query(0, ge=0), limit: int = Query(10, ge=0)) -> Tuple[int, int]:
    # 限制查询页面的显示条数
    capped_limit = min(limit, 100)
    return (skip, capped_limit)


async def get_post_or_404(id: int) -> PostTortoise:
    return await PostTortoise.get(id=id)

@app.post("/posts", response_model=PostDB, status_code=status.HTTP_201_CREATED)
async def create_post(post: PostCreate) -> PostDB:
    post_tortoise = await PostTortoise.create(**post.dict())
    return PostDB.from_orm(post_tortoise)
    # 因为 pydantic 中 开启了 orm_mode = True
    # 将 PostTortoise 转换成 Pydantic 模型

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app="app:app", host="127.0.0.1", port=8001, reload=True, debug=True)

5. 查询

https://tortoise-orm.readthedocs.io/en/latest/query.html#query-api

@app.get("/posts")
async def list_posts(pagination: Tuple[int, int] = Depends(pagination)) -> List[PostDB]:
    skip, limit = pagination
    posts = await PostTortoise.all().offset(skip).limit(limit)
    results = [PostDB.from_orm(post) for post in posts]
    return results


@app.get("/posts/{id}", response_model=PostDB)
async def get_post(post: PostTortoise = Depends(get_post_or_404)) -> PostDB:
    return PostDB.from_orm(post)

6. 修改、删除

@app.patch("/posts/{id}", response_model=PostDB)
async def update_post(post_update: PostPartialUpdate,
                      post: PostTortoise = Depends(get_post_or_404)) -> PostDB:
    post.update_from_dict(post_update.dict(exclude_unset=True))
    await post.save()
    return PostDB.from_orm(post)

@app.delete("/posts/{id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_post(post: PostTortoise = Depends(get_post_or_404)) -> None:
    await post.delete()

7. 添加关联

models.py

# 评论类
class CommentBase(BaseModel):
    post_id: int
    publication_date: datetime = Field(default_factory=datetime.now)
    content: str

    class Config:
        orm_mode = True


class CommentCreate(CommentBase):
    pass


class CommentDB(CommentBase):
    id: int

class PostPublic(PostDB):
    comments: List[CommentDB]

	# list强制转换 tortoise-orm 到 pydantic
	# pre=True 在pydantic验证之前进行调用
    @validator("comments", pre=True)
    def fetch_comments(cls, v):
        return list(v)


class CommentTortoise(Model):
	# 主键
    id = fields.IntField(pk=True, generated=True)
	# 外键
    post = fields.ForeignKeyField(
        "models.PostTortoise", related_name="comments", null=False
    )
    publication_date = fields.DatetimeField(null=False)
    content = fields.TextField(null=False)

    class Meta:
        table = "comments"

app.py

# _*_ coding: utf-8 _*_
# @Time : 2022/3/18 9:57
# @Author : Michael
# @File : app.py
# @desc :

from tortoise.contrib.fastapi import register_tortoise
from typing import Optional, List, Tuple
from fastapi import FastAPI, Depends, Query, status, HTTPException
from web_python_dev.sql_tortoise_orm.models import PostDB, PostCreate, PostPartialUpdate, PostTortoise, CommentDB, \
    CommentBase, CommentTortoise, PostPublic
from tortoise.exceptions import DoesNotExist

app = FastAPI()

TORTOISE_ORM = {
    "connections": {"default": "sqlite://cp6_tortoise.db"},
    "apps": {
        "models": {
            "models": ["web_python_dev.sql_tortoise_orm.models"],
            "default_connection": "default",
        },
    },
}

register_tortoise(
    app,
    config=TORTOISE_ORM,
    generate_schemas=True,
    add_exception_handlers=True,
)


async def pagination(skip: int = Query(0, ge=0), limit: int = Query(10, ge=0)) -> Tuple[int, int]:
    # 限制查询页面的显示条数
    capped_limit = min(limit, 100)
    return (skip, capped_limit)


async def get_post_or_404(id: int) -> PostTortoise:
    try:
        return await PostTortoise.get(id=id).prefetch_related("comments")
    except DoesNotExist:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)


@app.get("/posts")
async def list_posts(pagination: Tuple[int, int] = Depends(pagination)) -> List[PostDB]:
    skip, limit = pagination
    posts = await PostTortoise.all().offset(skip).limit(limit)
    results = [PostDB.from_orm(post) for post in posts]
    return results


@app.get("/posts/{id}", response_model=PostPublic)
async def get_post(post: PostTortoise = Depends(get_post_or_404)) -> PostPublic:
    return PostPublic.from_orm(post)


@app.post("/posts", response_model=PostPublic, status_code=status.HTTP_201_CREATED)
async def create_post(post: PostCreate) -> PostPublic:
    post_tortoise = await PostTortoise.create(**post.dict())
    await post_tortoise.fetch_related("comments")
    return PostPublic.from_orm(post_tortoise)
    # 因为 pydantic 中 开启了 orm_mode = True
    # 将 PostTortoise 转换成 Pydantic 模型


@app.patch("/posts/{id}", response_model=PostPublic)
async def update_post(post_update: PostPartialUpdate,
                      post: PostTortoise = Depends(get_post_or_404)) -> PostPublic:
    post.update_from_dict(post_update.dict(exclude_unset=True))
    await post.save()
    return PostPublic.from_orm(post)


@app.delete("/posts/{id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_post(post: PostTortoise = Depends(get_post_or_404)) -> None:
    await post.delete()


@app.post("/comments", response_model=CommentDB, status_code=status.HTTP_201_CREATED)
async def create_comment(comment: CommentBase) -> CommentDB:
    try:
        await PostTortoise.get(id=comment.post_id)
    except DoesNotExist:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
                            detail=f"Post {comment.post_id} does not exist.")
    comment_tortoise = await CommentTortoise.create(**comment.dict())
    return CommentDB.from_orm(comment_tortoise)


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app="app:app", host="127.0.0.1", port=8001, reload=True, debug=True)

8. 用Aerich建立数据库迁移系统

该工具由 Tortoise 创建者提供

pip install aerich

app.py 中有的配置信息

TORTOISE_ORM = {
    "connections": {"default": "sqlite://cp6_tortoise.db"},
    "apps": {
        "models": {
            "models": ["aerich.models", "web_python_dev.sql_tortoise_orm.models"],
            # 需要额外添加 "aerich.models"
            "default_connection": "default",
        },
    },
}

初始化迁移环境

(cv) PS D:\gitcode\Python_learning> aerich init -t web_python_dev.sql_tortoise_orm.app.TORTOISE_ORM
Success create migrate location ./migrations
Success write config to pyproject.toml

(cv) PS D:\gitcode\Python_learning> aerich init-db                                                 
Success create app migrate location migrations\models
Success generate schema for app "models"