1.cache/news_cache.pyNEWS_LIST_PREFIX news_list: # 写入缓存-新闻列表 key news_list:分类id:页码:每页数量 列表数据 过期时间 async def set_cache_news_list(category_id: Optional[int], page: int, size: int, news_list: List[Dict[str, Any]], expire: int 1800): category_part category_id if category_id is not None else all key f{NEWS_LIST_PREFIX}{category_part}:{page}:{size} return await set_cache(key, news_list, expire) # 读取缓存-新闻列表 key news_list:分类id:页码:每页数量 async def get_cache_news_list(category_id: Optional[int], page: int, size: int): category_part category_id if category_id is not None else all key f{NEWS_LIST_PREFIX}{category_part}:{page}:{size} return await get_json_cache(key)2. 改造crud/news.py# 获取新闻列表 async def get_news_list( db: AsyncSession, category_id: int, skip: int 0, limit: int 10 ): # 先尝试从缓存获取新闻列表 # 跳过的数量skip (页码 - 1) * 每页数量 - 页码 跳过的数量 // 每页数量 1 # await get_cache_news_list(分类id, 页码, 每页数量) page skip // limit 1 cached_list await get_cache_news_list(category_id, page, limit) # 缓存数据 json if cached_list: # return cached_list # 要的是 ORM return [News(**item) for item in cached_list] # 查询的是指定分类下的所有新闻 stmt select(News).where(News.category_id category_id).offset(skip).limit(limit) result await db.execute(stmt) news_list result.scalars().all() # 写入缓存 if news_list: # 先把 ORM 数据 转换 字典才能写入缓存 # ORM 转成 Pydantic再转为 字典 # by_aliasFalse 不适用别名保存 Python 风格因为 Redis 数据是给后端用的 news_data [ NewsItemBase.model_validate(item).model_dump(modejson, by_aliasFalse) for item in news_list ] await set_cache_news_list(category_id, page, limit, news_data) return news_list