FastAPI基础入门

02.FASTAPI框架简介

pydantic库校验数据是否合法，做类型转换，减少手动校验代码

可交互式文档

框架特点

        异步性能高/开发效率高/自动生成文档

03.第一个FastAPI程序

vscode如何书写第一个FastAPI程序

在项目目录下新建main.py，内容为：

from fastapi import FastAPI

#创建fastapi实例
app = FastAPI()

#这是一个 Python 装饰器（Decorator）。它告诉 FastAPI：它正下方的函数负责处理发往路径 /（即根路径）的 GET 请求。
@app.get("/")
async def root():  #这是与上方装饰器绑定的函数。当用户通过 GET 方法访问 / 路径时，FastAPI 就会调用这个 root 函数。
    return {"message": "Hello FastAPI"}

#这个装饰器定义了一个带有路径参数的路由。{name} 是一个占位符，表示 URL 中这一部分是一个动态变量。例如，当访问 /hello/Alice 时，name 的值就是 Alice。
@app.get("/hello/{name}")
async def say_hello(name: str):
    return {"message": f"Hello, {name}"}

随后运行项目，激活环境

然后在终端运行：

uvicorn main:app --reload
#这个命令是 Python 异步 Web 框架（如 FastAPI、Starlette）在开发阶段最常用的启动命令。它的作用是使用 Uvicorn 服务器来运行你的应用，并开启代码热重载功能。
#uvicorn：这是 ASGI（异步服务器网关接口）服务器的命令。你可以把它理解为 Web 服务器（就像 Nginx 或 Apache，但它是专门为 Python 异步应用设计的）。它的核心作用是监听网络端口，接收客户端的 HTTP 请求，并将其转发给你的 Python 代码，最后把代码的返回值响应给客户端。
#main:app：这是告诉 Uvicorn 去哪里找到你的应用实例，格式为 <模块名>:<应用实例名>：
###main：代表你的 Python 文件名。假设你的代码保存在 main.py 文件中，这里就写 main（不需要带 .py 后缀）。
###app：代表你在 main.py 文件中创建的 FastAPI 实例的变量名。正如我们之前讨论的，代码中通常写为 app = FastAPI()，所以这里就是 app。
#--reload：这是一个热重载（Hot Reload）参数，专为开发环境设计，表示修改了代码后会自动重启服务器。
###作用：当 Uvicorn 运行时，它会持续监控你的项目文件。一旦你修改了代码并保存，Uvicorn 会自动检测到变化，并自动重启服务器。
###好处：这极大地提升了开发效率。你不需要每次改完代码都手动停止服务器（Ctrl+C）再重新启动，只需刷新浏览器就能看到最新的代码效果。

看到类似这个就成功了：

Uvicorn running on http://127.0.0.1:8000

浏览器打开：

http://127.0.0.1:8000

就应该会看到：

{"message":"Hello FastAPI"}

打开接口文档

http://127.0.0.1:8000/docs

这个页面非常重要，FastAPI会自动生成接口测试文档

总结

04.路由

在 Web 开发中，路由（Routing） 可以通俗地理解为 Web 应用的“交通导航系统”或“前台接待员”。它的核心作用是：建立“用户请求”与“后端处理代码”之间的映射关系。

请求方法

在 Web 开发中，请求方法（HTTP Method） 就像是你对服务器发出的“动作指令”。如果把访问网页比作去餐厅点菜，那么 URL（路径）就是你要点的菜名，而请求方法就是你对服务员说的动词（比如“给我拿一份”、“我要加个菜”、“这道菜退掉”）。

在 HTTP 协议中，最常用的请求方法有以下几种（通常被称为 CRUD 操作）：

1. GET（获取资源）

  • 作用：向服务器请求获取数据。

  • 特点：它是绝对安全的，因为你只是“看”数据，不会修改服务器上的任何东西。

  • 生活比喻：去图书馆借阅/查看一本书。

  • FastAPI 示例：@app.get("/users") （获取所有用户列表）

2. POST（创建资源）

  • 作用：向服务器提交数据，通常用于创建一个新的资源。

  • 特点：数据通常放在请求体（Body）中，而不是 URL 里。

  • 生活比喻：在图书馆登记一本新书入库。

  • FastAPI 示例：@app.post("/users") （注册/创建一个新用户）

3. PUT / PATCH（更新资源）

  • 作用：向服务器修改现有的数据。

    • PUT 通常表示全量更新（把原来的数据整个替换掉）。

    • PATCH 通常表示局部更新（只修改其中几个字段）。

  • 生活比喻：去图书馆修改你之前登记的书籍信息（比如改个书名）。

  • FastAPI 示例：@app.put("/users/123") （更新 ID 为 123 的用户信息）

4. DELETE（删除资源）

  • 作用：要求服务器删除指定的数据。

  • 生活比喻：在图书馆销毁/下架一本书。

  • FastAPI 示例：@app.delete("/users/123") （删除 ID 为 123 的用户）

为什么同一个 URL 可以有不同的结果？

这就是请求方法的威力所在！在 RESTful 架构中，URL 代表“资源”，而请求方法代表“对资源的操作”。

假设我们有一个接口 /users/123：

  • 如果你用 GET 请求它：服务器会返回 123 号用户的资料。

  • 如果你用 PUT 请求它：服务器会更新 123 号用户的资料。

  • 如果你用 DELETE 请求它：服务器会把 123 号用户删掉。

from fastapi import FastAPI

#创建fastapi实例
app = FastAPI()


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


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

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

运行后：

总结

#练习：访问路径/user/hello,响应结果是{"msg":"我正在学习FastAPI"}
@app.get("/user/hello")
async def get_user_hello():
    return {"msg":"我正在学习FastAPI"}

这段代码中的路由为“/user/hello”

05.参数简介和路径参数

同一段接口逻辑，根据参数不同返回不同的数据

参数分类

        路径参数/查询参数/请求体

路径参数
@app.get("/book/{id}") #id就是路径参数的名字
async def get_book(id: int) 
    return {"id": id, "title": f"这是第{id}本书"}

练习

@app.get("/book/{id}")
async def get_info(id:int):
    return {"id":id,"name":f"这人名叫{id}"}

06.路径参数_path类型注解

允许用路径参数进行额外的注解

path路径参数

路径参数（Path Parameter）是指直接嵌入在 URL 路径中的变量，用于标识特定资源。它通过装饰器（如 @app.get()）定义，并在处理函数中作为参数接收，FastAPI 会自动进行类型校验和文档生成。

