Python Web个人学习记录04

Python Web Day04ORM:用于面向对象编程语言和关系型数据库之间建立映射pip installsqlalchemy[asyncio]aiomysqlORM建表ORM创建数据库异步引擎建立数据库CREATE DATABASE FastAPI_first CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;链接数据库fromsqlalchemy.ext.asyncioimportcreate_async_engine#创建异步引擎ASYNC_DATABASE_URLmysqlaiomysql://root:rootlocalhost:3307/FastAPI_first?charsetutf8async_enginecreate_async_engine(ASYNC_DATABASE_URL,echoTrue,# 输出SQL日志pool_size10,#设置连接池活跃的连接数max_overflow10#允许额外的连接数)ORM定义模型类基类继承DeclarativeBase定义数据库表对应的模型名classBase(DeclarativeBase):create_time:Mapped[datetime]mapped_column(DateTime,insert_defaultfunc.now(),defaultfunc.now,comment创建时间)update_time:Mapped[datetime]mapped_column(DateTime,insert_defaultfunc.now(),defaultfunc.now,onupdatefunc.now,comment修改时间)classBook(Base):__tablename__bookid:Mapped[int]mapped_column(primary_keyTrue,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创建数据库表从连接池获取异步连接开启事务执行ORMFastAPI应用启动时创建数据库表#建表定义函数建表 FastAPI启动时调用建表的函数asyncdefcreate_tables():# 获取异步引擎 创建事务asyncwithasync_engine.begin()asconn:awaitconn.run_sync(Base.metadata.create_all)#使用Base模型类源数据asynccontextmanagerasyncdefapp_lifespan(app:FastAPI):awaitcreate_tables()# 启动时执行yield# 应用运行阶段必须先定义生命周期函数再创建appappFastAPI(lifespanapp_lifespan)在路由当中使用ORM核心创建依赖项使用Depends注入到处理函数# 定义一个查询功能接口AsyncSessionLocalasync_sessionmaker(bindasync_engine,#绑定数据库引擎class_AsyncSession,#指定会话类expire_on_commitFalse#提交后会话不过期不会重新查询数据库)asyncdefget_datebase():asyncwithAsyncSessionLocal()assession:try:yieldsession#返回数据库会话给路由处理函数awaitsession.commit()#提交事务exceptException:awaitsession.rollback()#有异常raisefinally:awaitsession.close()# 关闭会话app.get(/book/books)asyncdefget_book_list(db:AsyncSessionDepends(get_datebase)):# 查询resultawaitdb.execute(select(Book))bookresult.scalars().all()returnbook增删改查查resultawaitdb.execute(select(模型类))#返回一个ORM对象获取所有数据result.scalars().all()获取单条数据result.scalars().first()get(模型类主键值)app.get(/book/books)asyncdefget_book_list(db:AsyncSessionDepends(get_datebase)):# 查询# result await db.execute(select(Book))# # book result.scalars().all()# book result.scalars().first()bookawaitdb.get(Book,2)returnbook查询条件学过数据库的应该都接触过条件比较判断 ; ; ;等模糊查询like %零个一个或多个字符 _一个单个字符与非查询|~包含查询in_()