大体来自黑马程序员加部分了解笔记

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
      start=time.time()
      for i in range (10):
          time.sleep(1)
      end=time.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():
    start=time.time()
    tasks=[asyncio.sleep(1) for i in range () 10]
    await asyncio.gather(*tasks)
    end=time.time()
    return {"time":f'{end-start:.2f}s'}

自动生成可交互式文档，浏览器中直接调用和测试API，访问方法：在项目跑起来以后的网页后面直接加/docs

FastAPI基础入门

使用FastAPI框架搭建Web服务

1.搭建项目

pycharm中自带FastAPI框架，点击右下角create自动下载FastAPI所需的包。

from fastapi import FastAPI

#创建FastAPI实例
app=FastAPI()

@app.get("/)
async def root():
    return {"message":"Hello World"}

@app.get("/hello/{name}")
async def say_hello(name:str):
    return {"message":f"Hello {name}"}

2.运行项目

命令行运行文件方式：

uvicorn 文件名:实例名 --reload

        |                                    |

(高性能服务器)          （修改代码后自动重启）（热部署，热启动） 

路由

定义：路由就是URL地址和处理函数之间的映射关系，它决定了当用户访问某个特定网址时，服务器应该执行哪段代码来返回结果

  FastAPI的路由定义基于python的装饰器

路由=装饰器＋函数

#访问/hello 响应结果 msg:你好 FastAPI
@app.get("/hello")
async def get_hello():
    return {"msg":"你好 FastAPI"}

参数简介和路径参数

同一段接口逻辑，根据参数返回不同数据，不同书名返回不同的书，类似多态

参数分类

参数相当于函数的参数，之前是无参函数，现在是有参函数 

路径参数（找哪一个）

位置：URL路径的一部分 /book/{id}

作用：指向唯一的，特定的资源

方法：GET

@app.get("/book/{id}")
async def get_book(id:int):
    return {"id":id,"title":f"这是第{id}本书"}

类型注解PATH  

FastAPI允许为参数声明额外的信息和校验

@app.get("/book/{id}")
async def get_book(id:int=Path(...,gt=0,lt=101)):#限制id为1~100
    return {"id":id,"title":f"这是第{id}本书"}

查询参数(怎么选，排序分页)

位置：URL?之后k1=v1&k2=v2

作用：对资源集合进行过滤，排序，分页等操作

方法：GET

声明的参数不是路径参数时，路径操作函数会把该参数自动解释为查询参数

类型注解Query

#from fastapi import FastAPI,Query
#进行引入查询函数的限制函数
#需求 查询新闻->分页，skip:跳过的记录数， limit:返回的记录数 10
@app.get("/news/new_list")
#这里就是同时限制和有默认值，...意味无默认值
async def get_news_list(skip:int=Query（0,descirption="跳过的记录数"),limit:int=10)
    return {"skip":skip,"list":list}

请求体（给服务器什么）

位置：HTTP请求的消息体（body）中

作用：创建，更新资源，携带大量数据，如：JSON

方法：POST,PUT等

在HTTP协议中，一个完整的请求由三部分组成

1.请求行：包含方法，URL，协议版本

2.请求头:元数据信息

3.请求体:实际要发送的数据内容 

#1.定义类型
from pydantic import BaseModel，Field
class User(BaseModel)
    username:str=Field(...)
    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_model=News)
async def get_news(id:int):
    return{
        "id":id,
        "title":f"这是第{id}本书"，
        "content":"这是一本好书"
}

响应类型设置方式

1.装饰器中指定相应类

场景：固定返回类型（HTML，纯文本等）response_class=

from fastapi.reponses import HTMLResponse
@app.get("html",reponse_class=HTMLReponse)
async def get_html():
    return <h1>这是标题</h1>

2.返回响应对象

场景：文件下载，图片，流式响应

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(status=404,detail="当前id不存在")
    return {"id":id}

FastAPI进阶

中间件

使用中间件为每个请求前后添加统一的处理逻辑，作用包含日志记录，身份认证，跨域处理，响应头处理，性能监控。（少写重复代码）

定义：中间件是一个在每次请求进入FastAPI应用时都会被执行的函数，他在请求到达实际的路径操作之前运行，并且在响应返回给客户端前再运行一次 

函数的顶部使用装饰器@app.middleware("http")

中间件就是这个装饰器加一个带有两个参数的函数，request:请求，call_next:传递请求给路径处理函数。非固定名称

