Fastapi0.115.6之Tortoise ORM0.23.0基本增删改查大全【亲测可用，仅供参考】

自己在有限的时间里写了一个Fastapi之Tortoise ORM增删改查大全，如有写错的地方还请多多指点。接口功能有可能不全，我会迭代的。主要是方便查找。先放出来一下浏览器的接口文档截图

一、models或ORM数据表文件

from tortoise.models import Model
from tortoise import fields

class P_user(Model):
    p_id = fields.IntField(pk=True)
    p_user = fields.CharField(null=False, unique=True, max_length=12, index=True, description='用户账号')
    p_pass = fields.CharField(null=False, unique=True, max_length=256, description='用户密码')
    role = fields.IntField(null=False, verbose_name="角色")
    permissions = fields.IntField(null=False, verbose_name="权限")
    created_at = fields.DatetimeField(auto_now_add=True, verbose_name="创建日期")
    is_deleted = fields.BooleanField(default=False, verbose_name="软删除")

    class Meta:
        db_table = 'p_user'


class P_Login(Model):
    p_id = fields.IntField(pk=True)
    p_user = fields.CharField(null=False, unique=True, max_length=18, description="用户账号")
    p_pass = fields.CharField(null=False, unique=True, max_length=256, description='用户密码')
    p_pic = fields.CharField(null=False, max_length=256, description="用户头像")
    is_deleted = fields.BooleanField(default=False, verbose_name="软删除")

    class Meta:
        db_table = 'p_login'

二、基础接口案例

import logging
from typing import Optional, List
from fastapi_pagination import Page, add_pagination
from fastapi_pagination.ext.tortoise import paginate as tortoise_paginate
from fastapi import APIRouter, HTTPException, Query, Body, Depends, Request
from fastapi.templating import Jinja2Templates
from pydantic import BaseModel, field_validator
from starlette import status
from tortoise.contrib.fastapi import HTTPNotFoundError
from study_ORM.models import P_user

# 设置日志记录
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

p_user_api = APIRouter()

# 查询所有用户信息
@p_user_api.get("/A", summary="查询所有信息接口(不带分页功能)")
async def get_all_users():
    users = await P_user.all()
    logger.info("全部信息查询: %s", users)
    return {"操作结果": users}

# 查询单个字段信息
@p_user_api.get("/B", summary="查询单个字段里的信息接口(精准搜索)")
async def get_user_by_username(p_user: str):
    user = await P_user.filter(p_user=p_user)
    logger.info("过滤查询: %s", user)
    return {"操作结果": user}

# 根据ID查询用户信息
@p_user_api.get("/C", summary="按ID号进行查询")
async def get_user_by_id(p_id: int):
    user = await P_user.get(p_id=p_id)
    logger.info("get查询: %s", user.p_user)
    return {"操作结果": user}

# 模糊查询
@p_user_api.get("/D", summary="按某个字段进行模糊查询")
async def get_user_by_partial_username(p_user: str):
    users = await P_user.filter(p_user__icontains=p_user)
    logger.info("模糊查询: %s", users)
    return {"操作结果": users}

# 查询两个字段信息
@p_user_api.get("/E", summary="只查询其中两个字段里的所有信息")
async def get_user_values():
    users = await P_user.all().values("permissions", "p_user")
    logger.info("values查询: %s", users)
    return {"操作结果": users}

# 分页查询
class UserOut(BaseModel):
    p_user: str
    p_pass: str
    role: Optional[int] = None
    permissions: Optional[int] = None
    is_deleted: Optional[bool]

@p_user_api.get("/p_users/page", response_model=Page[UserOut], summary="分页接口")
async def get_paginated_users():
    return await tortoise_paginate(P_user)
add_pagination(p_user_api)

# 前后端不分离查询
@p_user_api.get("/index", summary="前后端不分离查询")
async def get_all_users_html(request: Request):
    templates = Jinja2Templates(directory="templates")
    users = await P_user.all()
    return templates.TemplateResponse("index.html", {"request": request, "userinfo": users})

