首页 > 其他分享 >FastAPI + tortoise-orm基础使用

FastAPI + tortoise-orm基础使用

时间:2022-10-23 20:37:28浏览次数:44  
标签:tortoise models FastAPI site orm import data id MaintainedData

更改sqlite为mysql

from tortoise import Tortoise
import asyncio
async def init():
    user = 'root'
    password = '123456'
    db_name = 'test'
    await Tortoise.init(
        #指定mysql信息
        db_url=f'mysql://{user}:{password}@127.0.0.1:3306/{db_name}',
        #指定models
        modules={'models': ['sanic_bp.models']}
    )
    #按照模型生成表
    # await Tortoise.generate_schemas()
    asyncio.run(init())

在sanic中使用:

from sanic import Sanic
from tortoise import Tortoise
 
app = Sanic(__name__)
 
@app.listener('after_server_start')
async def notify_server_started(app, loop):
    print('sanic sanic服务启动后建立mysql连接')
    #实例化mysql连接
    await Tortoise.init(
        db_url='mysql://root:[email protected]:3306/test?maxsize=50&minsize=3',
        modules={'models': ['sanic_bp.models']}
    )

基础使用

 
from tortoise.contrib.sanic import register_tortoise
 
register_tortoise(
    app, db_url="sqlite://:memory:", modules={"models": ["models"]}, generate_schemas=True
)
 

sanic 和 fastapi中使用
tortoise-orm 连接多个库

async def run():
    await Tortoise.init(
        {    
            #这里面指定两个不同的数据库,虽然它填写的都是一样的
            "connections": {
                "first": {
                    "engine": "tortoise.backends.sqlite",
                    "credentials": {"file_path": "example.sqlite3"},
                },
                "second": {
                    "engine": "tortoise.backends.sqlite",
                    "credentials": {"file_path": "example1.sqlite3"},
                },
            },
            #可以指定多个不同的models, 嗯,它指定的也是一样的
            "apps": {
                "tournaments": {"models": ["__main__"], "default_connection": "first"},
                "events": {"models": ["__main__"], "default_connection": "second"},
            },
        }
    )
    #await Tortoise.generate_schemas()
    #选取数据库
    client = Tortoise.get_connection("first")
    #选取 models
    second_client = Tortoise.get_connection("second")
    #然后这下面就可以进行orm操作了

增删改查

from typing import List
from fastapi import APIRouter, HTTPException
from tortoise.contrib.fastapi import HTTPNotFoundError

from schemas.MaintainedData import MaintainedData_Pydantic, MaintainedDataIn_Pydantic
from models.MaintainedData import MaintainedData
from schemas.basic import Response404, Response200

data = APIRouter(tags=['数据模块'])


@data.get("/", summary="数据列表", response_model=List[MaintainedData_Pydantic])
async def data_list(order:str,limit: int = 10, page: int = 1):
    #  limit 是显示的条数 ,page是页数

    skip = (page - 1) * limit
    # select * from movie limit offset,limit
    # select * from movie limit 0,10
    return await MaintainedData_Pydantic.from_queryset(MaintainedData.all().offset(skip).limit(limit))


@data.post("/data", summary="新增数据", )
async def add_data(md_data: MaintainedData_Pydantic):
    movie_obj = await MaintainedData.create(**md_data.dict(exclude_unset=True))
    # movie_obj = await Movie.create(name="",year="",xx="")
    return await MaintainedData_Pydantic.from_tortoise_orm(movie_obj)


@data.put("/data/{data_id}", summary="编辑数据", responses={404: {"model": HTTPNotFoundError}})
async def update_data(site_id: str, md_data: MaintainedDataIn_Pydantic):
    updated_count = await MaintainedData.filter(site_id=site_id).update(**md_data.dict(exclude_unset=True))
    print(updated_count)
    if not updated_count:
        return Response404(msg=f"data {site_id} not found")
    return Response200(data=await MaintainedData_Pydantic.from_queryset_single(MaintainedData.get(site_id=site_id)))


@data.get("/data/{data_id}", summary="查找数据", response_model=MaintainedData_Pydantic,
          responses={404: {"model": HTTPNotFoundError}})
async def get_user(site_id: str):
    return await MaintainedData_Pydantic.from_queryset_single(MaintainedData.get(site_id=site_id))