@app.middleware("http")
async def middleware(request,call_next):
    print('中间件开始处理--start')
    response=await call_next(request)
    print('中间件处理完成--end')
    return response

执行顺序像洋葱一样切开，先写的在里面

依赖注入系统

和中间件一样封装代码，但是中间件控制所有代码，依赖注入只用在部分之中

依赖项：可重复的组件，负责提供某种功能或者数据

1.写依赖项

2.引入Depends

3.注入依赖项

from fastapi import FastAPI,Query,Depends
#分页参数逻辑共用：新闻列表和用户列表

app=FastAPI()
async def common_parameters(skip:int=Query(0,ge=0),limit:int=Query(10,le=60)):
    return {"skip":skip,"limit":limit}
@app.get("/news/news_list")
async def get_news_list(commons=Depends(common_parameters)):
    return commons

ORM工具

ORM是一种编程技术，用于面向对象编程语言和关系型数据库之间建立映射，可以通过操作对象的方式与数据库进行交互

减少重复的SQL代码，自动处理数据库连接和事务，自动防止SQL注入攻击

1.安装

记得安装mysql

pip install sqlalchemy[asyncio]

aiomysql(异步数据库驱动)

2.建库建表

1.创建数据库引擎

使用create_async_engine创建异步引擎

from sqlalchemy.ext.asyncio import create_async_engine
#用户名用自己数据库的root:123456@localhost:3306此为用户名加密码
ASYNC_DATABASE_URL="mysql+aiomysql://root:123456@localhost:3306/fastapi_test?charset=utf8"

#创建异步引擎
async_engine=create_async_engine(
    ASYNC_DATABASE_URL
    echo=True#可选：输出SQL日志
    pool_size=10#设置连接池中保持的持久连接数
    max_overflow=20#设置连接池允许创建的额外连接数
)

2.定义模型类

基类，继承DeclarativeBase(包含通用属性和字段的映射)

定义数据库表对应的模型类

from sqlalchemy.orm import DeclarativeBase,Mapped,mapped_column 
class Base(DeclarativeBase):#基类
    create_time:Mapped[datetime]=mapped_column(
        DateTime,insert_default=func.now(),default=datetime.now,comment="创建时间"
)
    update_time=Mapped[datetime]=mapped_column(
        DateTime,insert_default=func.now(),default=datetime.now,onupdate=func.now(),comment="修改时间")

class Book(Base):#表对应的的模型类
    __tablename__="book"

    id:Mapped[int]=mapped_column(primary_key=True)
    bookname:Mapped[str]=mapped_column(String(255))
    author:Mapped[str]=mapped_column(String(255))

• Mapped: 类型注解，表示该属性映射到数据库列。
• datetime: Python 的 datetime 类型，对应数据库的 DateTime 类型
• DateTime: 数据库列类型，存储日期时间。
• insert_default=func.now(): 插入记录时自动设置为数据库当前时间（通过 SQL 函数 NOW()）。
• default=datetime.now: Python 端的默认值（备用，通常优先用 insert_default）。
• comment: 列的注释，用于文档或数据库元数据。
• Mapped[int]：类型注解，表示该属性映射到数据库列（int 对应 Integer 类型）。
• mapped_column：定义列，参数包括：
  • 数据库类型（如 Integer、String）。
  • 约束（如 primary_key=True、nullable=False）。

  • 参数	说明
    primary_key	是否主键（True/False）
    nullable	是否允许 NULL（默认 True）
    unique	是否唯一约束
    index	是否创建索引
    default	Python 端默认值（插入时生效）
    insert_default	数据库端默认值（如 func.now()）
    onupdate	更新时自动设置的值（如 func.now() 用于 update_time）
    comment	列注释（用于数据库元数据）

“是否主键” 是数据库表设计中的一个核心概念，用于标识某一列（或一组列）是否作为表的唯一标识符。主键（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(lifespan=lifespan)

3.操作数据

路由匹配中使用ORM

核心：创建依赖项，使用Depends注入到处理函数

#创建异步会话工厂
AsyncSessionLocal=async_sessionmaker(
    bind=async_engine,#绑定数据类引擎
    class_=AsyncSession,#指定会话类
    expire_on_commit=False#会话对象不过期，不重新查询数据库
)
#依赖项，用于获取数据库会话
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:AsyncSession=Depends(get_database)#声明参数 db 的类型为 AsyncSession
):
    result=await db.execute(select(Book))#Book模型类
    user=result.scalars().all()
    return user