# 用户输入校验模型
class P_userIn(BaseModel):
    p_user: str = Body(..., description="项目名称")
    p_pass: str
    role: Optional[int] = None
    permissions: Optional[int] = None
    is_deleted: Optional[bool] = False

    @field_validator("p_user")
    def check_p_user(cls, v):
        if not v:
            raise ValueError('用户名不能为空')
        if not v.isalnum():
            raise ValueError('用户名只能包含字母和数字')
        if len(v) > 12:
            raise ValueError("用户名太长。最多允许 12 个字符。")
        return v

    @field_validator("p_pass")
    def check_p_pass(cls, v):
        if not v:
            raise ValueError('密码不能为空')
        if not 6 <= len(v) <= 16:
            raise ValueError('您输入的密码必须在6到16个字符之间')
        if not any(c.isupper() for c in v) or not any(c.islower() for c in v) or not any(c.isdigit() for c in v):
            raise ValueError('密码必须包含大小写字母和数字')
        return v

    @field_validator("role")
    def check_p_role(cls, v):
        if v is None:
            raise ValueError('角色不能为空')
        if not isinstance(v, int):
            raise TypeError('角色必须为整数')
        if v < 0:
            raise ValueError('角色不能为负数')
        return v

    @field_validator("permissions")
    def check_p_permissions(cls, v):
        if v is None:
            raise ValueError('权限不能为空')
        if not isinstance(v, int):
            raise TypeError('权限必须为整数')
        if v < 0:
            raise ValueError('权限不能为负数')
        return v

# 添加用户信息
@p_user_api.post("/", summary="信息添加接口")
async def add_user(form_data: P_userIn = Depends()):
    try:
        user = await P_user.create(**form_data.model_dump())
        return user
    except Exception as e:
        logger.error(f"数据库操作失败: {e}")
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"创建用户失败: {e}")

# 查询一条用户信息
@p_user_api.get("/{p_id}", summary="查询一条信息接口")
async def get_one_user(p_id: int):
    user = await P_user.get(p_id=p_id)
    return user

# 更新用户信息
@p_user_api.put("/{p_id}", summary="更新信息接口")
async def update_user(p_id: int, form_data: P_userIn = Depends()):
    data = form_data.model_dump()
    await P_user.filter(p_id=p_id).update(**data)
    return {"操作": f"更新id={p_id}一个用户"}

# 硬删除用户信息
@p_user_api.delete("/{p_id}", summary="硬删除一条信息接口")
async def delete_user(p_id: int):
    deleted_count = await P_user.filter(p_id=p_id).delete()
    if not deleted_count:
        raise HTTPException(status_code=404, detail=f"主键为{p_id}用户不存在")
    return {"操作": f"删除id={p_id}一个用户"}

# 硬删除多条用户信息
@p_user_api.delete("/", summary="硬删除多条信息接口")
async def delete_users(p_ids: List[int] = Query(..., description="要删除的用户ID列表")):
    deleted_count = await P_user.filter(p_id__in=p_ids).delete()
    if deleted_count != len(p_ids):
        existing_ids = await P_user.filter(p_id__in=p_ids).values_list('p_id', flat=True)
        not_found_ids = list(set(p_ids) - set(existing_ids))
        raise HTTPException(status_code=404, detail=f"以下用户ID不存在: {not_found_ids}")
    return {"message": f"成功删除 {deleted_count} 个用户", "deleted_ids": p_ids}

# 软删除用户信息
@p_user_api.delete("/user/{p_id}", summary="软删除一条信息接口")
async def soft_delete_user(p_id: int):
    deleted_count = await P_user.filter(p_id=p_id).update(is_deleted=True)
    if not deleted_count:
        raise HTTPException(status_code=404, detail=f"主键为{p_id}用户不存在")
    return {"操作": f"删除id={p_id}一个用户"}

# 软删除多条用户信息
@p_user_api.delete("/user/", summary="软删除多条信息接口")
async def soft_delete_users(p_ids: List[int] = Query(...)):
    if not p_ids:
        raise HTTPException(status_code=400, detail="请提供要删除的用户 ID 列表")
    deleted_count = await P_user.filter(p_id__in=p_ids).update(is_deleted=True)
    if deleted_count != len(p_ids):
        not_found_ids = list(
            set(p_ids) - set(await P_user.filter(p_id__in=p_ids, is_deleted=True).values_list('p_id', flat=True)))
        if not_found_ids:
            raise HTTPException(status_code=404, detail=f"以下用户 ID 不存在: {not_found_ids}")
    return {"操作": f"成功删除 {deleted_count} 个用户", "删除的ID": p_ids}

