同步干完一件之后才能再干下一件事。遇到程序需要等待时程序卡死不动异步当遇到需要等待的事件 异步先去干其他的事件别的事情干完之后再后来干这个事件同步浪费CPU等待时间异步最大优化空闲时间如果用同步代码有人发起请求接口、查询数据库程序就卡死等待新来的请求全部排队 稍微多个请求就会卡顿、超时等问题异步的优势就是单线程就你能处理大量并发请求 不用频繁占用系统资源提升效率Tortoise-ORM集成ORM有以下特点效率高不用手动基础语句 复杂查询可以封装 不用手动拼接语句语法跟友好用python 就能操控数据库不用专门学习复杂的sql语句 对开发者等加友好Tortoise-ORM集成在做异步python 时候 使用ORM有时会卡死 整个程序运行不了 不符合异步开发的理念 而Tortoise-ORM刚好适合异步开发有以下特点原生异步:基于asyncio构建所有数据库操作都是异步的不会阻塞事件循环。Django风格的API:如果你用过Django ORM会发现语法非常熟悉上手极快。轻量级:专注于核心功能没有冗余的抽象层性能出色。完整的关系支持:支持外键、一对一、多对多等所有常见关系类型内置Pydantic 集成:可以通过 pydantic_model_creator 自动从 Tortoise 模型生成 Pydantic模型无缝对接FastAPI的请求/响应校验和传统同步ORM的区别Fastapi的用法getuser_router.get(/all) async def getAllUsers(): dataawait User.all() return {code:1,message:获取所有用户成功,data:data }user_router.get(/all) 括号里面可以自己定义变量名后面可以传值比如 id:int,name:ste 等user_router.get(/all/{id}/{name}) async def getAllUsers(id:int,name:str): dataawait User.all() return {code:1,message:获取所有用户成功,data:data }postuser_router.post(/save,summary保存用户) async def save_user(user: UserCreateRequest): user1await User.create( usernameuser.username, email user.email , phone user.phone, is_admin user.is_admin ) # await User.create(**dict(user)) return {code:1,message:保存用户成功,data:user1}如果是/{name}类型的 那么方法中传的数据变量名要与路径中的变量名相等url……/hello/1/ 路由就相等这个样子 path 可以为变量加约束 如class UserCreateRequest(BaseModel): username: strField( ..., min_length3, max_length50, description用户名, exampleadmin123 ) email: strField( ..., min_length5, max_length100, description邮箱, exampleadmintest.com ) phone: strField( None, min_length11, max_length11, description手机号, example13800138000 ) is_admin: boolField( False, description是否是管理员, exampleFalse )这是添加商品设置字段的的格式和校验下面就是添加的请求体 基本都是必填的 也可以为选填putuser_router.put(/update/{id},summary更新用户) async def update_user(id:int,user: UserCreate): user1 await User.get_or_none(idid) if not user1: return {code:0,message:用户不存在} # 1 # if user.username is not None: # user1.usernameuser.username # if user.email is not None: # user1.emailuser.email # if user.phone is not None: # user1.phoneuser.phone # if user.is_admin is not None: # user1.is_adminuser.is_admin # await user1.save() # 2 # 只把有值的字段变成字典排除掉未提供实际值的字段 # user_dictuser.dict(exclude_unsetTrue) # for key, value in user_dict.items(): # setattr(user1, key, value) # 使用 setattr() 方法动态设置属性 # await user1.save() # 3 user_dictuser.dict(exclude_unsetTrue) await User.filter(idid).update(**user_dict) return {code:1,message:修改成功,data:}修改内容有三个方法方法一user1 await User.get_or_none(idid) if not user1: return {code:0,message:用户不存在} # 1 # if user.username is not None: # user1.usernameuser.username # if user.email is not None: # user1.emailuser.email # if user.phone is not None: # user1.phoneuser.phone # if user.is_admin is not None: # user1.is_adminuser.is_admin # await user1.save() return {code:1,message:修改成功}方法不叫简单但是比较繁琐方法二user_router.put(/update/{id},summary更新用户) async def update_user(id:int,user: UserCreate): user1 await User.get_or_none(idid) if not user1: return {code:0,message:用户不存在} # 只把有值的字段变成字典排除掉未提供实际值的字段 user_dictuser.dict(exclude_unsetTrue) for key, value in user_dict.items(): setattr(user1, key, value) # 使用 setattr() 方法动态设置属性 await user1.save() return {code:1,message:修改成功}只把有值的字段变成字典排除掉未提供实际值的字段方法三user_router.put(/update/{id},summary更新用户) async def update_user(id:int,user: UserCreate): user_dictuser.dict(exclude_unsetTrue) await User.filter(idid).update(**user_dict) return {code:1,message:修改成功}这一块是要修改的内容deleteuser_router.delete(/delete/{id},summary删除用户) async def delete_user(id:int): # 1 # user1 await User.get_or_none(idid) # if user1 is None: # return {code:0,message:用户不存在} # await user1.delete() # 2 await User.filter(idid).delete() return {code:1,message:删除成功}在task中查找id 然后直接调用方法 delete就可以删除了 前面要加一个await挂载才可以使用