• select(Book)

  • SQLAlchemy 的查询构造器，生成异步查询语句。

• await db.execute()

  • 执行异步 SQL 查询的关键方法。

• result.scalars().all()

  • 从查询结果中提取所有模型对象（如 Book 实例）。

• session.commit() 是在使用 SQLAlchemy（Python 最常用的 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:AsyncSession=Depends(get_database)#声明参数 db 的类型为 AsyncSession
):
    result=await db.execute(select(Book).where(Book.id==book.id))#Book模型类
    book=result.scalar_one_or_none()#满足就写，没有就返回None
    return book

模糊查询：like()%零个，一个或多个字符,_一个单个字符

下代码作用：寻找以曹开头的作者，不管后面几个字，改为_则只能查出曹x，只有一个字符

@app.get("/book/search_book")
async def get_search_book(
    db:AsyncSession=Depends(get_database)#声明参数 db 的类型为 AsyncSession
):
    result=await db.execute(select(Book).where(Book.author.like("曹%"))
    book=result.scalars().all()
    return book

与非查询:&;|;~取反

查询姓曹且价格大于一百的，注意括号数量。

@app.get("/book/search_book")
async def get_search_book(
    db:AsyncSession=Depends(get_database)#声明参数 db 的类型为 AsyncSession
):
    result=await db.execute(select(Book).where((Book.author.like("曹%")) & (Book.price>100)))
    book=result.scalars().all()
    return book

包含查询：in_()

注意下划线和括号数量

只要在id为列表里的书就都要查询

@app.get("/book/search_book")
async def get_search_book(
    db:AsyncSession=Depends(get_database)#声明参数 db 的类型为 AsyncSession
):
    id_list=[1,3,5,7]
    result=await db.execute(select(Book).where(Book.id.in_(id_list)))
    book=result.scalars().all()
    return book

所以查询的核心即为

语法糖确定网址后面的部分为什么

异步创建类似函数，但是参数包含数据库的类型并且通过Depends()注入

然后依旧异步的一个结果=await db.excute()将取出的orm传递，里面是一个orm语句

返回结果

聚合查询

聚合计算：func.方法（模型类.属性）

@app.get("/book/count")
async def get_count(
    db:AsyncSession=Depends(get_database)#声明参数 db 的类型为 AsyncSession
):
    result=await db.execute(select(func.avg(Book.price)))#Book模型类
    num=result.scalar()#提取一个数值->标量值
    return num

分页查询

select().offset().limit()

offset:跳过的记录数（之前页码的数据总和）offset=(current-1)*limit

limit:返回的记录数(每页的数据量)

@app.get("/book/get_books_list")
async def get_book_list(page:int=1,page_size:int=3,db:AsyncSession=Depends(get_database)):
    skip=(page-1)*page_size
    stmt=select(Book).offset(skip).limit(page_size)
    result=await db.execute(stmt)
    books=result.scalars().all()
    return {"books":books}

新增

用户输入信息新增到数据库

定义ORM对象->add(对象)->commit提交到数据库

此处是post

class BookBase(BaseModel):
    id:int
    bookname:str
    author:str
    price:float
    publisher:str

@app.post("/book/add_book")
async def add_book(book:BookBase,db:AsyncSession=Depends(get_database)):
    ab.add(Book(**book.__dict__))#将其转为字典再展开运算
    await db.commit()
    return book

book 是请求体参数，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:AsyncSession=Depends(get_database)):
    book=await db.get(Book,book_id)
    if book is None:
        raise HTTPExcepetion(status_code=404,detail="Book not found")
    book.bookname=data.bookname
    book.author=data.author
    book.price=data.price

    await db.commit
    return book

get() 是 AsyncSession 提供的一个便捷方法，用于 通过主键直接获取一条记录。

• 第一个参数：模型类（Book）

• 第二个参数：主键值（book_id，比如 1）

删除

delete

 查询get，delete删除，commit提交到数据库

@app.delete("/book/delete_book/{book_id}")
async def delete_book(book_id:int,db:AsyncSession=Depends(get_database)):
    db_book=await db.get(Book,book_id)
    if db_book is None:
        raise HTTPException(status_code=404,detail="Book not found")
    await db.delete(db_book)
    await db.commit()
    return {"message":"Book deleted"}