三、带文件上传功能接口案例

import os
import uuid
from typing import List

import bcrypt
from fastapi import APIRouter, File, UploadFile, HTTPException, Depends, Query
from pydantic import BaseModel, field_validator
from starlette import status
from tortoise.contrib.fastapi import HTTPNotFoundError
from tortoise.exceptions import IntegrityError

from study_ORM.models import P_Login
from study_ORM.utils import allowed_file, validate_image, save_upload_file

p_login_api = APIRouter()

UPLOAD_DIR = "uploads"
os.makedirs(UPLOAD_DIR, exist_ok=True)


class UploadForm(BaseModel):
    p_user: str
    p_pass: str
    is_deleted: bool = False

    @field_validator("p_user")
    def check_p_user(cls, v):
        if not v:
            raise ValueError('用户名不能为空')
        if not v.isalnum():
            raise ValueError('用户名只能包含字母和数字')
        if not 1 <= len(v) <= 12:
            raise ValueError("用户名长度必须在 1 到 12 个字符之间。")
        return v

    @field_validator("p_pass")
    def check_p_pass(cls, v):
        if not v:
            raise ValueError('密码不能为空')
        if not 6 <= len(v) <= 16:
            raise ValueError('密码长度必须在 6 到 16 个字符之间')
        if not any(c.isupper() for c in v) or not any(c.islower() for c in v) or not any(c.isdigit() for c in v):
            raise ValueError('密码必须包含大小写字母和数字')
        return bcrypt.hashpw(v.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')

    @field_validator("is_deleted")
    def check_is_deleted(cls, v):
        if not isinstance(v, bool):
            raise ValueError("软删除必须是布尔类型")
        return v


@p_login_api.post("/upload/", status_code=status.HTTP_201_CREATED, summary="信息添加接口",
                  description="实现信息添加和图片上传")
async def upload_image(file: UploadFile = File(...), form_data: UploadForm = Depends()):
    if not allowed_file(file.filename):
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="无效的文件类型")

    validate_image(file)

    file_ext = file.filename.rsplit('.', 1)[1].lower()
    unique_filename = str(uuid.uuid4()) + "." + file_ext
    file_path = os.path.join(UPLOAD_DIR, unique_filename)

    try:
        await save_upload_file(file, file_path)

        image = await P_Login.create(
            p_user=form_data.p_user,
            p_pass=form_data.p_pass,
            p_pic=file_path,
            is_deleted=form_data.is_deleted
        )
        return {"id": image.p_id, "path": image.p_pic, "user": image.p_user}
    except IntegrityError as e:
        os.remove(file_path)
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=f"用户已存在: {e}")
    except Exception as e:
        if os.path.exists(file_path):
            os.remove(file_path)
        print(f"数据库操作失败: {e}")
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"创建用户失败: {e}")


@p_login_api.delete("/{p_id}", summary="单条信息软删除接口", description="软删除单条信息")
async def delete_p_login(p_id: int):
    updated_count = await P_Login.filter(p_id=p_id).update(is_deleted=True)
    if not updated_count:
        raise HTTPNotFoundError(detail=f"主键为 {p_id} 的用户不存在")
    return {"message": f"成功软删除 id 为 {p_id} 的用户"}


@p_login_api.delete("/", summary="多条信息软删除接口", description="软删除多条信息")
async def delete_multiple_p_users(p_ids: List[int] = Query(...)):
    if not p_ids:
        raise HTTPException(status_code=400, detail="请提供要删除的用户 ID 列表")

    updated_count = await P_Login.filter(p_id__in=p_ids).update(is_deleted=True)
    if updated_count != len(p_ids):
        existing_ids = await P_Login.filter(p_id__in=p_ids).values_list('p_id', flat=True)
        not_found_ids = list(set(p_ids) - set(existing_ids))
        if not_found_ids:
            raise HTTPException(status_code=404, detail=f"以下用户 ID 不存在: {not_found_ids}")

    return {"message": f"成功软删除 {updated_count} 个用户，ID 为 {p_ids}"}


