【FastAPI】

大体来自黑马程序员加部分了解笔记FASTAPI特点优点同步定义任务按顺序逐个执行当前任务必须完成后才能开始下一个任务。特点阻塞Blocking执行到耗时操作如网络请求、文件读写时程序会暂停“卡住”直到操作完成。顺序执行代码按书写顺序一步步执行逻辑简单直观。单线程效率低如果任务中有 I/O 操作如等待数据库返回数据线程会空闲浪费资源。# 使用FastAPI框架的装饰器定义一个GET请求的端点路径为/sync #装饰器Decorator 是Python中一种特殊的语法它允许你在不修改原函数代码的情况下为函数添加新的功能。可以把它想象成一个“包装盒”把函数放进去函数就多了些新特性。 #add_stars # 等价于 hello add_stars(hello)语法糖 app.get(/sync) #app.get(/sync)的作用总结 #注册路由告诉FastAPI当收到GET /sync请求时调用下面的函数 #定义接口创建了一个可以通过HTTP访问的API端点 #生成文档自动为这个端点生成API文档 #处理请求将HTTP请求转换为Python函数调用 def func_sync(): # 调用time.time()函数获取当前时间戳浮点数单位秒赋值给变量start starttime.time() for i in range (10): time.sleep(1) endtime.time() # 返回一个字典Python中的键值对集合 # 字典的键是字符串time值是一个格式化字符串 # f{end-start:.2f}s 是一个f-string格式化字符串字面量 # end-start 计算时间差浮点数 # :.2f 表示将浮点数格式化为保留两位小数 # s 是字面量字符s表示秒 return {time:f{end-start:.2f}s}异步定义任务可以并发执行当前任务在等待耗时操作时会主动让出控制权让其他任务运行等耗时操作完成后再次恢复执行。特点非阻塞Non-blocking遇到 I/O 操作时线程不会被挂起而是继续执行其他任务。并发执行通过事件循环Event Loop调度多个任务提高资源利用率。需要协程Coroutine支持Python 中通过async/await实现。app.get(/async) async def func_async(): starttime.time() tasks[asyncio.sleep(1) for i in range () 10] await asyncio.gather(*tasks) endtime.time() return {time:f{end-start:.2f}s}自动生成可交互式文档浏览器中直接调用和测试API访问方法在项目跑起来以后的网页后面直接加/docsFastAPI基础入门使用FastAPI框架搭建Web服务1.搭建项目pycharm中自带FastAPI框架点击右下角create自动下载FastAPI所需的包。from fastapi import FastAPI #创建FastAPI实例 appFastAPI() app.get(/) async def root(): return {message:Hello World} app.get(/hello/{name}) async def say_hello(name:str): return {message:fHello {name}}2.运行项目命令行运行文件方式uvicorn 文件名:实例名 --reload| |(高性能服务器) 修改代码后自动重启热部署热启动路由定义路由就是URL地址和处理函数之间的映射关系它决定了当用户访问某个特定网址时服务器应该执行哪段代码来返回结果FastAPI的路由定义基于python的装饰器路由装饰器函数#访问/hello 响应结果 msg:你好 FastAPI app.get(/hello) async def get_hello(): return {msg:你好 FastAPI}参数简介和路径参数同一段接口逻辑根据参数返回不同数据不同书名返回不同的书类似多态参数分类参数相当于函数的参数之前是无参函数现在是有参函数路径参数找哪一个位置URL路径的一部分 /book/{id}作用指向唯一的特定的资源方法GETapp.get(/book/{id}) async def get_book(id:int): return {id:id,title:f这是第{id}本书}类型注解PATHFastAPI允许为参数声明额外的信息和校验app.get(/book/{id}) async def get_book(id:intPath(...,gt0,lt101)):#限制id为1~100 return {id:id,title:f这是第{id}本书}查询参数(怎么选排序分页)位置URL?之后k1v1k2v2作用对资源集合进行过滤排序分页等操作方法GET声明的参数不是路径参数时路径操作函数会把该参数自动解释为查询参数类型注解Query#from fastapi import FastAPI,Query #进行引入查询函数的限制函数 #需求 查询新闻-分页skip:跳过的记录数 limit:返回的记录数 10 app.get(/news/new_list) #这里就是同时限制和有默认值...意味无默认值 async def get_news_list(skip:intQuery0,descirption跳过的记录数),limit:int10) return {skip:skip,list:list}请求体给服务器什么位置HTTP请求的消息体body中作用创建更新资源携带大量数据如JSON方法POST,PUT等在HTTP协议中一个完整的请求由三部分组成1.请求行包含方法URL协议版本2.请求头:元数据信息3.请求体:实际要发送的数据内容#1.定义类型 from pydantic import BaseModelField class User(BaseModel) username:strField(...) password:str #2.类型注解 app.post(/register) async def register(user:User): return user浏览器请求体-d后类型注解Field这三个参数发挥不同的作用都可以进行python原生类型注解和函数类型注解可以用来对前端的数据类型进行检测。请求与响应响应类型JSON,HTML文件自定义等默认情况下FastAPI会自动将路径操作返回的Python对象经由jsonable_encoder转换为JSON兼容格式如果需要返回其他的也有丰富的响应类型来返回不同数据。自定义响应数据格式response_model是路径操作装饰器如app.get的关键参数定义和约束API端点的输出格式from pydantic import BaseModel class News(BaseModel): id:int title:str content:str app.get(/news/{id},response_modelNews) async def get_news(id:int): return{ id:id, title:f这是第{id}本书 content:这是一本好书 }响应类型设置方式1.装饰器中指定相应类场景固定返回类型HTML纯文本等response_classfrom fastapi.reponses import HTMLResponse app.get(html,reponse_classHTMLReponse) async def get_html(): return h1这是标题/h12.返回响应对象场景文件下载图片流式响应FIleResponse是FastAPI一共的专门高效返回文件内容的响应类能够智能处理文件路径媒体类型推断范围请求和缓存头部app.get(/file) async def get_file(): file_path./files/1.jpeg return FileResponse(file_path)异常处理对于客户端引发的错误应用fast.apiHTTPException来中断正常处理流程并返回标准错误响应from fastapi import FastAPI,HTTPException app.get(/news/{id}) async def get_news(id:int): id_list[1,2,3,4,5,6] if id not in id_list: raise HTTPException(status404,detail当前id不存在) return {id:id}FastAPI进阶中间件使用中间件为每个请求前后添加统一的处理逻辑作用包含日志记录身份认证跨域处理响应头处理性能监控。少写重复代码定义中间件是一个在每次请求进入FastAPI应用时都会被执行的函数他在请求到达实际的路径操作之前运行并且在响应返回给客户端前再运行一次函数的顶部使用装饰器app.middleware(http)中间件就是这个装饰器加一个带有两个参数的函数request:请求call_next:传递请求给路径处理函数。非固定名称app.middleware(http) async def middleware(request,call_next): print(中间件开始处理--start) responseawait call_next(request) print(中间件处理完成--end) return response执行顺序像洋葱一样切开先写的在里面依赖注入系统和中间件一样封装代码但是中间件控制所有代码依赖注入只用在部分之中依赖项可重复的组件负责提供某种功能或者数据1.写依赖项2.引入Depends3.注入依赖项from fastapi import FastAPI,Query,Depends #分页参数逻辑共用新闻列表和用户列表 appFastAPI() async def common_parameters(skip:intQuery(0,ge0),limit:intQuery(10,le60)): return {skip:skip,limit:limit} app.get(/news/news_list) async def get_news_list(commonsDepends(common_parameters)): return commonsORM工具ORM是一种编程技术用于面向对象编程语言和关系型数据库之间建立映射可以通过操作对象的方式与数据库进行交互减少重复的SQL代码自动处理数据库连接和事务自动防止SQL注入攻击1.安装记得安装mysqlpip install sqlalchemy[asyncio]aiomysql(异步数据库驱动)2.建库建表1.创建数据库引擎使用create_async_engine创建异步引擎from sqlalchemy.ext.asyncio import create_async_engine #用户名用自己数据库的root:123456localhost:3306此为用户名加密码 ASYNC_DATABASE_URLmysqlaiomysql://root:123456localhost:3306/fastapi_test?charsetutf8 #创建异步引擎 async_enginecreate_async_engine( ASYNC_DATABASE_URL echoTrue#可选输出SQL日志 pool_size10#设置连接池中保持的持久连接数 max_overflow20#设置连接池允许创建的额外连接数 )2.定义模型类基类继承DeclarativeBase(包含通用属性和字段的映射)定义数据库表对应的模型类from sqlalchemy.orm import DeclarativeBase,Mapped,mapped_column class Base(DeclarativeBase):#基类 create_time:Mapped[datetime]mapped_column( DateTime,insert_defaultfunc.now(),defaultdatetime.now,comment创建时间 ) update_timeMapped[datetime]mapped_column( DateTime,insert_defaultfunc.now(),defaultdatetime.now,onupdatefunc.now(),comment修改时间) class Book(Base):#表对应的的模型类 __tablename__book id:Mapped[int]mapped_column(primary_keyTrue) bookname:Mapped[str]mapped_column(String(255)) author:Mapped[str]mapped_column(String(255))Mapped: 类型注解表示该属性映射到数据库列。datetime: Python 的datetime类型对应数据库的DateTime类型DateTime: 数据库列类型存储日期时间。insert_defaultfunc.now(): 插入记录时自动设置为数据库当前时间通过 SQL 函数NOW()。defaultdatetime.now: Python 端的默认值备用通常优先用insert_default。comment: 列的注释用于文档或数据库元数据。Mapped[int]类型注解表示该属性映射到数据库列int对应Integer类型。mapped_column定义列参数包括数据库类型如Integer、String。约束如primary_keyTrue、nullableFalse。参数说明primary_key是否主键True/Falsenullable是否允许NULL默认Trueunique是否唯一约束index是否创建索引defaultPython 端默认值插入时生效insert_default数据库端默认值如func.now()onupdate更新时自动设置的值如func.now()用于update_timecomment列注释用于数据库元数据“是否主键”是数据库表设计中的一个核心概念用于标识某一列或一组列是否作为表的唯一标识符。主键Primary Key是数据库中用于唯一区分每一行记录的约束条件3.启动应用时建表视频中貌似是重新启动就重新建表造成数据丢失第二种Lifespan类可使用async def create_tables(): async with async_engine.begin() as conn: #async with: 确保连接在退出代码块时自动提交成功或回滚异常。 await conn.run_sync(Base.metadata.create_all) #run_sync(): 异步驱动提供的适配器用于在异步上下文中运行同步代码。 #Base.metadata.create_all: SQLAlchemy 的同步方法根据模型定义创建数据库表。 app.on_event(startup)#数据库启动时 async def startup_event(): await create_tables()from fastapi import FastAPI from contextlib import asynccontextmanager asynccontextmanager async def lifespan(app: FastAPI): # 启动逻辑 print(应用启动中...) yield # 关键点在此处暂停等待应用运行 # 关闭逻辑 print(应用关闭中...) app FastAPI(lifespanlifespan)3.操作数据路由匹配中使用ORM核心创建依赖项使用Depends注入到处理函数#创建异步会话工厂 AsyncSessionLocalasync_sessionmaker( bindasync_engine,#绑定数据类引擎 class_AsyncSession,#指定会话类 expire_on_commitFalse#会话对象不过期不重新查询数据库 ) #依赖项用于获取数据库会话 async def get_database(): async with AsyncSessionLocal() as session try: yield session#返回数据库会话给路由处理函数 await session.commit()#无异常提交事务 except Exception: await session.rollback()#有异常则回滚 raise finally: await session.close()#关闭会话 app.get(/book/books) async def get_book_list( db:AsyncSessionDepends(get_database)#声明参数 db 的类型为 AsyncSession ): resultawait db.execute(select(Book))#Book模型类 userresult.scalars().all() return userselect(Book)SQLAlchemy 的查询构造器生成异步查询语句。await db.execute()执行异步 SQL 查询的关键方法。result.scalars().all()从查询结果中提取所有模型对象如Book实例。session.commit()是在使用SQLAlchemyPython 最常用的 ORM 框架时用来提交当前事务的方法。简单来说它负责把你对数据库所做的修改真正保存到数据库中注意这里depends和select需要引入第一个是fastapi的第二个是sqlalchemy的然后运行查看数据库时打开右侧圆柱形图标怕选择新建数据源选择mysql注意这里会显示未下载直接下载就行下载的和mysql连接的插件而非mysql本身。查询查询数据核心语句await db.execute(select(模型类))返回一个ORM对象获取所有数据scalars.all()获取单条数据scalars().first()get(模型类主键值)-直接基于模型类不需要取出的语句根据主键查询条件select(Book).where(条件条件2...)比较判断:;;app.get(/book/{book_id}) async def get_book_list( db:AsyncSessionDepends(get_database)#声明参数 db 的类型为 AsyncSession ): resultawait db.execute(select(Book).where(Book.idbook.id))#Book模型类 bookresult.scalar_one_or_none()#满足就写没有就返回None return book模糊查询like()%零个一个或多个字符,_一个单个字符下代码作用寻找以曹开头的作者不管后面几个字改为_则只能查出曹x只有一个字符app.get(/book/search_book) async def get_search_book( db:AsyncSessionDepends(get_database)#声明参数 db 的类型为 AsyncSession ): resultawait db.execute(select(Book).where(Book.author.like(曹%)) bookresult.scalars().all() return book与非查询:;|;~取反查询姓曹且价格大于一百的注意括号数量。app.get(/book/search_book) async def get_search_book( db:AsyncSessionDepends(get_database)#声明参数 db 的类型为 AsyncSession ): resultawait db.execute(select(Book).where((Book.author.like(曹%)) (Book.price100))) bookresult.scalars().all() return book包含查询in_()注意下划线和括号数量只要在id为列表里的书就都要查询app.get(/book/search_book) async def get_search_book( db:AsyncSessionDepends(get_database)#声明参数 db 的类型为 AsyncSession ): id_list[1,3,5,7] resultawait db.execute(select(Book).where(Book.id.in_(id_list))) bookresult.scalars().all() return book所以查询的核心即为语法糖确定网址后面的部分为什么异步创建类似函数但是参数包含数据库的类型并且通过Depends()注入然后依旧异步的一个结果await db.excute()将取出的orm传递里面是一个orm语句返回结果聚合查询聚合计算func.方法模型类.属性app.get(/book/count) async def get_count( db:AsyncSessionDepends(get_database)#声明参数 db 的类型为 AsyncSession ): resultawait db.execute(select(func.avg(Book.price)))#Book模型类 numresult.scalar()#提取一个数值-标量值 return num分页查询select().offset().limit()offset:跳过的记录数之前页码的数据总和offset(current-1)*limitlimit:返回的记录数(每页的数据量)app.get(/book/get_books_list) async def get_book_list(page:int1,page_size:int3,db:AsyncSessionDepends(get_database)): skip(page-1)*page_size stmtselect(Book).offset(skip).limit(page_size) resultawait db.execute(stmt) booksresult.scalars().all() return {books:books}新增用户输入信息新增到数据库定义ORM对象-add(对象)-commit提交到数据库此处是postclass BookBase(BaseModel): id:int bookname:str author:str price:float publisher:str app.post(/book/add_book) async def add_book(book:BookBase,db:AsyncSessionDepends(get_database)): ab.add(Book(**book.__dict__))#将其转为字典再展开运算 await db.commit() return bookbook是请求体参数BookBase是一个 Pydantic 模型FastAPI 会自动从请求体中解析 JSON 并验证为BookBase实例。fastapi从 HTTP 请求的请求体Request Body中自动提取并按照BookBase这个 Pydantic 模型进行解析和校验。感觉好像git add. git commit更新先查再改重新赋值查询get属性重新赋值commit提交到数据库class BookUpdate(BaseModel): bookname:str author:str price:float publisher:str app.put(“/book/update_book/{book_id}”) async def update_book(book_id:int,data:BookUpdate,db:AsyncSessionDepends(get_database)): bookawait db.get(Book,book_id) if book is None: raise HTTPExcepetion(status_code404,detailBook not found) book.booknamedata.bookname book.authordata.author book.pricedata.price await db.commit return bookget()是AsyncSession提供的一个便捷方法用于 通过主键直接获取一条记录。第一个参数模型类Book第二个参数主键值book_id比如1删除delete查询getdelete删除commit提交到数据库app.delete(/book/delete_book/{book_id}) async def delete_book(book_id:int,db:AsyncSessionDepends(get_database)): db_bookawait db.get(Book,book_id) if db_book is None: raise HTTPException(status_code404,detailBook not found) await db.delete(db_book) await db.commit() return {message:Book deleted}