核心特性
  • URL 结构绑定：路径参数是 URL 的一部分，例如 /users/{user_id} 中的 {user_id}，访问 /users/123 时，user_id 的值就是 123。
  • 自动类型转换与校验：通过在函数参数中声明类型（如 int、str），FastAPI 会自动将字符串转换为对应类型，若转换失败则返回 422 Unprocessable Entity 错误。
  • 文档自动生成：所有路径参数都会在 Swagger UI 或 ReDoc 等交互式文档中自动展示，包括名称、类型和描述。
进阶用法
  • 添加校验规则：可使用 Path() 函数设置约束，如最小值、最大值、正则表达式等。
  • 别名支持：通过 alias 参数允许 URL 中使用不同名称的参数。
  • 默认值与可选性：虽然路径参数通常必填，但可通过 Optional 类型配合默认值实现灵活设计。

使用示范

先导入

from fastapi import FastAPI, Path

代码

@app.get("/book/{id}")
async def get_book(id: int = Path(..., gt=0, lt=101, description="书的id，取值方位必须在1到100之间")):
    return {"id": id,"title": f"这是第{id}本书"}
#需求：查找书籍作者，路径参数name，长度范围在2-10之间
@app.get("/author/{name}")
async def get_name(name:str = Path(..., min_length=2, max_length=10)):
    return {"msg": f"这是{name}的信息"}
总结

练习

@app.get("/id/{id}")
async def get_id(id:int = Path(..., ge = 1, le = 100)):
    return {"id":id}

@app.get("/class_name/{class_name}")
async def get_class(class_name:str = Path(min_length = 2,max_length = 10)):
    return {"class":class_name}

07.查询参数和Query类型注解

查询参数不需要在路径上定义参数，直接给处理函数添加形参，这些形参就会自动被解释为查询参数，形参还允许等号赋默认值，

示例
#需求 查询新闻->分页，skip：记录跳过的记录数，limit：返回的记录数 10
@app.get("/news/news_list")
async def get_news_list(skip: int,limit: int = 10):
    return {"skip":skip, "limit":limit}
#开发中对大量参数进行筛选过滤就用查询参数即可
注解示例

查询参数允许做额外的类型注解，query

导入query
#需求 查询新闻->分页，skip：记录跳过的记录数，limit：返回的记录数 10
@app.get("/news/news_list")
async def get_news_list(
    skip: int = Query(0, description="跳过的记录数",lt = 100), #又需要默认值有需要额外校验，就把“...”换成默认值即可
    limit: int = Query(10, description="返回的记录数") #又需要默认值有需要额外校验，就把“...”换成默认值即可
    ):
    return {"skip":skip, "limit":limit}
#开发中对大量参数进行筛选过滤就用查询参数即可
#路径上无需再定义参数了，直接给处理函数定义形参，形参会被自动解释为查询参数。
总结

练习

@app.get("/find_book")
async def find_book_function(class_name:str = Query("Python开发",min_length=5,max_length=255),price: int = Query(...,ge=50,le=100)):
    return {"class":class_name,"price":price}

08.请求体参数

在前端或后端开发中，路由是指URL 请求与具体处理逻辑（如页面、组件或 API 接口）之间的映射关系。

示例

先导包

from pydantic import BaseModel
from fastapi import FastAPI,Path,Query
from pydantic import BaseModel

#创建fastapi实例
app = FastAPI()


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

#定义类型
#注册：用户名和密码->str
class User(BaseModel):
    username: str
    password: str


#写路由
@app.post("/register")
async def register(user: User):
    return user

请求体参数不在url里面，在消息体body里面


在 Web 开发中，POST 请求专门用于向服务器发送数据，通常用于创建新资源或提交表单。在代码中，它意味着这个接口是用来接收用户提交的注册信息（如用户名和密码）的。

代码拆解：
  • @app：代表Web 应用程序实例。
  • .post("/register")：这是一个装饰器（Decorator）。它告诉程序：“当客户端（比如浏览器或前端页面）向 /register 这个路径发送一个 POST 类型的请求时，请执行下面紧挨着的 register 函数”。
  • user: 程序会自动从 POST 请求的“请求体（Body）”中提取数据，并将其转换为 User 对象传给函数。
为什么用 POST 而不是 GET？
  • 数据位置：POST 请求将数据放在请求体（Body）中，而不是像 GET 请求那样暴露在 URL 链接中。
  • 安全性：因为数据不在 URL 中，所以它更适合传输密码等敏感信息，避免密码被记录在浏览器历史或服务器日志中。
  • 数据量：POST 请求理论上没有长度限制，适合传输包含较多字段的用户注册数据。
练习

from fastapi import FastAPI,Path,Query
from pydantic import BaseModel

#创建fastapi实例
app = FastAPI()

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

#定义类型
#：用户名和密码->str
class Book(BaseModel):
    book_name: str
    book_author: str
    publisher: str
    price: int


#写路由
@app.post("/add_book")
async def add_books(book: Book):
    return book

09.请求体参数-类型注解Field

Python原生注解

这是 Python 语言本身支持的语法特性，自 Python 3.5 起引入，主要用于类型提示。它通过在变量、函数参数或返回值后加冒号和类型表达式来声明预期数据类型。

  • 作用：提升代码可读性，辅助静态类型检查工具（如 mypy），但不强制执行运行时校验。
  • 示例：
class User(BaseModel):
    username: str       # 原生注解：声明 username 应为字符串
    password: str       # 原生注解：声明 password 应为字符串
  • 特点：简洁直观，仅表达“期望类型”，不包含校验规则（如长度、范围、是否必填等）。
Field 注解

这是 Pydantic 库提供的专用函数/标记，用于在模型字段上添加更丰富的元信息和校验约束。它通常作为字段的默认值或显式赋值使用。

  • 作用：不仅声明类型，还可定义校验规则（如最小/最大长度、正则匹配、默认值、描述信息等），并影响 OpenAPI 文档生成。
  • 示例：
from pydantic import BaseModel, Field

class User(BaseModel):
    username: str = Field(..., min_length=3, max_length=50, description="用户名，3-50字符")
    password: str = Field(..., min_length=8, description="密码至少8位")
  • 特点：
    • ... 表示该字段为必填项；
    • 可设置 default、title、example、ge（大于等于）、le（小于等于）等参数；
    • 与 Pydantic 的校验引擎深度集成，自动在请求解析时执行校验。
Pydantic库

Pydantic 是一个用于数据验证和设置管理的 Python 库，它通过 Python 类型注解（Type Hints）来定义数据结构，并自动验证输入数据的合法性。

