當前位置：首頁 > 运维知识 > 数据库 >内容正文

数据库

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

發布時間：2024/7/5 数据库 48 豆豆

生活随笔收集整理的這篇文章主要介紹了数据库和ORMS：使用Tortoise ORM与数据库通信小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

文章目錄

- 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 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 是需要手動編寫的

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

總結

以上是生活随笔為你收集整理的数据库和ORMS：使用Tortoise ORM与数据库通信的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

上一篇： LeetCode 1886. 判断矩阵经
下一篇： LeetCode MySQL 1777.