数据库操作

10.1 数据准备

10.1.3 表关系说明

我们准了两张表Books (图书) 表 和 Characters (人物) 表,两张表的关系是一对多的关系

表名称	描述	关键字段
`Books`	图书信息，是关系中的“一”方。	`book_id` (主键)
`Characters`	人物信息，是关系中的“多”方。	`char_id` (主键)

关系类型	描述
一对多 (1:N)	一本图书（`Books`）可以包含多位人物（`Characters`）。
外键	关系通过 `Characters` 表中的 `book_id` 字段实现，该字段引用了 `Books` 表中的 `book_id` 主键。

从多到一：要找到某个人物（例如“郭靖”）所属的书，只需查看该人物记录中的 book_id。

从一到多：要找到某本书（例如《射雕英雄传》）中的所有人物，只需查询 Characters 表中所有 book_id 等于该书 ID 的记录。

10.1.2 建表

使用SQL语句建表，方便写入测试数据

-- 确保在执行前选择正确的数据库，并在需要时删除旧表
-- DROP TABLE IF EXISTS Characters;
-- DROP TABLE IF EXISTS Books;
USE orm_db;
-- 1. 创建 Books (图书) 表
CREATE TABLE Books (
    -- 主键：图书唯一标识符
    book_id INT PRIMARY KEY AUTO_INCREMENT, 
    -- 书名，限制长度
    title VARCHAR(100) NOT NULL UNIQUE, 
    -- 作者
    author VARCHAR(50) NOT NULL, 
    -- 类型：古典名著或武侠小说
    genre VARCHAR(50) NOT NULL, 
    -- 创作或故事背景时代
    era VARCHAR(50) 
);

-- 2. 创建 Characters (人物) 表
CREATE TABLE Characters (
    -- 主键：人物唯一标识符
    char_id INT PRIMARY KEY AUTO_INCREMENT, 
    -- 人物姓名
    char_name VARCHAR(50) NOT NULL, 
    -- 性别
    gender VARCHAR(10), 
    -- 人物主要身份或角色类型
    role_type VARCHAR(50), 
    -- 外键：关联到所属图书的 ID
    book_id INT NOT NULL, 

    -- 定义外键约束：确保 book_id 必须存在于 Books 表中
    FOREIGN KEY (book_id) REFERENCES Books(book_id) 
);

-- 插入 Books 表数据 (共 15 条)
INSERT INTO Books (book_id, title, author, genre, era) VALUES
-- 四大名著 (ID 1-4)
(1, '《红楼梦》', '曹雪芹', '古典名著', '清朝'),
(2, '《三国演义》', '罗贯中', '古典名著', '明朝'),
(3, '《水浒传》', '施耐庵', '古典名著', '元末明初'),
(4, '《西游记》', '吴承恩', '古典名著', '明朝'),
-- 金庸小说 (ID 5-15)
(5, '《射雕英雄传》', '金庸', '武侠小说', '南宋'),
(6, '《神雕侠侣》', '金庸', '武侠小说', '南宋'),
(7, '《倚天屠龙记》', '金庸', '武侠小说', '元末明初'),
(8, '《天龙八部》', '金庸', '武侠小说', '北宋'),
(9, '《笑傲江湖》', '金庸', '武侠小说', '明朝（架空）'),
(10, '《鹿鼎记》', '金庸', '武侠小说', '清朝'),
(11, '《书剑恩仇录》', '金庸', '武侠小说', '清朝'),
(12, '《侠客行》', '金庸', '武侠小说', '明朝'),
(13, '《连城诀》', '金庸', '武侠小说', '清朝'),
(14, '《雪山飞狐》', '金庸', '武侠小说', '清朝'),
(15, '《飞狐外传》', '金庸', '武侠小说', '清朝');

-- 插入 Characters 表数据 (共 55 条)

