基于FastAPI、Vue、PostgreSQL、Peewee和SmartAdmin的Toy项目设计源码

1：注册插件 SqlalchemyPluginClient(app=app, settings=SqlalchemyPluginClient.SqlalchemySettings( )) 2：使用依赖注入方式 @app.get("/async_session_depends") async def sync_route(async_session: AsyncSession = Depends(get_async_session_depends)): async with get_async_session_context() as async_session: pass logger.info(await SysCategoryService.get_sys_category_items_all(db_session=async_session)) print("ssssssssssssssss", await SysCategoryService.get_sys_category_items_all(db_session=async_session)) # eventsaaa.async_emit_evnet("app_mention", "我是发送的数据信息") return 'ok' 3：使用上下文的方式（依然全局的request） @app.get("/async_session_with") async def sync_route(): async with get_async_session_context() as async_session: pass logger.info(await SysCategoryService.get_sys_category_items_all(db_session=async_session)) print("ssssssssssssssss",await SysCategoryService.get_sys_category_items_all(db_session=async_session)) # eventsaaa.async_emit_evnet("app_mention", "我是发送的数据信息") return 'ok' 使用装饰器方式的话，需要处理比较多，需要依赖一个中间件以及上下文对象处理 在中间件或获取上下文对象 ................................................ AsyncSession_var: ContextVar[AsyncSession] = ContextVar("AsyncSession") async_session: AsyncSession = bind_contextvar(AsyncSession_var) class AsyncSessionLoadMiddleware: ''' 此类的中间件无法读取响应报文的内容 ''' pass def __init__(self, app: ASGIApp) -> None: self.app = app def bind_to_request_state(self, request: Request, **kwargs): """ Takes in a set of kwargs and binds them to gziprequest state """ for key, value in kwargs.items(): setattr(request.state, key, value) async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: if scope["type"] != "http": await self.app(scope, receive, send) return # 解析当前的请求体 request = Request(scope, receive=receive) # 开启中间件的时候创建对应的会话对象 token = AsyncSession_var.set(request.app.async_session_maker()) try: await self.app(scope, receive, send) finally: # 释放 AsyncSession_var.reset(token) ................................................ 然后在装饰器使用对应的 token = AsyncSession_var.set(request.app.async_session_maker())生成的对象，也就是下面中 async_session: AsyncSession = bind_contextvar(AsyncSession_var)中async_session对象 然后在对应的查询中引入上面的async_session对象，并启用装饰： ................................................ class SysCategoryService: @staticmethod @Transactional() async def get_sys_category_items_all22222(): print("查询async_scoped_session装饰器！！！？？？？",id(async_scoped_session)) query = select(SysCategory) data = (await async_scoped_session.execute(query)) return [it.name for it in data.scalars().all()] ................................................ 然后在对应路由逻辑中调用！ @app.get("/async_session_zhuangshiqi") async def sync_route(): logger.info(await SysCategoryService.get_sys_category_items_all22222()) # print("ssssssssssssssss",await SysCategoryService.get_sys_category_items_all()) # eventsaaa.async_emit_evnet("app_mention", "我是发送的数据信息") return 'ok' # 全局变量定义的方式使用： 1:定义变量： # 自定义全局的变量的方式来使用对应的对象 global_async_engine = None global_async_session: Union[Callable, AsyncSession, AsyncGenerator, sessionmaker] = lambda: None ................................................ 2：初始化全局变量： @app.on_event("startup") async def startup_event(): try: # 创建异步会话对象 global global_async_engine global global_async_session self.async_engine = create_async_engine(url=self.settings.ASYNC_SQLALCHEMY_DATABASE_URI, echo=self.settings.SQLALCHEMY_DATABASE_ECHO, pool_recycle=self.settings.SQLALCHEMY_POOL_RECYCLE, max_overflow=self.settings.SQLALCHEMY_MAX_OVERFLOW, pool_pre_ping=self.settings.SQLALCHEMY_POOL_PRE_PING, pool_size=self.settings.SQLALCHEMY_POOL_SIZE, ) self.async_session_maker = sessionmaker(bind=self.async_engine, class_=AsyncSession, expire_on_commit=False) app.async_session = self app.async_session_maker = self.async_session_maker # 全局变量的定义处理 global_async_engine = self.async_engine global_async_session = self.async_session_maker except Exception as e: raise e ................................................ 3：定义上下文对象 ................................................ @asynccontextmanager async def get_async_session_context_globa() -> AsyncGenerator: async_session: AsyncSession = global_async_session() try: yield async_session # Exception during reset or similar # Traceback (most recent call last): # 使用自动提交在某些情况下 会突发的asyncio.exceptions.CancelledError异常 # await async_session.commit() except OperationalError as ex: pass raise ex except SQLAlchemyError as ex: await async_session.rollback() raise ex finally: await async_session.close() ................................................ 4： 使用全局变量 @app.get("/async_session_with") async def sync_route(): async with get_async_session_context_globa() as async_session: pass logger.info(await SysCategoryService.get_sys_category_items_all(db_session=async_session)) print("ssssssssssssssss",await SysCategoryService.get_sys_category_items_all(db_session=async_session)) # eventsaaa.async_emit_evnet("app_mention", "我是发送的数据信息") return 'ok' ................................................