它的核心功能包括：

  • 数据验证：自动检查输入数据是否符合类型和约束条件（如字符串长度、数字范围等）。
  • 类型转换：将原始数据（如 JSON、字典）转换为 Python 类型（如 datetime、Enum）。
  • 序列化：轻松将模型转换为字典或 JSON。
  • IDE 友好：配合类型提示，提供代码补全和错误检查。
示例
from fastapi import FastAPI,Path,Query
from pydantic import BaseModel,Field

#创建fastapi实例
app = FastAPI()

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

#注册：用户名和密码->str
class User(BaseModel):
    username: str = Field(default="张三",min_length=2,max_length=10,description="用户名，长度要求2-10个字")
    password: str = Field(min_length=3,max_length=20) #Field()表示字段的描述信息，可以省略


#写路由
@app.post("/register")
async def add_books(user: User):
    return user
练习

from fastapi import FastAPI
from pydantic import BaseModel,Field

#创建fastapi实例
app = FastAPI()

#注册
class Book(BaseModel):
    title: str = Field(..., min_length=2, max_length=20)
    author:str = Field(...,min_length=2,max_length=10)
    publisher:str = Field(default = "黑马出版社")
    price:float = Field(..., gt=0)

@app.post("/add_book")
async def add_book(book:Book):
    return book

10.响应类型-JSON格式

序列化

序列化是将程序中的对象、数据结构或类实例转换为可存储或可传输格式（如字节流、JSON字符串、XML等）的过程，其核心目的是让数据能够脱离内存环境，在不同系统、网络或存储介质中安全传递与持久化保存。反向过程称为反序列化，即将序列化后的数据还原为原始对象结构。

在Python生态中，序列化常用于以下场景：

  • 网络通信：将对象转为JSON或二进制格式通过HTTP、gRPC等协议传输；
  • 数据持久化：将对象状态保存到文件或数据库，供后续读取恢复；
  • 缓存机制：将复杂对象序列化后存入Redis、Memcached等缓存系统；
  • 跨语言交互：使用通用格式（如JSON、Protobuf）实现不同编程语言间的数据交换。

11.响应类型-HTML格式

json是fastapi自动转换，如果不是json的话需要程序员来设置响应类型，fastapi提供了两种方式

本处使用在装饰器中指定响应类即可

示例
from fastapi import FastAPI
from fastapi.responses import HTMLResponse

app = FastAPI()

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

#接口 -> 响应HTML 代码
@app.get("/html",response_class=HTMLResponse)
async def heml():
    return "<h1>这是一级标题</h1>"

12.相应类型-文件格式

响应类不再是添加到装饰器里面，而是直接返回响应对象

13.自定义响应数据格式

当一个接口响应的数据格式是json时，我们可以自定义一个数据格式，其实就是一个类，用这个类来约定响应的数据包含哪些属性是什么类型

from fastapi import FastAPI
from fastapi.responses import HTMLResponse,FileResponse
from pydantic import BaseModel


app = FastAPI()

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

#需求：新闻接口 -> 响应数据格式 id title content
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": "这是一本好书"
    }

约定了响应数据格式，响应数据就必须是这个格式

Json格式

14.异常响应处理

from fastapi import FastAPI,HTTPException
from fastapi.responses import HTMLResponse,FileResponse
from pydantic import BaseModel


app = FastAPI()

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


#需求：按id查询新闻 -> 1-6
@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_code=404,detail="您查找的新闻不存在")
    return {"id": id}
raise和return的区别
return（正常返回）
  • 含义：函数正常执行完毕，把结果交还给调用者。
  • 作用：结束当前函数的执行，并返回一个值（如果不写返回值，默认返回 None）。
  • 类比：就像去餐厅点餐，厨师做好了菜，正常端给你。
  • 代码示例：
def get_user():
    return {"username": "alice"}  # 正常返回数据
raise（抛出异常）
  • 含义：函数在执行过程中遇到了无法处理的错误，主动中断执行，并向上传递错误信息。
  • 作用：强制停止当前函数的执行，抛出一个异常（Exception）。如果调用者没有用 try...except 捕获这个异常，整个程序就会崩溃并报错。
  • 类比：厨师在做菜时发现食材坏了，直接拉响火警警报，把菜倒了，并大喊“食材有问题，这菜做不了！”。
def get_user(age):
    if age < 0:
        raise ValueError("年龄不能是负数！")  # 发现错误，抛出异常
    return {"age": age}
总结

想要传递数据，用return。想要报告错误，用raise。

FastAPI进阶

15.中间件

实际开发中可以有多个中间件

fastapi的执行顺序，按照代码顺序自下而上

示例
from fastapi import FastAPI

app = FastAPI()

@app.middleware("http") #这是一个装饰器。它的作用是将下面定义的函数注册为 FastAPI 应用的中间件，并指定其类型为 "http"。这意味着该函数专门用于拦截和处理所有的 HTTP 请求和响应。
async def middleware1(request,call_next):
    print("中间件1 start")
    response = await call_next(request) #saync代表异步函数，真有异步代码，开头要加上await，await后放要异步执行的代码
#request：第一个固定参数，代表当前传入的 HTTP 请求对象。你可以从中获取请求头、路径、方法等信息。
#call_next：第二个固定参数，它是一个函数。调用它的作用是将当前的 request 传递给下一个中间件，或者传递给最终的路由处理函数。
    print("中间件1 end")
    return response

@app.middleware("http") #指定为 "http" 后，该中间件仅应用于 HTTP 请求
async def middleware2(request,call_next):
    print("中间件2 start")
    response = await call_next(request) #saync代表异步函数，真有异步代码，开头要加上await，await后放要异步执行的代码
    print("中间件2 end")
    return response

@app.get("/")
async def root():
    return {"message": "hello FastAPI"}
总结

16.依赖注入

问：中间件也能封装通用的逻辑，也能减少代码重复，为什么这里用以来系统更合适而不能用中间件?

中间件控制所有请求，每个请求都自动执行中间件注入的代码，中间件不一样，中间件是程序员说了算

创建依赖项：就是封装函数

示例
from fastapi import FastAPI,Query,Depends #2.导入Depends

app = FastAPI()

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

#分页参数逻辑共用：新闻列表和用户列表共用
#1.依赖项
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


@app.get("/user/user_list")
async def get_user_list():
    return {"message": "hello"}
总结

17.orm简介和安装

18.ORM建表

什么ORM

ORM 是 Object Relational Mapping 的缩写，中文叫：对象关系映射。大白话就是：

用 Python 里的“类”和“对象”来操作数据库，而不是直接写 SQL。

