表数据

1.查询

测试接口

2.条件查询
比较判断查询


测试接口


3.模糊查询 ，与非查询，包含查询

完整代码

from unittest import result

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

app = FastAPI()

# 1.创建异步引擎
#ASYNC_DATAASE_URL为数据库的地址
ASYNC_DATAASE_URL = "mysql+aiomysql://root:630229@localhost:3306/fast_api_db?charset=utf8"
async_engine = create_async_engine(
    ASYNC_DATAASE_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=datetime.now,
        comment="创建时间"
    )
    update_time: Mapped[datetime] = mapped_column(
        DateTime,
        insert_default=func.now(),
        onupdate=func.now(),
        default=datetime.now,
        comment="修改时间"
    )


class Book(Base):
    __tablename__ = "book"
    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(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"}


# 路由匹配中使用 ORM ,创建依赖项，使用 Depends注入到处理函数
# 需求：查询功能的接口，查询图书=》依赖注入：创建依赖项获取数据库会话+Depends注入路由处理函数

# 4.创建异步会话工厂
AsyncSessionLocal = async_sessionmaker(
    bind=async_engine, # 绑定异步引擎
    class_=AsyncSession, # 设置会话类为异步会话
    expire_on_commit=False # 设置会话提交时是否自动刷新
)

# 5.依赖项：获取数据库会话
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() # 关闭会话

# 6.路由：查询图书（获取所有）
@app.get("/book/books")
async def get_books_list(session: AsyncSession = Depends(get_database)):
    result = await session.execute(select(Book)) #查询返回一个ORM对象
    books = result.scalars().all() #获取所有
    return books

# 7.路由：查询图书（获取第一个）
@app.get("/book/firstbook")
async def get_books_list(session: AsyncSession = Depends(get_database)):
    result = await session.execute(select(Book)) #查询返回一个ORM对象
    book = result.scalars().first() #获取第一个
    return book


# 7.路由：查询图书（获取单条数据根据ID查询）
@app.get("/book/IDbook")
async def get_books_list(session: AsyncSession = Depends(get_database)):
    book = await session.get(Book,3) #获取单条数据根据ID查询
    return book


# 8.需求：路径参数，书籍id
@app.get("/book/get_book/{book_id}")
async def get_book(book_id: int,session: AsyncSession = Depends(get_database)):
     res = await session.execute(select(Book).where(Book.id == book_id))
     book = res.scalar_one_or_none()
     return book

# 9.需求：条件查询，价格大于等于
@app.get("/book/search_book")
async def get_books_list(book_price: int,session: AsyncSession = Depends(get_database)):
    res = await session.execute(select(Book).where(Book.price >= book_price))
    books = res.scalars().all()
    return books

#10 模糊查询，与非查询，包含查询
@app.get("/book/like_book")
async def get_books_list(session: AsyncSession = Depends(get_database)):
    #需求 ：作者author以周开头查询 % _
    #like() 模糊查询：% 任意个字符; _一个单个字符
    #res = await session.execute(select(Book).where(Book.author.like("周%")))
    #res = await session.execute(select(Book).where(Book.author.like("周_")))

    # & | ~ 逻辑运算符
    # res = await session.execute(select(Book).where((Book.author.like("周%")) % (Book.price>30)))
    # res = await session.execute(select(Book).where((Book.author.like("周%")) | (Book.price > 30)))

    #in() 包含
    res = await session.execute(select(Book).where(Book.author.in_(["周树人", "周国平"])))
    books = res.scalars().all()
    return books