Python Web Day04

ORM:用于面向对象编程语言和关系型数据库之间建立映射

pip install"sqlalchemy[asyncio]"aiomysql

ORM建表

ORM创建数据库异步引擎

建立数据库

CREATE DATABASE FastAPI_first CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

链接数据库

fromsqlalchemy.ext.asyncioimportcreate_async_engine

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

ORM定义模型类

基类，继承DeclarativeBase

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

classBase(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="修改时间")classBook(Base):__tablename__="book"id:Mapped[int]=mapped_column(primary_key=True,comment="书籍")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="出版社")

ORM创建数据库表

从连接池获取异步连接，开启事务执行ORM

FastAPI应用启动时，创建数据库表

#建表：定义函数建表 FastAPI启动时调用建表的函数asyncdefcreate_tables():# 获取异步引擎 创建事务asyncwithasync_engine.begin()asconn:awaitconn.run_sync(Base.metadata.create_all)#使用Base模型类源数据@asynccontextmanagerasyncdefapp_lifespan(app:FastAPI):awaitcreate_tables()# 启动时执行yield# 应用运行阶段

必须先定义生命周期函数，再创建app

app=FastAPI(lifespan=app_lifespan)

在路由当中使用ORM

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

# 定义一个查询功能接口AsyncSessionLocal=async_sessionmaker(bind=async_engine,#绑定数据库引擎class_=AsyncSession,#指定会话类expire_on_commit=False#提交后会话不过期，不会重新查询数据库)asyncdefget_datebase():asyncwithAsyncSessionLocal()assession:try:yieldsession#返回数据库会话给路由处理函数awaitsession.commit()#提交事务exceptException:awaitsession.rollback()#有异常raisefinally:awaitsession.close()# 关闭会话@app.get("/book/books")asyncdefget_book_list(db:AsyncSession=Depends(get_datebase)):# 查询result=awaitdb.execute(select(Book))book=result.scalars().all()returnbook

增删改查

查：

result=awaitdb.execute(select(模型类))#返回一个ORM对象

获取所有数据：

result.scalars().all()

获取单条数据

result.scalars().first()get(模型类，主键值)

@app.get("/book/books")asyncdefget_book_list(db:AsyncSession=Depends(get_datebase)):# 查询# result = await db.execute(select(Book))# # book = result.scalars().all()# book = result.scalars().first()book=awaitdb.get(Book,2)returnbook

查询条件：（学过数据库的应该都接触过）