@data.delete("/data/{data_id}", summary="删除数据", responses={404: {"model": HTTPNotFoundError}})
async def del_movie(site_id: str):
    deleted_count = await MaintainedData.filter(site_id=site_id).delete()
    if not deleted_count:
        raise HTTPException(status_code=404, detail=f"data {site_id} not found")
    return HTTPException(status_code=200, detail=f"Deleted Movie {site_id}")

models.py

from datetime import datetime
from typing import Optional, Iterable

from tortoise import fields, models, BaseDBAsyncClient


class MaintainedData(models.Model):
    """
    The User model
    """
    id = fields.SmallIntField(pk=True, index=True, description="ID", )
    site_id = fields.CharField(max_length=20, unique=True, description="网站ID")
    site_name = fields.CharField(max_length=128, null=False, description="网站名称")
    site_type = fields.CharField(max_length=16, null=False, description="网站类型")
    description = fields.CharField(max_length=255, default="",description="网站备注", )
    site_path_name = fields.CharField(max_length=128, null=True, description="目录名称")
    site_path_url = fields.CharField(max_length=128, null=True, description="目录链接")
    status = fields.CharField(max_length=128, null=True, description="脚本状态")
    run_computer = fields.CharField(max_length=255, null=True, description="运行电脑")
    run_directory = fields.CharField(max_length=255, null=True, description="运行目录")
    crawling_time :Optional[datetime]= fields.DatetimeField(auto_add=True, description="运行时间")
    err_message = fields.CharField(max_length=255, null=True, description="错误信息")

order by

db_Secc = MaintainedData.all().order_by("crawling_time")

Eum 枚举

class order_byEnum(str, Enum):
    pear = ""
    banana = '-'


@data.get("/", summary="数据列表", response_model=List[MaintainedData_Pydantic])
async def data_list(
        order_by: order_byEnum = "", 
        limit: int = 10, 
        page: int = 1):
    #  limit 是显示的条数 ,page是页数
    skip = (page - 1) * limit
    return await MaintainedData_Pydantic.from_queryset(MaintainedData.all().offset(
        skip).limit(
        limit).order_by(
        f"{order_by}crawling_time")) 

标签:tortoise,models,FastAPI,site,orm,import,data,id,MaintainedData
From: https://www.cnblogs.com/zichliang/p/16819406.html

相关文章

  • 【758】Transformer结构图
    参考:Transformer-Attention参考:TheTransformerFamilyFullArchitecture分解(FeedForward即是FullyConnected)Encoder部分Decoder部分Multi-HeadSelf-Attent......
  • 自定义transform的重塑
    #cartoontransformdefct(names,*args):'''names:alistofmobjectsargsr......
  • ABAP函数_2及Smart Forms_1
         ......
  • Django-ORM常用字段及参数
     目录-常用字段--内容--自定义字段-字段参数--普通---null---unique---db_index---default--DateField和DateTimeField---auto_now_add---auto_now--关系......
  • Django-ORM
    目录-定义-使用方法--models.py--数据库迁移命令 --注意-实际操作--ORM实现字段的增删改查--ORM实现数据的增删改查--ORM创建表与表之间的联系-定义ORM:对......
  • FastApi定时任务发送钉钉消息
    fastapi是一个高性能的异步框架,实现 定时任务 需要 task.py 提供一个装饰器#task.py#网上百度的,地址:https://blog.csdn.net/hekaiyou/article/details/125072249......
  • 基于FastApi的微信公众号开发
    个人申请的订阅号,未认证,可用功能可在 微信公众号平台 - 接口权限 处查看使用代码开发,首先需要在 基础配置-服务器配置 中进行设置 填写服务器地址(URL)、Token和Enc......
  • element-ui form表单的input框按回车键后提交表单的问题
    在项目中,经常有表单提交操作,当提交的表单只有一行且均有必填校验时,按回车键会执行form的submit,但这个submit并不一定是我们想要的,因此要想办法避免此问题避免方法:通过在f......
  • C# 输入法切换 C#中FormClosing与FormClosed的区别详细解析
    引用///<summary>///输入法切ENG///</summary>publicvoidChangeENG(){WinAPI.PostMessage(WinAPI.HWND_BROADC......
  • Matlab_normpdf
    >>x=normrnd(15,5,1000,1);>>hist(x)>>x2=rand(1000,1);>>hist(x2,50)>>hist(x,50)>>y=normrnd(15,1,1000,1);>>figure;hist(y,50)>>x=0:30;>>pdf=normp......