创建数据库引擎：

        先建立 Python 程序和 MySQL 数据库之间的连接通道。

        可以理解为：

        FastAPI 想操作 MySQL，必须先知道连接哪个数据库、用户名密码是什么、用什么驱动连接。

定义模型类：

        用 Python 类来描述数据库表的结构。

        也就是告诉 SQLAlchemy：

我要创建一张 books 表，
里面有 id、bookname、author、price、publisher、create_time、update_time 这些字段。

启动应用时建表

        当 FastAPI 项目启动时，自动执行建表函数。

        也就是你运行：

uvicorn main:app --reload

        之后，程序会自动去 MySQL 里检查有没有 books 表。如果没有，就创建。如果已经有了，一般不会重复创建。

ORM-创建数据库引擎

ASYNC_DATABASE_URL数据库+驱动：//用户名：密码@localhost：mysql数据库端口号/数据库名？字符编码（可以省略也可以换为别的）

from datetime import datetime
from fastapi import FastAPI
from sqlalchemy import DateTime, Float, String, func
from sqlalchemy.orm import DeclarativeBase,Mapped,mapped_column
from sqlalchemy.ext.asyncio import create_async_engine

app = FastAPI()

#1.创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://root:123456@localhost:3306/FastAPI_first?charset=utf8"
async_engine = create_async_engine(
    ASYNC_DATABASE_URL, 
    echo = True, #可选，输出SQL日志
    pool_size = 10, #设置连接池活跃的链接数
    max_overflow = 20 #允许额外的连接数
)

#2.定义模型类：基类 + 表对应的模型类
#基类：创建时间、更新时间：书籍表：id、书名、作者、价格、出版社
class Base(DeclarativeBase):
    create_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(),default=func.now(),comment="创建时间")
    update_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(),default=func.now, onupdate=func.now(),comment="更新时间")   #创建时间

class Book(Base):
    __tablename__ = "books"
    id: Mapped[int] = mapped_column(primary_key=True,comment="书籍id")
    bookname: Mapped[str] = mapped_column(String(255),comment="书名")
    author: Mapped[str] = mapped_column(String(255),comment="作者")
    price: Mapped[float] = mapped_column(Float,comment="价格")
    publisher: Mapped[str] = mapped_column(String(255),comment="出版社")

示例
from datetime import datetime
from re import S
from fastapi import FastAPI
from sqlalchemy import DateTime, Float, String, func
from sqlalchemy.orm import DeclarativeBase,Mapped,mapped_column
from sqlalchemy.ext.asyncio import create_async_engine

app = FastAPI()

#1.创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://root:123456@localhost:3306/FastAPI_first?charset=utf8"
async_engine = create_async_engine(
    ASYNC_DATABASE_URL, 
    echo = True, #可选，输出SQL日志
    pool_size = 10, #设置连接池活跃的链接数
    max_overflow = 20 #允许额外的连接数
)


#2.定义模型类：基类 + 表对应的模型类
#基类：创建时间、更新时间：书籍表：id、书名、作者、价格、出版社
class Base(DeclarativeBase):
    create_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(),default=func.now(),comment="创建时间")
    update_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(),default=func.now(), onupdate=func.now(),comment="更新时间")   #创建时间

class Book(Base):
    __tablename__ = "books"
    id: Mapped[int] = mapped_column(primary_key=True,comment="书籍id")
    bookname: Mapped[str] = mapped_column(String(255),comment="书名")
    author: Mapped[str] = mapped_column(String(255),comment="作者")
    price: Mapped[float] = mapped_column(Float,comment="价格")
    publisher: Mapped[str] = mapped_column(String(255),comment="出版社")


#3.建表：定义函数建表->FastAPI启动的时候调用建表的函数
async def create_tables():
    #获取异步引擎，创建事务-建表
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)


@app.on_event("startup")
async def startup_event():
    await create_tables()


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

from datetime import datetime
from fastapi import FastAPI
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import DeclarativeBase,Mapped,mapped_column
from sqlalchemy import DateTime, Float, String, func


app = FastAPI()

ASYNC_DATABASE_URL = "mysql+aiomysql://root:123456@localhost:3306/FastAPI_first?charset=utf8"

async_engine = create_async_engine(
    ASYNC_DATABASE_URL, 
    echo = True, #可选，输出SQL日志
    pool_size = 10,
    max_overflow= 20
)

class Base(DeclarativeBase):
    create_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(),default=func.now(),comment="创建时间") #创建时间
    update_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(),default=func.now(), onupdate=func.now(),comment="更新时间")   #更新时间（mapped_column() 是 SQLAlchemy 用来定义数据库字段的函数。）

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(255),comment="用户名")
    password: Mapped[str] = mapped_column(String(255),comment="密码")


#3.建表：定义函数建表->FastAPI启动的时候调用建表的函数
async def create_tables():
    #获取异步引擎，创建事务-建表
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)


@app.on_event("startup")
async def startup_event():
    await create_tables()


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

19.ORM在路由中使用ORM

from datetime import datetime
from fastapi import Depends, FastAPI
from sqlalchemy import DateTime, Float, String, func, select
from sqlalchemy.orm import DeclarativeBase,Mapped,mapped_column
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine,AsyncSession

app = FastAPI()

#1.创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://root:123456@localhost:3306/FastAPI_first?charset=utf8"
async_engine = create_async_engine(
    ASYNC_DATABASE_URL, 
    echo = True, #可选，输出SQL日志
    pool_size = 10, #设置连接池活跃的链接数
    max_overflow = 20 #允许额外的连接数
)


#2.定义模型类：基类 + 表对应的模型类
#基类：创建时间、更新时间：书籍表：id、书名、作者、价格、出版社
#Base 是所有表的公共基类，负责公共字段
class Base(DeclarativeBase): #DeclarativeBase 的作用是：让 SQLAlchemy 知道：这个类以及它的子类，可以被当成数据库表模型来处理。
    create_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(),default=func.now(),comment="创建时间") #创建时间
    update_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(),default=func.now(), onupdate=func.now(),comment="更新时间")   #更新时间（mapped_column() 是 SQLAlchemy 用来定义数据库字段的函数。）

#Book 是真正对应数据库中 books 表的模型类。
class Book(Base):
    __tablename__ = "books" #这个 ORM 模型对应数据库中的 books 表。
    id: Mapped[int] = mapped_column(primary_key=True,comment="书籍id")
    bookname: Mapped[str] = mapped_column(String(255),comment="书名")
    author: Mapped[str] = mapped_column(String(255),comment="作者")
    price: Mapped[float] = mapped_column(Float,comment="价格")
    publisher: Mapped[str] = mapped_column(String(255),comment="出版社")