INSERT INTO Characters (char_id, char_name, gender, role_type, book_id) VALUES
-- 《红楼梦》 (book_id: 1)
(101, '贾宝玉', '男', '主角', 1),
(102, '林黛玉', '女', '主角', 1),
(103, '薛宝钗', '女', '主角', 1),
(104, '王熙凤', '女', '主要角色', 1),
(105, '史湘云', '女', '金陵十二钗', 1),
-- 《三国演义》 (book_id: 2)
(201, '刘备', '男', '蜀汉君主', 2),
(202, '关羽', '男', '五虎上将', 2),
(203, '张飞', '男', '五虎上将', 2),
(204, '诸葛亮', '男', '军师', 2),
(205, '曹操', '男', '魏国奠基者', 2),
-- 《水浒传》 (book_id: 3)
(301, '宋江', '男', '梁山寨主', 3),
(302, '武松', '男', '行者', 3),
(303, '林冲', '男', '豹子头', 3),
(304, '鲁智深', '男', '花和尚', 3),
(305, '李逵', '男', '黑旋风', 3),
-- 《西游记》 (book_id: 4)
(401, '孙悟空', '男', '主角', 4),
(402, '唐僧', '男', '主角', 4),
(403, '猪八戒', '男', '配角', 4),
(404, '沙悟净', '男', '配角', 4),
(405, '观音菩萨', '女', '神仙', 4),
-- 《射雕英雄传》 (book_id: 5)
(501, '郭靖', '男', '主角', 5),
(502, '黄蓉', '女', '主角', 5),
(503, '杨康', '男', '主要反派', 5),
(504, '洪七公', '男', '北丐', 5),
(505, '黄药师', '男', '东邪', 5),
-- 《神雕侠侣》 (book_id: 6)
(601, '杨过', '男', '主角', 6),
(602, '小龙女', '女', '主角', 6),
(603, '郭芙', '女', '配角', 6),
(604, '李莫愁', '女', '反派', 6),
-- 《倚天屠龙记》 (book_id: 7)
(701, '张无忌', '男', '主角', 7),
(702, '赵敏', '女', '主角', 7),
(703, '周芷若', '女', '主要角色', 7),
(704, '张三丰', '男', '武学宗师', 7),
(705, '小昭', '女', '配角', 7),
-- 《天龙八部》 (book_id: 8)
(801, '乔峰', '男', '主角', 8),
(802, '段誉', '男', '主角', 8),
(803, '虚竹', '男', '主角', 8),
(804, '王语嫣', '女', '主要角色', 8),
(805, '阿朱', '女', '配角', 8),
-- 《笑傲江湖》 (book_id: 9)
(901, '令狐冲', '男', '主角', 9),
(902, '任盈盈', '女', '主角', 9),
(903, '岳不群', '男', '伪君子', 9),
(904, '东方不败', '待定', '大反派', 9),
-- 《鹿鼎记》 (book_id: 10)
(1001, '韦小宝', '男', '主角', 10),
(1002, '康熙', '男', '皇帝', 10),
(1003, '陈近南', '男', '天地会总舵主', 10),
(1004, '双儿', '女', '主要伴侣', 10),
-- 《书剑恩仇录》 (book_id: 11)
(1101, '陈家洛', '男', '主角', 11),
(1102, '霍青桐', '女', '主要角色', 11),
-- 《侠客行》 (book_id: 12)
(1201, '石破天', '男', '主角', 12),
(1202, '丁当', '女', '主要角色', 12),
-- 《连城诀》 (book_id: 13)
(1301, '狄云', '男', '主角', 13),
(1302, '戚芳', '女', '主要角色', 13),
-- 《雪山飞狐》 (book_id: 14)
(1401, '胡斐', '男', '主角', 14),
(1402, '苗若兰', '女', '主要角色', 14),
-- 《飞狐外传》 (book_id: 15)
(1501, '袁紫衣', '女', '主要角色', 15),
(1502, '程灵素', '女', '主要角色', 15);

10.1.3 定义模型类

在models.py文件下增加编写如下代码

from sqlalchemy import  ForeignKey
from sqlalchemy.orm import relationship, Mapped, mapped_column

# -------------------
# 2. Books (图书) ORM 模型
# 对应 SQL 表: Books
# -------------------
class Book(Base):
    __tablename__ = 'Books'

    # 字段映射 (对应 SQL 表中的列)
    book_id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    title: Mapped[str] = mapped_column(String(100), nullable=False, unique=True)
    author: Mapped[str] = mapped_column(String(50), nullable=False)
    genre: Mapped[str] = mapped_column(String(50), nullable=False)
    era: Mapped[str | None] = mapped_column(String(50), nullable=True) # 使用 str | None 表示可选

    # 关系字段：Book 是 "一" 方
    # back_populates='book'：告诉 SQLAlchemy 在 Character 模型中有一个对应的 'book' 字段
    # lazy='select'：表示在访问 characters 属性时才加载关联的人物列表
    characters: Mapped[list["Character"]] = relationship(
        back_populates="book",
        cascade="all, delete-orphan", # 级联删除，如果删除了 Book，则关联的 Character 也会被删除
    )

    def __repr__(self):
        return f"Book(id={self.book_id!r}, title={self.title!r}, author={self.author!r})"

