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 fieldsclass PostBase(BaseModel):title: strcontent: strpublication_date: datetime = Field(default_factory=datetime.now)class Config:orm_mode = True# 此选项将允许我们将ORM对象实例转换为Pydantic对象实例# 因为FastAPI设计用Pydantic模型，而不是ORM模型class PostPartialUpdate(BaseModel):title: Optional[str] = Nonecontent: Optional[str] = Noneclass PostCreate(PostBase):passclass PostDB(PostBase):id: intclass 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, PostTortoiseapp = 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 uvicornuvicorn.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 = paginationposts = 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: intpublication_date: datetime = Field(default_factory=datetime.now)content: strclass Config:orm_mode = Trueclass CommentCreate(CommentBase):passclass CommentDB(CommentBase):id: intclass 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 DoesNotExistapp = 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 = paginationposts = 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 uvicornuvicorn.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"

• 应用迁移

aerich upgrade

aerich migrate --name added_new_tables

注意：Aerich 迁移脚本 不兼容 跨数据库，在本地和生产环境中都应该使用相同的数据库引擎

• 降级

aerich downgrade

• 迁移历史

aerich history