#3.建表：定义函数建表->FastAPI启动的时候调用建表的函数
async def create_tables():
    #获取异步引擎，创建事务-建表
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)


@app.on_event("startup")
async def startup_event():
    await create_tables()


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


#需求：查询功能的接口，查询图书->依赖注入：创建依赖项获取数据库会话+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)):
    #查询
    result = await db.execute(select(Book))
    user = result.scalars().all()
    return user

20.ORM操作数据-查询数据

from datetime import datetime
from fastapi import Depends, FastAPI
from sqlalchemy import DateTime, Float, String, func, select
from sqlalchemy.orm import DeclarativeBase,Mapped,mapped_column
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine,AsyncSession

app = FastAPI()

#1.创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://root:123456@localhost:3306/FastAPI_first?charset=utf8"
async_engine = create_async_engine(
    ASYNC_DATABASE_URL, 
    echo = True, #可选，输出SQL日志
    pool_size = 10, #设置连接池活跃的链接数
    max_overflow = 20 #允许额外的连接数
)


#2.定义模型类：基类 + 表对应的模型类
#基类：创建时间、更新时间：书籍表：id、书名、作者、价格、出版社
#Base 是所有表的公共基类，负责公共字段
class Base(DeclarativeBase): #DeclarativeBase 的作用是：让 SQLAlchemy 知道：这个类以及它的子类，可以被当成数据库表模型来处理。
    create_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(),default=func.now(),comment="创建时间") #创建时间
    update_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(),default=func.now(), onupdate=func.now(),comment="更新时间")   #更新时间（mapped_column() 是 SQLAlchemy 用来定义数据库字段的函数。）

#Book 是真正对应数据库中 books 表的模型类。
class Book(Base):
    __tablename__ = "books" #这个 ORM 模型对应数据库中的 books 表。
    id: Mapped[int] = mapped_column(primary_key=True,comment="书籍id")
    bookname: Mapped[str] = mapped_column(String(255),comment="书名")
    author: Mapped[str] = mapped_column(String(255),comment="作者")
    price: Mapped[float] = mapped_column(Float,comment="价格")
    publisher: Mapped[str] = mapped_column(String(255),comment="出版社")


#3.建表：定义函数建表->FastAPI启动的时候调用建表的函数
async def create_tables():
    #获取异步引擎，创建事务-建表
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)


@app.on_event("startup")
async def startup_event():
    await create_tables()


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


#需求：查询功能的接口，查询图书->依赖注入：创建依赖项获取数据库会话+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)):
    #查询
    result =await db.execute(select(Book)) #查询->返回一个ORM对象
    # book = result.scalars().all() #获取所有
    # book = result.scalars().first() #获取第一条条
    book = await db.get(Book,3) #获取单条，根据主键
    return book

21.ORM操作数据-条件查询-比较数据

from datetime import datetime
from fastapi import Depends, FastAPI
from sqlalchemy import DateTime, Float, String, func, select
from sqlalchemy.orm import DeclarativeBase,Mapped,mapped_column
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine,AsyncSession

app = FastAPI()

#1.创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://root:123456@localhost:3306/FastAPI_first?charset=utf8"
async_engine = create_async_engine(
    ASYNC_DATABASE_URL, 
    echo = True, #可选，输出SQL日志
    pool_size = 10, #设置连接池活跃的链接数
    max_overflow = 20 #允许额外的连接数
)


#2.定义模型类：基类 + 表对应的模型类
#基类：创建时间、更新时间：书籍表：id、书名、作者、价格、出版社
#Base 是所有表的公共基类，负责公共字段
class Base(DeclarativeBase): #DeclarativeBase 的作用是：让 SQLAlchemy 知道：这个类以及它的子类，可以被当成数据库表模型来处理。
    create_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(),default=func.now(),comment="创建时间") #创建时间
    update_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(),default=func.now(), onupdate=func.now(),comment="更新时间")   #更新时间（mapped_column() 是 SQLAlchemy 用来定义数据库字段的函数。）

#Book 是真正对应数据库中 books 表的模型类。
class Book(Base):
    __tablename__ = "books" #这个 ORM 模型对应数据库中的 books 表。
    id: Mapped[int] = mapped_column(primary_key=True,comment="书籍id")
    bookname: Mapped[str] = mapped_column(String(255),comment="书名")
    author: Mapped[str] = mapped_column(String(255),comment="作者")
    price: Mapped[float] = mapped_column(Float,comment="价格")
    publisher: Mapped[str] = mapped_column(String(255),comment="出版社")


#3.建表：定义函数建表->FastAPI启动的时候调用建表的函数
async def create_tables():
    #获取异步引擎，创建事务-建表
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)


@app.on_event("startup")
async def startup_event():
    await create_tables()


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


#需求：查询功能的接口，查询图书->依赖注入：创建依赖项获取数据库会话+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)):
    #查询
    result =await db.execute(select(Book)) #查询->返回一个ORM对象
    # book = result.scalars().all() #获取所有
    # book = result.scalars().first() #获取第一条条
    book = await db.get(Book,3) #获取单条，根据主键
    return book

#需求：路径参数 书籍id
@app.get("/book/get_book/{book_id}")
async def get_book_list(book_id:int,db:AsyncSession = Depends(get_database)):
    result = await db.execute(select(Book).where(Book.id == book_id))
    book = result.scalar_one_or_none() #获取单条，根据主键
    return book

#需求：条件 价格大于等于200
@app.get("/book/search_book")
async def get_search_book(db: AsyncSession = Depends(get_database)):
    result = await db.execute(select(Book).where(Book.price >=50))
    books = result.scalars().all()
    return books

22.ORM操作数据-条件查询-模糊&与非&包含

from datetime import datetime
from fastapi import Depends, FastAPI
from sqlalchemy import DateTime, Float, String, func, select
from sqlalchemy.orm import DeclarativeBase,Mapped,mapped_column
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine,AsyncSession

app = FastAPI()

#1.创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://root:123456@localhost:3306/FastAPI_first?charset=utf8"
async_engine = create_async_engine(
    ASYNC_DATABASE_URL, 
    echo = True, #可选，输出SQL日志
    pool_size = 10, #设置连接池活跃的链接数
    max_overflow = 20 #允许额外的连接数
)