@p_login_api.delete("/users/{p_id}", summary="单条信息硬删除接口", description="硬删除单条信息及关联文件")
async def delete_user(p_id: int):
    user = await P_Login.get_or_none(p_id=p_id)
    if not user:
        raise HTTPNotFoundError

    file_path = user.p_pic
    if file_path and os.path.exists(file_path):
        try:
            os.remove(file_path)
        except OSError as e:
            print(f"删除文件出错: {e}")

    await user.delete()
    return {"message": "用户删除成功"}


@p_login_api.delete("/users/", summary="多条信息硬删除接口", description="硬删除多条信息及关联文件")
async def delete_users(user_ids: List[int] = Query(...)):
    if not user_ids:
        raise HTTPException(status_code=400, detail="未提供用户 ID")

    deleted_count = 0
    for user_id in user_ids:
        user = await P_Login.get_or_none(p_id=user_id)
        if user:
            file_path = user.p_pic
            if file_path and os.path.exists(file_path):
                try:
                    os.remove(file_path)
                except OSError as e:
                    print(f"删除用户 {user_id} 的文件出错: {e}")
            await user.delete()
            deleted_count += 1

    if deleted_count == 0:
        raise HTTPNotFoundError(detail="未找到指定 ID 的用户")

    return {"message": f"成功删除 {deleted_count} 个用户"}


class UploadForm(BaseModel):
    p_user: str
    p_pass: str
    is_deleted: bool = False

    @field_validator("p_user")
    def check_p_user(cls, v):
        if not v:
            raise ValueError('用户名不能为空')
        if not v.isalnum():
            raise ValueError('用户名只能包含字母和数字')
        if not 1 <= len(v) <= 12:
            raise ValueError("用户名长度必须在 1 到 12 个字符之间。")
        return v

    @field_validator("p_pass")
    def check_p_pass(cls, v):
        if not v:
            raise ValueError('密码不能为空')
        if not 6 <= len(v) <= 16:
            raise ValueError('密码长度必须在 6 到 16 个字符之间')
        if not any(c.isupper() for c in v) or not any(c.islower() for c in v) or not any(c.isdigit() for c in v):
            raise ValueError('密码必须包含大小写字母和数字')
        return bcrypt.hashpw(v.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')

    @field_validator("is_deleted")
    def check_is_deleted(cls, v):
        if not isinstance(v, bool):
            raise ValueError("软删除必须是布尔类型")
        return v


@p_login_api.put("/up/{p_id}", status_code=status.HTTP_200_OK, summary="更新信息接口",
                 description="实现信息与头像的更新功能")
async def update_user(p_id: int, file: UploadFile = File(None), form_data: UploadForm = Depends()):
    user = await P_Login.get_or_none(p_id=p_id)
    if not user:
        raise HTTPNotFoundError(detail=f"ID 为 {p_id} 的用户不存在")

    if file:
        if not allowed_file(file.filename):
            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="无效的文件类型")
        validate_image(file)

        file_path = user.p_pic  # 获取旧文件路径
        if file_path and os.path.exists(file_path):
            try:
                os.remove(file_path)  # 先删除旧文件
            except OSError as e:
                print(f"删除旧文件出错: {e}")

        file_ext = file.filename.rsplit('.', 1)[1].lower()
        unique_filename = str(uuid.uuid4()) + "." + file_ext
        file_path = os.path.join(UPLOAD_DIR, unique_filename)
        await save_upload_file(file, file_path)
    else:
        file_path = user.p_pic  # 如果没有新文件上传，则保持原路径

    try:
        await P_Login.filter(p_id=p_id).update(
            p_user=form_data.p_user,
            p_pass=form_data.p_pass,
            p_pic=file_path,
            is_deleted=form_data.is_deleted
        )
        return {"message": f"成功更新 ID 为 {p_id} 的用户"}
    except IntegrityError as e:
        if file and os.path.exists(file_path):  # 如果数据库操作失败，并且新文件已保存，则删除新文件
            os.remove(file_path)
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=f"更新用户失败: {e}")
    except Exception as e:
        if file and os.path.exists(file_path):  # 如果数据库操作失败，并且新文件已保存，则删除新文件
            os.remove(file_path)
        print(f"数据库操作失败: {e}")
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"更新用户信息失败: {e}")


@p_login_api.get("/A", summary="查询所有信息接口", description="查询数据表里所有信息，不带分页")
async def get_all_users():
    userinfo = await P_Login.all()
    return userinfo