核心功能：
- 添加图书（书名、作者、ISBN、出版年份）
- 搜索图书（按书名、作者）
- 更新图书信息
- 删除图书
- 获取图书统计（总数、按作者分组）
技术栈：
- SQLModel数据库集成
- 基础查询和过滤
- 简单数据统计
- 错误处理
挑战点：数据库操作、查询过滤、数据关系
遇到的问题与解决方法
-
在编写搜索图书这个接口时，题目要求是根据书名和作者来搜索，那它的路径参数应该要怎么编写？
-
题目要求是按书名、作者来搜索，并非是id，可使用SQLModel中的select语句
-
先导入查询工具 from sqlmodel import select
-
在定义的search_books方法中，传入两个可选参数title和author
-
创建一个数据库会话对象session
-
查询所有图书，并使用一个statement来接收
-
再根据传入的可选参数，使用contains()进行模糊匹配
statement=statement.where(Book.author.contains(author)) -
执行查询并返回结果
-
-
用什么方法来进行按照书名和作者搜索？
contains()是一个常用的方法，用于判断某个集合、字符串或其他数据结构中是否包含指定的元素，返回值通常为布尔类型 -
使用数据库怎么对图书进行信息更新或删除？
定义一个变量book_id,用来接收传入的要进行操作的图书的id,再使用 session.get(Book，book_id),来查询要找的图书；若不为空：
updated_data = updated_book.model_dump(exclude_unset=True)排除未传入的字段，只更新传入的非空字段，保留原字段
由于updated_data是字典
.items() 返回键值对的可迭代对象，例如：[('title', '新标题'), ('price', 30.0)]
再使用：setattr(db_book, key, value)即为db_book.key = value
from fastapi import FastAPI, HTTPException from sqlmodel import SQLModel, Field, create_engine, Session, select, func from datetime import datetime from typing import List, Optional, Dict, Any import traceback import logging# 设置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__)app = FastAPI(title="图书管理系统",description="通过使用数据库来进行对图书的管理",version="1.0.0" )class Book(SQLModel, table=True):id: Optional[int] = Field(default=None, primary_key=True)book_name: str = Field(index=True)author: str = Field(index=True)ISBN: Optional[str] = Field(default=None)publication_year: Optional[int] = Field(default=None)add_time: datetime = Field(default_factory=datetime.now)sqlite_file_name = "books.db" sqlite_url = f"sqlite:///{sqlite_file_name}"connect_args = {"check_same_thread": False} engine = create_engine(sqlite_url, connect_args=connect_args)def create_db_and_tables():"""创建数据库和表"""try:SQLModel.metadata.create_all(engine)logger.info("数据库和表创建成功")except Exception as e:logger.error(f"创建数据库和表时出错: {e}")traceback.print_exc()@app.on_event("startup") def on_startup():create_db_and_tables()# 添加图书 - 添加详细的错误处理 @app.post("/books/", summary="添加图书", response_model=Book) def add_book(book: Book):try:logger.info(f"接收到图书数据: {book.dict()}")with Session(engine) as session:session.add(book)session.commit()session.refresh(book)logger.info(f"图书添加成功，ID: {book.id}")return bookexcept Exception as e:logger.error(f"添加图书时出错: {e}")traceback.print_exc()raise HTTPException(status_code=500,detail=f"添加图书失败: {str(e)}")# 搜索图书(根据书名和作者) @app.get("/books/",summary="搜索图书(根据书名和作者)",response_model=List[Book] ) def search_books(title: Optional[str] = None,author: Optional[str] = None ):with Session(engine) as session:statement = select(Book)if title:statement = statement.where(Book.book_name.contains(title))if author:statement = statement.where(Book.author.contains(author))results = session.exec(statement)books = results.all()if not books:raise HTTPException(status_code=404, detail="未找到匹配的书")return books# 更新图书信息 @app.put("/books/{book_id}",summary="更新图书信息",description="根据id选择要更新的图书",response_model=Book ) def update_book(book_id: int, updated_book: Book):with Session(engine) as session:db_book = session.get(Book, book_id)if not db_book:raise HTTPException(status_code=404, detail="未找到该书")updated_data = updated_book.dict(exclude_unset=True)for key, value in updated_data.items():setattr(db_book, key, value)session.add(db_book)session.commit()session.refresh(db_book)return db_book# 删除图书 @app.delete("/books/{book_id}",summary="删除图书",description="根据id选择要删除的图书" ) def delete_book(book_id: int):with Session(engine) as session:db_book = session.get(Book, book_id)if not db_book:raise HTTPException(status_code=404, detail="未找到该书")session.delete(db_book)session.commit()return {"status": "success","message": f"序号为{book_id}的图书已删除"}# 获取图书统计（总数、按作者分组） @app.get("/stats/",summary="图书统计",description="统计图书总数，按作者分组" ) def get_total_books():with Session(engine) as session:total_books = session.exec(select(func.count(Book.id))).first()author_stats = session.exec(select(Book.author, func.count(Book.id)).group_by(Book.author)).all()return {"total_books": total_books,"books_by_author": {author: count for author, count in author_stats}}
)