#2.定义模型类：基类 + 表对应的模型类
#基类：创建时间、更新时间：书籍表：id、书名、作者、价格、出版社
#Base 是所有表的公共基类，负责公共字段
class Base(DeclarativeBase): #DeclarativeBase 的作用是：让 SQLAlchemy 知道：这个类以及它的子类，可以被当成数据库表模型来处理。
    create_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(),default=func.now(),comment="创建时间") #创建时间
    update_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(),default=func.now(), onupdate=func.now(),comment="更新时间")   #更新时间（mapped_column() 是 SQLAlchemy 用来定义数据库字段的函数。）

#Book 是真正对应数据库中 books 表的模型类。
class Book(Base):
    __tablename__ = "books" #这个 ORM 模型对应数据库中的 books 表。
    id: Mapped[int] = mapped_column(primary_key=True,comment="书籍id")
    bookname: Mapped[str] = mapped_column(String(255),comment="书名")
    author: Mapped[str] = mapped_column(String(255),comment="作者")
    price: Mapped[float] = mapped_column(Float,comment="价格")
    publisher: Mapped[str] = mapped_column(String(255),comment="出版社")


#3.建表：定义函数建表->FastAPI启动的时候调用建表的函数
async def create_tables():
    #获取异步引擎，创建事务-建表
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)


@app.on_event("startup")
async def startup_event():
    await create_tables()


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


#需求：查询功能的接口，查询图书->依赖注入：创建依赖项获取数据库会话+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)):
    #查询
    result =await db.execute(select(Book)) #查询->返回一个ORM对象
    # book = result.scalars().all() #获取所有
    # book = result.scalars().first() #获取第一条条
    book = await db.get(Book,3) #获取单条，根据主键
    return book

#需求：路径参数 书籍id
@app.get("/book/get_book/{book_id}")
async def get_book_list(book_id:int,db:AsyncSession = Depends(get_database)):
    result = await db.execute(select(Book).where(Book.id == book_id))
    book = result.scalar_one_or_none() #获取单条，根据主键
    return book

#需求：条件 价格大于等于200
@app.get("/book/search_book")
async def get_search_book(db: AsyncSession = Depends(get_database)):
    result = await db.execute(select(Book).where(Book.price >=50))
    books = result.scalars().all()
    return books

#需求：作者以曹开头
@app.get("/book/search_book1")
async def get_search_book1(db: AsyncSession = Depends(get_database)):
    result = await db.execute(select(Book).where(Book.author.like("曹%")))
    book = result.scalars().all()
    return book


@app.get("/book/search_book2")
async def get_search_book2(db: AsyncSession = Depends(get_database)):
    # result = await db.execute(select(Book).where(Book.author.like("曹%")))
    result = await db.execute(select(Book).where((Book.author.like("曹%")) & (Book.price >= 100)))
    book = result.scalars().all()
    return book

@app.get("/book/search_book3")
async def get_search_book3(db: AsyncSession = Depends(get_database)):
    # result = await db.execute(select(Book).where(Book.author.like("曹%")))
    #result = await db.execute(select(Book).where((Book.author.like("曹%")) & (Book.price >= 100)))
    #需求：书籍id列表，数据库里面的id如果在书籍id列表里面就返回
    id_list = [1,3,5,7]
    result = await db.execute(select(Book).where(Book.id.in_(id_list)))
    book = result.scalars().all()
    return book

23.ORM操作数据-聚合查询

#聚合查询
select( func.方法名(模型类.属性))
from datetime import datetime
from fastapi import Depends, FastAPI
from sqlalchemy import DateTime, Float, String, func, select
from sqlalchemy.orm import DeclarativeBase,Mapped,mapped_column
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine,AsyncSession

app = FastAPI()

#1.创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://root:123456@localhost:3306/FastAPI_first?charset=utf8"
async_engine = create_async_engine(
    ASYNC_DATABASE_URL, 
    echo = True, #可选，输出SQL日志
    pool_size = 10, #设置连接池活跃的链接数
    max_overflow = 20 #允许额外的连接数
)


#2.定义模型类：基类 + 表对应的模型类
#基类：创建时间、更新时间：书籍表：id、书名、作者、价格、出版社
#Base 是所有表的公共基类，负责公共字段
class Base(DeclarativeBase): #DeclarativeBase 的作用是：让 SQLAlchemy 知道：这个类以及它的子类，可以被当成数据库表模型来处理。
    create_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(),default=func.now(),comment="创建时间") #创建时间
    update_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(),default=func.now(), onupdate=func.now(),comment="更新时间")   #更新时间（mapped_column() 是 SQLAlchemy 用来定义数据库字段的函数。）

#Book 是真正对应数据库中 books 表的模型类。
class Book(Base):
    __tablename__ = "books" #这个 ORM 模型对应数据库中的 books 表。
    id: Mapped[int] = mapped_column(primary_key=True,comment="书籍id")
    bookname: Mapped[str] = mapped_column(String(255),comment="书名")
    author: Mapped[str] = mapped_column(String(255),comment="作者")
    price: Mapped[float] = mapped_column(Float,comment="价格")
    publisher: Mapped[str] = mapped_column(String(255),comment="出版社")


#3.建表：定义函数建表->FastAPI启动的时候调用建表的函数
async def create_tables():
    #获取异步引擎，创建事务-建表
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)


@app.on_event("startup")
async def startup_event():
    await create_tables()


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


#需求：查询功能的接口，查询图书->依赖注入：创建依赖项获取数据库会话+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)):
    #查询
    result =await db.execute(select(Book)) #查询->返回一个ORM对象
    # book = result.scalars().all() #获取所有
    # book = result.scalars().first() #获取第一条条
    book = await db.get(Book,3) #获取单条，根据主键
    return book

#需求：路径参数 书籍id
@app.get("/book/get_book/{book_id}")
async def get_book_list(book_id:int,db:AsyncSession = Depends(get_database)):
    result = await db.execute(select(Book).where(Book.id == book_id))
    book = result.scalar_one_or_none() #获取单条，根据主键
    return book

#需求：条件 价格大于等于200
@app.get("/book/search_book")
async def get_search_book(db: AsyncSession = Depends(get_database)):
    result = await db.execute(select(Book).where(Book.price >=50))
    books = result.scalars().all()
    return books

#需求：作者以曹开头
@app.get("/book/search_book1")
async def get_search_book1(db: AsyncSession = Depends(get_database)):
    result = await db.execute(select(Book).where(Book.author.like("曹%")))
    book = result.scalars().all()
    return book


@app.get("/book/search_book2")
async def get_search_book2(db: AsyncSession = Depends(get_database)):
    # result = await db.execute(select(Book).where(Book.author.like("曹%")))
    result = await db.execute(select(Book).where((Book.author.like("曹%")) & (Book.price >= 100)))
    book = result.scalars().all()
    return book