# -------------------
# 3. Characters (人物) ORM 模型
# 对应 SQL 表: Characters
# -------------------
class Character(Base):
    __tablename__ = 'Characters'

    # 字段映射
    char_id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    char_name: Mapped[str] = mapped_column(String(50), nullable=False)
    gender: Mapped[str | None] = mapped_column(String(10), nullable=True)
    role_type: Mapped[str | None] = mapped_column(String(50), nullable=True)

    # 外键字段：关联到 Books.book_id
    book_id: Mapped[int] = mapped_column(ForeignKey("Books.book_id"))

    # 关系字段：Character 是 "多" 方
    # back_populates='characters'：告诉 SQLAlchemy 在 Book 模型中有一个对应的 'characters' 列表
    # 这个字段允许我们通过 character.book 访问所属的 Book 对象
    book: Mapped["Book"] = relationship(back_populates="characters")

    def __repr__(self):
        return f"Character(id={self.char_id!r}, name={self.char_name!r}, book_id={self.book_id!r})"

10.2 单表操作

10.2.1 数据查询

数据写格式：

await db.execute(select(模型类))

db.execute是指定异步执行

I. 查询所有数据

from models import Book

@app.get('/book/')
async def get_book(db: AsyncSession = Depends(get_database)):
    result = await db.execute(select(Book))
    # 获取所有数据
    book = result.scalars().all()
    return book

II.单一过滤条件

书写格式:

select(模型类).where(模型类.字段 == 查询的数据)

返回所有过滤结果
- result.scalars().all()

@app.get('/book/{book_id}')
async def get_book(book_id: int,db: AsyncSession = Depends(get_database)):
    result = await db.execute(select(Book).where(Book.book_id == book_id))
    book = result.scalars().all()
    return book

返回单条过滤结果
- result.scalar_one_or_none()

@app.get('/book/{book_id}')
async def get_book(book_id: int,db: AsyncSession = Depends(get_database)):
    result = await db.execute(select(Book).where(Book.book_id == book_id))
    book = result.scalar_one_or_none()  # 返回单个对象或None
    return book

观察两种方法在返回结果的结构上有什么不同？

III.多条件过滤

书写格式:

select(模型类).where(模型类.字段 == 查询的数据,第二个查询条件)

根据作者和时代查询

@app.get('/book/{author}/{era}')
async def get_book(author: str,era: str,db: AsyncSession = Depends(get_database)):
    result = await db.execute(select(Book).where(Book.author == author,Book.era == era))
    book = result.scalars().all()
    return book

IV.like模糊查询

书写格式:

select(模型类).where(模型类.字段.like(通配符))

like操作符通常与两个通配符一起使用：
- %：代表零个、一个或多个字符
- _：代表一个单个字符
- like("%@gmail.com")会匹配所有以"@gmail.com"结尾的字符串，例如"example@gmail.com"。

查询罗开头的作者的图书

@app.get('/booklike')
async def get_book(db: AsyncSession = Depends(get_database)):
    result = await db.execute(select(Book).where(Book.author.like('罗%')))
    book = result.scalars().all()
    return book

V.与或非查询

使用符号

| 代表或
& 代表与
~ 代表非

# 与  多个条件都成立
select(模型类).where((判断条件1) & (判断条件2)) 
# 等价于：
select(模型类).where(判断条件1,判断条件2)  

# 或 满足任意一个条件
select(模型类).where((判断条件1) | (判断条件2))

# 非
select(模型类).where(~(判断条件))

查询id大于4 或时代是清朝的

@app.get('/bookandor')
async def get_book(db: AsyncSession = Depends(get_database)):
    result = await db.execute(select(Book).where((Book.book_id > 4) |  (Book.era == '清朝')))
    book = result.scalars().all()
    return book

VI.in包含查询

书写格式：

select(模型类).where(模型类.字段._in(数据))

查询id包含1234的图书

@app.get('/bookin')
async def get_book(db: AsyncSession = Depends(get_database)):
    result = await db.execute(select(Book).where(Book.book_id.in_([1,2,3,4])))
    book = result.scalars().all()
    return book

10.2.2 聚合计算

书写格式:

select(func.方法(字段))

使用func模块提供的函数进行计算

from sqlalchemy import func

@app.get('/bookagg')
async def get_book(db: AsyncSession = Depends(get_database)):
    # 计算图书的数量
    total_result  = await db.execute(select(func.count(Book.book_id)))
    total_book = total_result.scalars().all()

    # 计算不同时代的图书数量
    era_result = await db.execute(select(Book.era, func.count(Book.book_id)).group_by(Book.era))
    agg_book = era_result.scalars().all()

    # 获取id最大的图书
    max_result = await db.execute(select(Book).order_by(Book.book_id.desc()).limit(1))
    max_book = max_result.scalars().all()

    return {
        'total_book': total_book,
        'agg_book': agg_book,
        'max_book': max_book
    }

