Litestar Repository 测试指南GenericAsyncMockRepository 与 GenericSyncMockRepository 内存 Mock 仓库详解【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar本篇技术指南聚焦 Litestar 仓库Repository抽象层内置的测试利器——litestar.repository.testing模块中的GenericAsyncMockRepository与GenericSyncMockRepository。它们以dict作为存储介质在内存中完整实现了仓库抽象基类的全部 CRUD 接口用于在单元测试中替代真实数据库验证仓储逻辑与业务代码。读完本文你将掌握 mock 仓库的泛型参数化、种子数据注入、审计时间戳自动维护、get_or_create匹配语义等核心用法并能直接照搬到自己的测试代码中。该模块由 docs/reference/repository/testing.rst 通过 Sphinxautomodule指令对外公开其完整实现位于 litestar/repository/testing/generic_mock_repository.py配套的专项测试见 tests/unit/test_repository/test_generic_mock_repository.py。模块定位为什么需要内存 Mock 仓库在服务端应用中Repository 模式将数据访问逻辑与业务逻辑解耦但单元测试如果直接依赖真实数据库SQLite、PostgreSQL 等会带来环境搭建慢、测试互相污染、CI 不稳定等问题。Litestar 的仓库抽象层给出了一个轻量解法在测试中使用基于dict的内存实现让测试只验证仓储接口的行为契约与上层业务逻辑的调用方式不触碰任何真实的持久化设施。GenericAsyncMockRepository与GenericSyncMockRepository正是为此设计的两个类前者继承AbstractAsyncRepository异步接口见 litestar/repository/abc/_async.py方法均为async def后者继承AbstractSyncRepository同步接口见 litestar/repository/abc/_sync.py方法均为普通函数。两个类在行为上完全对称代码中通过__all__统一导出见generic_mock_repository.py第 19 行可单独按需导入from litestar.repository.testing.generic_mock_repository import ( GenericAsyncMockRepository, GenericSyncMockRepository, )说明模块 docstring 明确指出其定位——A repository implementation for tests. Uses adictfor storage.即这是一个仅服务于测试的内存实现不应在生产环境中使用。泛型参数化为每种模型生成独立仓库类mock 仓库通过__class_getitem__支持泛型下标语法例如GenericAsyncMockRepository[UUIDAuthor]。每次用不同模型类型做下标访问时会动态生成一个全新的子类并携带该模型专属的类属性AuthorRepository GenericAsyncMockRepository[UUIDAuthor] BookRepository GenericAsyncMockRepository[UUIDBook]从源码看generic_mock_repository.py第 52-68 行__class_getitem__会返回一个以GenericAsyncMockRepository[ModelName]命名的动态子类并在其上绑定类属性含义collection初始化为空dict作为该模型类型的独立存储model_type记录参数化的模型类型本身_model_has_created_at通过hasattr(item, created_at)探测模型是否具备创建时间戳字段_model_has_updated_at通过hasattr(item, updated_at)探测模型是否具备更新时间戳字段这意味着每种模型类型拥有彼此隔离的collection测试中不会出现不同类型数据互相串扰。这一行为在测试 tests/unit/test_repository/test_generic_mock_repository.py 的test_generic_mock_repository_parametrization中得到验证分别参数化UUIDAuthor与UUIDBook后两者的model_type各自正确。构造参数详解两个 mock 仓库的构造函数签名完全一致def __init__( self, id_factory: Callable[[], Any] uuid4, tz: tzinfo UTC, allow_ids_on_add: bool False, **_: Any, ) - None参数默认值作用id_factoryuuid4uuid.uuid4生成新记录主键的可调用对象。当allow_ids_on_addFalse时add/add_many会调用它为主键字段赋值需要整型自增主键时可替换为自定义工厂函数tzUTC生成审计时间戳created_at/updated_at所用的时区。注意_now()内部先按tz生成 aware datetime再调用replace(tzinfoNone)去掉时区信息得到 naive datetime 后写入模型allow_ids_on_addFalse是否允许调用方在add时传入已带主键的实例。为False时若实例已存在主键值将抛出ConflictError并强制由id_factory重新生成主键**_—构造函数接受任意多余关键字参数并忽略保证与真实仓库实现可能接收连接、会话等依赖在接口上保持兼容测试数据准备seed_collection 与 clear_collection测试最常用的手法是先向仓库预置一批数据。两个类都提供了两个类方法# 批量注入初始数据以实例主键作为 dict 键 AuthorRepository.seed_collection([ UUIDAuthor(iduuid4(), nameAgatha Christie, dobdate(1890, 9, 15)), UUIDAuthor(iduuid4(), nameLeo Tolstoy, dobdate(1828, 9, 9)), ]) # 清空该类仓库的 collection用于测试间隔离 AuthorRepository.clear_collection()seed_collection(instances)遍历实例通过基类的get_id_attribute_value取主键写入cls.collectiongeneric_mock_repository.py第 391-399 行clear_collection()将cls.collection重置为空dict第 401-404 行。由于collection是类属性在同一测试会话中多次参数化同一模型会共享存储因此测试套件通常配合 pytest fixture 在每个用例前后调用clear_collection()来保证隔离参见测试文件中的author_repository_typefixture 用法。完整 CRUD API 一览mock 仓库完整实现了抽象基类声明的全部仓储接口。下表汇总了 16 个公开方法及其行为契约方法签名要点行为说明add(data)async def add(self, data: ModelT) - ModelT追加单条默认校验主键冲突、生成主键、写入审计时间戳add_many(data)Iterable[ModelT] - list[ModelT]批量追加整批使用同一个now时间戳delete(item_id)Any - ModelT删除指定主键记录不存在时抛NotFoundErrordelete_many(item_ids)list[Any] - list[ModelT]逐个按主键查找并删除返回被删实例列表get(item_id)Any - ModelT按主键取单条不存在时抛NotFoundErrorget_one(**kwargs)关键字过滤返回满足过滤条件的第一条无结果抛NotFoundErrorget_one_or_none(**kwargs)关键字过滤返回第一条或None不抛异常get_or_create(match_fields, **kwargs)返回(ModelT, bool)按匹配字段查找存在则就地合并更新并返回(实例, False)否则创建并返回(实例, True)count(*filters, **kwargs)- int返回过滤后的记录总数忽略分页语义exists(*filters, **kwargs)- bool基于count判断是否存在list(*filters, **kwargs)- list[ModelT]返回过滤后的全部实例list_and_count(*filters, **kwargs)- tuple[list[ModelT], int]同时返回列表与总数update(data)ModelT - ModelT按主键找到存储实例用传入实例的字段覆盖不存在抛NotFoundErrorupdate_many(data)list[ModelT] - list[ModelT]批量更新逐条校验存在性使用同一nowupsert(data)ModelT - ModelT主键已存在则update否则addupsert_many(data)list[ModelT] - list[ModelT]按主键拆分待更新/待新增两批分别处理返回合并结果需要留意的是list、count等方法虽然接受*filters: FilterTypes位置参数即 litestar/repository/filters.py 中定义的分页/过滤类型但内存实现中仅应用**kwargs的字段等值过滤FilterTypes过滤条件在当前 mock 实现中不产生实际效果——测试时若需验证分页等过滤逻辑应使用真实仓库实现或直接构造断言。主键冲突与未携带主键约定add/add_many在allow_ids_on_addFalse默认时会先检查实例主键if self.allow_ids_on_add is False and self.get_id_attribute_value(data) is not None: raise ConflictError(add() received identified item.)即向仓库添加一个已经带主键的实例会直接抛ConflictError。这是有意设计的约束——模仿了真实数据库中插入已存在主键记录的冲突语义。对应测试test_repo_raises_conflict_if_add_with_id与test_repo_raises_conflict_if_add_many_with_id分别覆盖了单条与批量场景。kwargs 过滤语义AND 等值匹配list、get_one等方法的**kwargs过滤委托给filter_collection_by_kwargsgeneric_mock_repository.py第 372-389 行其语义为if all(getattr(item, name) value for name, value in kwargs.items()): new_collection[item.id] item多个关键字之间是AND且关系即实例必须同时满足所有字段值相等才会命中过滤依据是getattr按属性名取值属性不存在时抛RepositoryError测试test_generic_mock_repository_raises_repository_exception_if_named_attribute_doesnt_exist用不存在的cricketball验证了这一点字段取值是严格等值比较注意区分类型测试test_generic_mock_repository_filter_collection_by_kwargs_and_semantics展示了name与dob双条件 AND 语义下 0 命中的场景。审计时间戳的自动维护当参数化的模型具备created_at/updated_at属性时__class_getitem__阶段通过hasattr探测mock 仓库会自动维护这两个字段add/add_many同时写入created_at与updated_atdo_createdTrueupdate/update_many仅刷新updated_at模型没有这两个属性时任何写入操作都不会触碰它们测试test_does_not_set_created_updated验证了这一点。时间戳统一由_now()生成datetime.now(tzself.tz).replace(tzinfoNone)。测试test_sets_created_updated_on_add断言了 add 后两个字段被填充test_sets_updated_on_update断言 update 后updated_at单调递增。这意味着使用带审计字段的模型如 advanced-alchemy 的UUIDAuditBase时mock 仓库的行为与真实数据库保持一致可直接用于校验记录创建/修改时间的业务断言。get_or_create 与 match_fields 匹配策略get_or_create(match_fieldsNone, **kwargs)是较为复杂的接口其匹配逻辑为generic_mock_repository.py第 190-220 行取match_fields or self.match_fields类属性match_fields可在参数化后覆盖默认None若match_fields为字符串则包装成单元素列表若提供了match_fields仅从kwargs中抽取这些字段名对应的值构成匹配过滤器None值被剔除否则用全部kwargs匹配命中已有实例用kwargs中非空且不同的字段值就地更新该实例返回(实例, False)未命中以self.model_type(**kwargs)创建新实例并add返回(实例, True)。测试test_get_or_create与test_get_or_create_match_fields分别验证了按全部字段匹配不新增、指定match_fields[random_column]时仅按该字段判定存在性其余字段用于更新的行为。这在需要按业务唯一键获取或创建的测试场景中非常实用。与抽象基类的继承关系两个 mock 类都继承自仓库抽象基类因此天然获得以下基础设施id_attribute类属性默认id标识模型主键字段名可覆盖get_id_attribute_value(item, id_attributeNone)从实例上读取主键值set_id_attribute_value(item_id, item, id_attributeNone)给实例写入主键值check_not_found(item_or_none)值为None时抛NotFoundError见 litestar/repository/abc/_async.py 第 262 行附近。mock 类正是通过这些助手实现get、delete、update等方法的未找到即抛错语义。例如get的实现async def get(self, item_id: Any, **kwargs: Any) - ModelT: return self._find_or_raise_not_found(item_id) def _find_or_raise_not_found(self, item_id: Any) - ModelT: return self.check_not_found(self.collection.get(item_id))异常体系mock 仓库可能抛出的异常统一由 litestar/repository/exceptions.py 导出异常触发场景ConflictErrorallow_ids_on_addFalse时向add/add_many传入已带主键的实例NotFoundErrorget/delete/update等按主键操作时目标记录不存在或get_one无结果RepositoryErrorfilter_collection_by_kwargs中过滤属性在模型上不存在值得注意的底层细节exceptions.py采用可选依赖优雅降级策略——优先从advanced_alchemy.exceptions导入IntegrityError被重命名为ConflictError再导出若未安装 advanced-alchemy 则回退到仓库内部的 litestar/repository/_exceptions.py 实现。这保证了 mock 仓库在不同依赖环境下都能正常工作且测试断言时引用的异常类型与生产仓库完全一致。实战完整测试示例结合测试文件 tests/unit/test_repository/test_generic_mock_repository.py 的用法一个典型的测试流程如下from uuid import uuid4 from datetime import date from litestar.repository.exceptions import ConflictError, NotFoundError from litestar.repository.testing.generic_mock_repository import ( GenericAsyncMockRepository, GenericSyncMockRepository, ) # 1. 参数化为 Author 模型生成独立的 mock 仓库类 AuthorRepository GenericAsyncMockRepository[UUIDAuthor] # 2. 预置种子数据 AuthorRepository.seed_collection([ UUIDAuthor(iduuid4(), nameAgatha Christie, dobdate(1890, 9, 15)), UUIDAuthor(iduuid4(), nameLeo Tolstoy, dobdate(1828, 9, 9)), ]) # 3. 实例化并进行 CRUD 断言 async def test_crud() - None: repo AuthorRepository() # 等值过滤AND 语义 assert len(await repo.list(nameLeo Tolstoy)) 1 # 单条查询无结果抛 NotFoundError try: await repo.get_one(nameNobody) except NotFoundError: pass # 新增默认不允许携带主键 try: await repo.add(UUIDAuthor(iduuid4(), nameNew)) except ConflictError: pass # 统计 assert await repo.count() 2 # 用例结束清理避免类属性存储跨用例污染 AuthorRepository.clear_collection()测试文件中的 fixture 写法同样值得借鉴——用repository_typefixture 参数化[GenericAsyncMockRepository, GenericSyncMockRepository]配合tests/helpers.py中的maybe_async辅助函数可以让同一套断言同时覆盖异步与同步两种实现大幅提升测试复用率。小结litestar.repository.testing模块为 Litestar 仓库抽象层提供了开箱即用的内存测试实现通过泛型参数化获得按模型隔离的dict存储通过seed_collection/clear_collection便捷管理测试数据完整覆盖全部 CRUD 接口并忠实复刻了ConflictError、NotFoundError、审计时间戳等真实数据库行为。在需要验证仓储契约、编写不依赖数据库的服务层测试时它是成本最低、语义最贴近生产仓库的选择。如需继续深入了解仓库抽象层可参考 docs/reference/repository/index.rst 下的其他页面抽象基类定义见 abc、过滤类型见 filters、异常体系见 exceptions、路由处理器集成见 handlers。【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考