learn from 《Building Data Science Applications with FastAPI》
Tortoise ORM 是一种现代异步 ORM,非常适合 FastAPI项目
pip install tortoise-orm
https://tortoise-orm.readthedocs.io/en/latest/getting_started.html#installation
本文目录结构
# _*_ 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
Tortoise
引擎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
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)
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)
@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()
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)
该工具由 Tortoise 创建者提供
pip install aerich
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