实现向量搜索应用层的数据传输对象(DTOs),用于在不同层之间传输数据。
任务要求:
application/vector_search/dtos.py,定义 DocumentDTO定义了三个主要的 DTO 类:
DocumentDTO - 文档数据传输对象
from_entity() - 从领域实体创建 DTOfrom_dict() - 从字典创建 DTOto_entity() - 转换为领域实体to_dict() - 转换为字典SearchResultDTO - 搜索结果数据传输对象
from_entity() - 从领域实体创建 DTOfrom_dict() - 从字典创建 DTOto_dict() - 转换为字典SearchResponseDTO - 搜索响应数据传输对象
from_dict() - 从字典创建 DTOto_dict() - 转换为字典完整的单元测试套件,包含 24 个测试用例:
TestDocumentDTO 类(13 个测试):
test_from_entity_basic - 测试从领域实体创建 DTO(基本情况)test_from_entity_with_embedding - 测试包含嵌入向量的转换test_from_entity_without_embedding - 测试不包含嵌入向量的转换test_from_entity_with_score - 测试包含分数的转换test_to_entity_basic - 测试将 DTO 转换为领域实体(基本情况)test_to_entity_with_embedding - 测试包含嵌入向量的转换test_to_dict_basic - 测试将 DTO 转换为字典(基本情况)test_to_dict_with_optional_fields - 测试包含可选字段的转换test_from_dict_basic - 测试从字典创建 DTO(基本情况)test_from_dict_with_optional_fields - 测试包含可选字段的转换test_roundtrip_entity_to_dto_to_entity - 测试实体 -> DTO -> 实体的往返转换test_roundtrip_dict_to_dto_to_dict - 测试字典 -> DTO -> 字典的往返转换test_metadata_is_copied - 测试元数据被复制而不是引用TestSearchResultDTO 类(5 个测试):
test_from_entity_basic - 测试从领域实体创建搜索结果 DTOtest_from_entity_with_embedding - 测试包含嵌入向量的转换test_to_dict - 测试将搜索结果 DTO 转换为字典test_from_dict - 测试从字典创建搜索结果 DTOtest_roundtrip_dict_to_dto_to_dict - 测试往返转换TestSearchResponseDTO 类(6 个测试):
test_to_dict_basic - 测试将搜索响应 DTO 转换为字典(基本情况)test_to_dict_with_timing - 测试包含耗时的转换test_from_dict_basic - 测试从字典创建搜索响应 DTOtest_from_dict_with_timing - 测试包含耗时的转换test_roundtrip_dict_to_dto_to_dict - 测试往返转换test_empty_results - 测试空结果列表添加了 DTOs 的完整文档,包括:
from_entity() 和 to_entity()embedding 字段可选,通过 include_embedding 参数控制score 字段可选,用于搜索结果场景took_ms 字段可选,用于性能监控to_dict() 和 from_dict() 方法支持 JSON 序列化Timestamp.to_iso_string() 和 Timestamp.from_iso_string() 进行转换所有 24 个测试用例全部通过:
========================= test session starts =========================
collected 24 items
tests/unit/application/vector_search/test_dtos.py::TestDocumentDTO::test_from_entity_basic PASSED [ 4%]
tests/unit/application/vector_search/test_dtos.py::TestDocumentDTO::test_from_entity_with_embedding PASSED [ 8%]
tests/unit/application/vector_search/test_dtos.py::TestDocumentDTO::test_from_entity_without_embedding PASSED [ 12%]
tests/unit/application/vector_search/test_dtos.py::TestDocumentDTO::test_from_entity_with_score PASSED [ 16%]
tests/unit/application/vector_search/test_dtos.py::TestDocumentDTO::test_to_entity_basic PASSED [ 20%]
tests/unit/application/vector_search/test_dtos.py::TestDocumentDTO::test_to_entity_with_embedding PASSED [ 25%]
tests/unit/application/vector_search/test_dtos.py::TestDocumentDTO::test_to_dict_basic PASSED [ 29%]
tests/unit/application/vector_search/test_dtos.py::TestDocumentDTO::test_to_dict_with_optional_fields PASSED [ 33%]
tests/unit/application/vector_search/test_dtos.py::TestDocumentDTO::test_from_dict_basic PASSED [ 37%]
tests/unit/application/vector_search/test_dtos.py::TestDocumentDTO::test_from_dict_with_optional_fields PASSED [ 41%]
tests/unit/application/vector_search/test_dtos.py::TestDocumentDTO::test_roundtrip_entity_to_dto_to_entity PASSED [ 45%]
tests/unit/application/vector_search/test_dtos.py::TestDocumentDTO::test_roundtrip_dict_to_dto_to_dict PASSED [ 50%]
tests/unit/application/vector_search/test_dtos.py::TestDocumentDTO::test_metadata_is_copied PASSED [ 54%]
tests/unit/application/vector_search/test_dtos.py::TestSearchResultDTO::test_from_entity_basic PASSED [ 58%]
tests/unit/application/vector_search/test_dtos.py::TestSearchResultDTO::test_from_entity_with_embedding PASSED [ 62%]
tests/unit/application/vector_search/test_dtos.py::TestSearchResultDTO::test_to_dict PASSED [ 66%]
tests/unit/application/vector_search/test_dtos.py::TestSearchResultDTO::test_from_dict PASSED [ 70%]
tests/unit/application/vector_search/test_dtos.py::TestSearchResultDTO::test_roundtrip_dict_to_dto_to_dict PASSED [ 75%]
tests/unit/application/vector_search/test_dtos.py::TestSearchResponseDTO::test_to_dict_basic PASSED [ 79%]
tests/unit/application/vector_search/test_dtos.py::TestSearchResponseDTO::test_to_dict_with_timing PASSED [ 83%]
tests/unit/application/vector_search/test_dtos.py::TestSearchResponseDTO::test_from_dict_basic PASSED [ 87%]
tests/unit/application/vector_search/test_dtos.py::TestSearchResponseDTO::test_from_dict_with_timing PASSED [ 91%]
tests/unit/application/vector_search/test_dtos.py::TestSearchResponseDTO::test_roundtrip_dict_to_dto_to_dict PASSED [ 95%]
tests/unit/application/vector_search/test_dtos.py::TestSearchResponseDTO::test_empty_results PASSED [100%]
========================= 24 passed in 0.40s ==========================
✅ 已满足 - DTOs 提供了应用层和表现层之间的数据传输机制,支持领域实体的转换。
✅ 已满足 - DTOs 位于 src/application/vector_search/ 模块中,与向量搜索功能相关的其他组件组织在一起。
from src.application.vector_search.dtos import DocumentDTO
from src.domain.vector_search.entities import Document
# 创建领域实体
doc = Document(...)
# 转换为 DTO(不包含嵌入向量)
dto = DocumentDTO.from_entity(doc)
# 转换为 DTO(包含嵌入向量)
dto = DocumentDTO.from_entity(doc, include_embedding=True)
# 转换为 DTO(包含搜索分数)
dto = DocumentDTO.from_entity(doc, score=0.95)
# 从 DTO 转换为领域实体
doc = dto.to_entity()
# 转换为字典(用于 JSON 序列化)
data = dto.to_dict()
# 从字典创建 DTO(用于 JSON 反序列化)
dto = DocumentDTO.from_dict(data)
from src.application.vector_search.dtos import SearchResultDTO, SearchResponseDTO
# 从领域实体创建搜索结果 DTO
result_dto = SearchResultDTO.from_entity(search_result)
# 创建搜索响应
response = SearchResponseDTO(
results=[result_dto1, result_dto2],
total=2,
query_text="test query",
took_ms=150
)
# 序列化为 JSON
json_data = response.to_dict()
Task 5.3 已完成。下一个任务是:
Task 5.4: 实现向量搜索命令处理器
application/vector_search/handlers.pyTask 5.3 成功实现了向量搜索应用层的 DTOs,提供了:
实现遵循了分层架构原则,DTOs 作为应用层和表现层之间的桥梁,与领域实体解耦,提供了灵活的数据传输和序列化支持。