10.2.4 分页查询

offset表示跳过的记录数，也就是从第几条数据数据开始，limit 表示返回的记录数，也就是返回多少条数据

分页的固定写法，替换不同数据表

@app.get('/bookpaginate')
async def paginate(page: int = 1,page_size: int = 20,db: AsyncSession = Depends(get_database)):
    """
    基础分页函数

    Args:
        db: 数据库会话
        page: 页码，从1开始
        page_size: 每页数量
        base_query: 基础查询对象
    """
    # 计算跳过的记录数
    skip = (page - 1) * page_size

    # 构建查询
    base_query = select(Book)

    # 获取分页数据  offset表示跳过的记录数，也就是从第几条数据数据开始，limit 表示返回的记录数，也就是返回多少条数据
    stmt = base_query.offset(skip).limit(page_size)
    result = await db.execute(stmt)
    items = result.scalars().all()

    # 获取总数
    count_stmt = select(func.count()).select_from(base_query.subquery())
    total_result = await db.execute(count_stmt)
    total = total_result.scalar_one()

    # 计算总页数
    total_pages = (total + page_size - 1) // page_size

    return {
        "items": items,
        "total": total,
        "page": page,
        "page_size": page_size,
        "total_pages": total_pages,
        "has_previous": page > 1,
        "has_next": page < total_pages
    }

10.2.3 添加数据

需要使用post方法进行数据的提交，借助pycharm的http工具

定义验证类
- 为了保证数据安全，定义一个pydantic数据验证
- 编写一个schema.py文件

from typing import Optional
from pydantic import BaseModel

class BookBase(BaseModel):
    title: str
    author: str
    genre: str
    era: Optional[str] = None

main.py使用add方法添加

from schema import BookBase
@app.post('/book/')
async def create_book(book: BookBase,db: AsyncSession = Depends(get_database)):
    # 获取传递的数据，创建的图书对象  将book.__dict__转为字典，然后对字典数据拆包
    book_obj = Book(**book.__dict__)
    # 添加图书数据
    db.add(book_obj)
    # 提交添加
    await db.commit()
    return book

http文件的代码

### 添加图书数据
POST http://127.0.0.1:8000/book/
Accept: application/json

{
    "title": "aaa",
    "author": "bbb",
    "genre": "ccc",
    "era": "ddd"
}

10.2.4 更新数据

需要使用put方法进行数据的提交，借助pycharm的http工具

定义验证类
- 为了保证数据安全，定义一个pydantic数据验证
- 因为更新可能更新一个或多个字段，所以验证时，允许字段为空
- 编写一个schema.py文件

from typing import Optional
from pydantic import BaseModel

class BookBaseUpdate(BaseModel):
    title: Optional[str] = None
    author: Optional[str] = None
    genre: Optional[str] = None
    era: Optional[str] = None

main.py

from schema import BookBaseUpdate

@app.put('/book/{book_id}')
async def update_book(book_id: int,book: BookBaseUpdate,db: AsyncSession = Depends(get_database)):
    # 获取传递的数据，创建的图书对象  将book.__dict__转为字典，然后对字典数据拆包
    book_obj = Book(**book.__dict__)

    # 获取数据库中的图书对象
    db_book = await db.get(Book, book_id)

    if db_book is None:
        return {'message': '图书不存在'}
    # 更新图书对象 ，判断是否有空数据
    if book_obj.title:
        db_book.title = book_obj.title
    if book_obj.author:
        db_book.author = book_obj.author
    if book_obj.genre:
        db_book.genre = book_obj.genre
    if book_obj.era:
        db_book.era = book_obj.era
    await db.commit()

    return book

http文件

### 更新图书数据
PUT http://127.0.0.1:8000/book/16
Accept: application/json

{
    "title": "111",
    "author": "222"
}

10.2.4 数据删除

main.py文件中使用delete方法

@app.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:
        return {'message': '图书不存在'}
    # 删除图书对象
    await db.delete(db_book)
    await db.commit()

    return {'message': '图书删除成功'}

http文件

### 删除图书数据
DELETE http://127.0.0.1:8000/book/16

10.2 关联表操作

图书表和人物表存在一对多的关联关系，可以使用join关联方法查询数据

书写格式:

await db.execute(select(模型类1,模型类2).join(关联字段))

查询某本图书中包含的人物有哪些

@app.get('/bookcorrelation/{book_id}')
async def get_book_correlation(book_id: int,db: AsyncSession = Depends(get_database)):
    # 获取图书对象
    result = await db.execute(select(Book, Character).join(Character, Book.book_id == Character.book_id).where(Book.book_id == book_id))
    book = result.scalars().all()
    return book