@app.get("/book/search_book3")
async def get_search_book3(db: AsyncSession = Depends(get_database)):
    # result = await db.execute(select(Book).where(Book.author.like("曹%")))
    #result = await db.execute(select(Book).where((Book.author.like("曹%")) & (Book.price >= 100)))
    #需求：书籍id列表，数据库里面的id如果在书籍id列表里面就返回
    id_list = [1,3,5,7]
    result = await db.execute(select(Book).where(Book.id.in_(id_list)))
    book = result.scalars().all()
    return book

@app.get("/book/count")
async def get_count(db: AsyncSession = Depends(get_database)):
    #result =await db.execute(select(func.count(Book.id)))
    result =await db.execute(select(func.max(Book.price)))
    num = result.scalar()
    return num

24.ORM操作数据-分页查询

from datetime import datetime
from fastapi import Depends, FastAPI
from sqlalchemy import DateTime, Float, String, func, select
from sqlalchemy.orm import DeclarativeBase,Mapped,mapped_column
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine,AsyncSession

app = FastAPI()

#1.创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://root:123456@localhost:3306/FastAPI_first?charset=utf8"
async_engine = create_async_engine(
    ASYNC_DATABASE_URL, 
    echo = True, #可选，输出SQL日志
    pool_size = 10, #设置连接池活跃的链接数
    max_overflow = 20 #允许额外的连接数
)


#2.定义模型类：基类 + 表对应的模型类
#基类：创建时间、更新时间：书籍表：id、书名、作者、价格、出版社
#Base 是所有表的公共基类，负责公共字段
class Base(DeclarativeBase): #DeclarativeBase 的作用是：让 SQLAlchemy 知道：这个类以及它的子类，可以被当成数据库表模型来处理。
    create_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(),default=func.now(),comment="创建时间") #创建时间
    update_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(),default=func.now(), onupdate=func.now(),comment="更新时间")   #更新时间（mapped_column() 是 SQLAlchemy 用来定义数据库字段的函数。）

#Book 是真正对应数据库中 books 表的模型类。
class Book(Base):
    __tablename__ = "books" #这个 ORM 模型对应数据库中的 books 表。
    id: Mapped[int] = mapped_column(primary_key=True,comment="书籍id")
    bookname: Mapped[str] = mapped_column(String(255),comment="书名")
    author: Mapped[str] = mapped_column(String(255),comment="作者")
    price: Mapped[float] = mapped_column(Float,comment="价格")
    publisher: Mapped[str] = mapped_column(String(255),comment="出版社")


#3.建表：定义函数建表->FastAPI启动的时候调用建表的函数
async def create_tables():
    #获取异步引擎，创建事务-建表
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)


@app.on_event("startup")
async def startup_event():
    await create_tables()


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


#需求：查询功能的接口，查询图书->依赖注入：创建依赖项获取数据库会话+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)):
    #查询
    result =await db.execute(select(Book)) #查询->返回一个ORM对象
    # book = result.scalars().all() #获取所有
    # book = result.scalars().first() #获取第一条条
    book = await db.get(Book,3) #获取单条，根据主键
    return book


@app.get("/book/get_book_list")
async def get_book_list(
    page: int = 1,
    page_size: int = 2,
    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

25.ORM操作数据-查询总结

26.ORM操作数据-新增数据

from datetime import datetime
from fastapi import Depends, FastAPI
from sqlalchemy import DateTime, Float, String, func, select
from sqlalchemy.orm import DeclarativeBase,Mapped,mapped_column
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine,AsyncSession
from pydantic import BaseModel #请求体参数要定义一个类还得要继承一个basemodel基类

app = FastAPI()

#1.创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://root:123456@localhost:3306/FastAPI_first?charset=utf8"
async_engine = create_async_engine(
    ASYNC_DATABASE_URL, 
    echo = True, #可选，输出SQL日志
    pool_size = 10, #设置连接池活跃的链接数
    max_overflow = 20 #允许额外的连接数
)


#2.定义模型类：基类 + 表对应的模型类
#基类：创建时间、更新时间：书籍表：id、书名、作者、价格、出版社
#Base 是所有表的公共基类，负责公共字段
class Base(DeclarativeBase): #DeclarativeBase 的作用是：让 SQLAlchemy 知道：这个类以及它的子类，可以被当成数据库表模型来处理。
    create_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(),default=func.now(),comment="创建时间") #创建时间
    update_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(),default=func.now(), onupdate=func.now(),comment="更新时间")   #更新时间（mapped_column() 是 SQLAlchemy 用来定义数据库字段的函数。）

#Book 是真正对应数据库中 books 表的模型类。#针对表
class Book(Base):
    __tablename__ = "books" #这个 ORM 模型对应数据库中的 books 表。
    id: Mapped[int] = mapped_column(primary_key=True,comment="书籍id")
    bookname: Mapped[str] = mapped_column(String(255),comment="书名")
    author: Mapped[str] = mapped_column(String(255),comment="作者")
    price: Mapped[float] = mapped_column(Float,comment="价格")
    publisher: Mapped[str] = mapped_column(String(255),comment="出版社")


#3.建表：定义函数建表->FastAPI启动的时候调用建表的函数
async def create_tables():
    #获取异步引擎，创建事务-建表
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)


@app.on_event("startup")
async def startup_event():
    await create_tables()


#需求：查询功能的接口，查询图书->依赖注入：创建依赖项获取数据库会话+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)):
    #查询
    result =await db.execute(select(Book)) #查询->返回一个ORM对象
    # book = result.scalars().all() #获取所有
    # book = result.scalars().first() #获取第一条条
    book = await db.get(Book,3) #获取单条，根据主键
    return book


@app.get("/book/get_book_list")
async def get_book_list(
    page: int = 1,
    page_size: int = 2,
    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


#需求：用户输入图书信息（id、书名、作者、价格、出版社）->新增
#用户输入->参数->请求体参数
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)):
    book_obj = Book(**book.__dict__)
    db.add(book_obj)
    await db.commit()
    return book

27.ORM操作数据-更新数据

from datetime import datetime
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy import DateTime, Float, String, func, select
from sqlalchemy.orm import DeclarativeBase,Mapped,mapped_column
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine,AsyncSession
from pydantic import BaseModel #请求体参数要定义一个类还得要继承一个basemodel基类

app = FastAPI()

#1.创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://root:123456@localhost:3306/FastAPI_first?charset=utf8"
async_engine = create_async_engine(
    ASYNC_DATABASE_URL, 
    echo = True, #可选，输出SQL日志
    pool_size = 10, #设置连接池活跃的链接数
    max_overflow = 20 #允许额外的连接数
)


#2.定义模型类：基类 + 表对应的模型类
#基类：创建时间、更新时间：书籍表：id、书名、作者、价格、出版社
#Base 是所有表的公共基类，负责公共字段
class Base(DeclarativeBase): #DeclarativeBase 的作用是：让 SQLAlchemy 知道：这个类以及它的子类，可以被当成数据库表模型来处理。
    create_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(),default=func.now(),comment="创建时间") #创建时间
    update_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(),default=func.now(), onupdate=func.now(),comment="更新时间")   #更新时间（mapped_column() 是 SQLAlchemy 用来定义数据库字段的函数。）

#Book 是真正对应数据库中 books 表的模型类。#针对表
class Book(Base):
    __tablename__ = "books" #这个 ORM 模型对应数据库中的 books 表。
    id: Mapped[int] = mapped_column(primary_key=True,comment="书籍id")
    bookname: Mapped[str] = mapped_column(String(255),comment="书名")
    author: Mapped[str] = mapped_column(String(255),comment="作者")
    price: Mapped[float] = mapped_column(Float,comment="价格")
    publisher: Mapped[str] = mapped_column(String(255),comment="出版社")


#3.建表：定义函数建表->FastAPI启动的时候调用建表的函数
async def create_tables():
    #获取异步引擎，创建事务-建表
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)


@app.on_event("startup")
async def startup_event():
    await create_tables()


#需求：查询功能的接口，查询图书->依赖注入：创建依赖项获取数据库会话+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)):
    #查询
    result =await db.execute(select(Book)) #查询->返回一个ORM对象
    # book = result.scalars().all() #获取所有
    # book = result.scalars().first() #获取第一条条
    book = await db.get(Book,3) #获取单条，根据主键
    return book


@app.get("/book/get_book_list")
async def get_book_list(
    page: int = 1,
    page_size: int = 2,
    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


#需求：用户输入图书信息（id、书名、作者、价格、出版社）->新增
#用户输入->参数->请求体参数
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)):
    book_obj = Book(**book.__dict__)
    db.add(book_obj)
    await db.commit()
    return book



#需求：修改图书信息：先查再改
#设计思路：路径参数书籍id：作用是查找；请求体参数：作用是新数据（书名、作者、价格、出版社）
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)):
    db_book = await db.get(Book,book_id)
    if db_book is None:
        raise HTTPException(
            status_code=404, 
            detail="查无此书"
            )
    
    #重新赋值
    db_book.bookname = data.bookname
    db_book.author = data.author
    db_book.price = data.price
    db_book.publisher = data.publisher
    await db.commit()
    return db_book

28.ORM操作数据-删除数据

from datetime import datetime
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy import DateTime, Float, String, func, select
from sqlalchemy.orm import DeclarativeBase,Mapped,mapped_column
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine,AsyncSession
from pydantic import BaseModel #请求体参数要定义一个类还得要继承一个basemodel基类

app = FastAPI()

#1.创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://root:123456@localhost:3306/FastAPI_first?charset=utf8"
async_engine = create_async_engine(
    ASYNC_DATABASE_URL, 
    echo = True, #可选，输出SQL日志
    pool_size = 10, #设置连接池活跃的链接数
    max_overflow = 20 #允许额外的连接数
)


#2.定义模型类：基类 + 表对应的模型类
#基类：创建时间、更新时间：书籍表：id、书名、作者、价格、出版社
#Base 是所有表的公共基类，负责公共字段
class Base(DeclarativeBase): #DeclarativeBase 的作用是：让 SQLAlchemy 知道：这个类以及它的子类，可以被当成数据库表模型来处理。
    create_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(),default=func.now(),comment="创建时间") #创建时间
    update_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(),default=func.now(), onupdate=func.now(),comment="更新时间")   #更新时间（mapped_column() 是 SQLAlchemy 用来定义数据库字段的函数。）

#Book 是真正对应数据库中 books 表的模型类。#针对表
class Book(Base):
    __tablename__ = "books" #这个 ORM 模型对应数据库中的 books 表。
    id: Mapped[int] = mapped_column(primary_key=True,comment="书籍id")
    bookname: Mapped[str] = mapped_column(String(255),comment="书名")
    author: Mapped[str] = mapped_column(String(255),comment="作者")
    price: Mapped[float] = mapped_column(Float,comment="价格")
    publisher: Mapped[str] = mapped_column(String(255),comment="出版社")


#3.建表：定义函数建表->FastAPI启动的时候调用建表的函数
async def create_tables():
    #获取异步引擎，创建事务-建表
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)


@app.on_event("startup")
async def startup_event():
    await create_tables()


#需求：查询功能的接口，查询图书->依赖注入：创建依赖项获取数据库会话+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)):
    #查询
    result =await db.execute(select(Book)) #查询->返回一个ORM对象
    # book = result.scalars().all() #获取所有
    # book = result.scalars().first() #获取第一条条
    book = await db.get(Book,3) #获取单条，根据主键
    return book


@app.get("/book/get_book_list")
async def get_book_list(
    page: int = 1,
    page_size: int = 2,
    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


#需求：用户输入图书信息（id、书名、作者、价格、出版社）->新增
#用户输入->参数->请求体参数
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)):
    book_obj = Book(**book.__dict__)
    db.add(book_obj)
    await db.commit()
    return book



#需求：修改图书信息：先查再改
#设计思路：路径参数书籍id：作用是查找；请求体参数：作用是新数据（书名、作者、价格、出版社）
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)):
    db_book = await db.get(Book,book_id)
    if db_book is None:
        raise HTTPException(
            status_code=404, 
            detail="查无此书"
            )
    
    #重新赋值
    db_book.bookname = data.bookname
    db_book.author = data.author
    db_book.price = data.price
    db_book.publisher = data.publisher
    await db.commit()
    return db_book


#删除图书
@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="查无此书"
            )
    
    await db.delete(db_book)
    await db.commit()
    return {"msg":"删除图书成功"}

29.ORM总结

# fastapi
Logo
亚马逊云科技技术品牌专区

更多推荐

cover

别再只聊大模型！AI的下一场革命，藏在“身体”里

avatar 亚马逊云科技技术品牌专区
cover

别再只聊大模型！人工智能的下一场革命，藏在“身体”里

avatar 亚马逊云科技技术品牌专区
cover

别再只聊大模型！AI的下一场革命，藏在“身体”里

avatar 亚马逊云科技技术品牌专区