MongoDB 数据加载器集成测试指南基于 Docker 与 workspace/datalake 架构的 Data Formulator 验证方案【免费下载链接】data-formulator Data Formulator is an interactive AI-powered data analysis system makes it easy to connect, explore and visualize data.项目地址: https://gitcode.com/GitHub_Trending/da/data-formulator本篇指南围绕 Data Formulator 项目中 MongoDB 数据加载器MongoDBDataLoader的集成测试展开通过 Docker 快速拉起一个真实 MongoDB 7 实例并以项目自研的 workspace/datalake 落盘架构为验证底座覆盖连接、集合枚举、文档扁平化、Parquet 落盘、连接生命周期管理等一系列可重复执行的测试场景。读完本文你将掌握从零启动测试数据库、运行完整测试套件、理解 MongoDB 文档到 DataFrame/Arrow 的转换原理以及如何用环境变量控制测试目标的完整能力。一、测试目标与架构背景1.1 被测对象MongoDBDataLoader被测核心是 py-src/data_formulator/data_loader/mongodb_data_loader.py 中定义的MongoDBDataLoader类它继承自 external_data_loader.py 中的ExternalDataLoader抽象基类。该类在项目中的定位是Connect to a MongoDB database and load documents from collections.即负责建立到 MongoDB 的连接、枚举集合collection、把文档取回并转换为 PyArrow Table最终以 Parquet 形式写入工作区workspace。整条链路遵循基类定义的约定外部数据源MongoDB Collection → PyArrow Table → Parquetworkspace/datalakefetch_data_as_arrow()是每个加载器必须实现的主取数方法ingest_to_workspace()则由基类统一实现先通过fetch_data_as_arrow()取回 Arrow 表再调用Workspace.write_parquet_from_arrow()落盘全程避免 pandas 中转带来的额外开销见 external_data_loader.py 中ingest_to_workspace的实现与文档注释。1.2 测试设计真实数据库 workspace/datalake本测试套件tests/database-dockers/mongodb/README.md并非单元级 mock而是要求真实的 MongoDB 实例配合workspace/datalake 设计进行端到端验证用 Docker 运行 MongoDB 7 容器内置种子数据测试过程中创建临时 workspacetempfile.mkdtemp前缀df_test_mongo_验证文档数据被正确写入 Parquet 文件并可读回。测试还默认跳过式设计mongo_available()会先尝试client.admin.command(ping)探测连接连接不可用时跳过运行时用例静态用例list_params、auth_instructions则始终可跑。二、环境准备Prerequisites依赖说明Docker运行 MongoDB 测试容器镜像mongo:7-jammyPython 依赖pymongo、pyarrow、pandas、bson等项目环境见 requirements.txt项目环境使用仓库根目录的uv run/ pytest 运行测试种子数据脚本 init_data.js 会在容器首次启动时自动执行创建products12 条每条含嵌套对象specs与数组tags专门用于验证文档扁平化逻辑customers10 条含signup_date日期字段orders10 条含order_date与金额字段app_settings4 条轻量配置型集合。其中products的 12 条数据与 MySQL/Postgres 种子数据保持一致便于跨数据库加载器对拍见 init_data.js 顶部注释。三、快速开始Quick Start以下命令均从仓库根目录执行。3.1 方式 A统一脚本启动# 一次性启动全部测试数据库 ./tests/database-dockers/run_test_dbs.sh start # 仅启动 MongoDB ./tests/database-dockers/run_test_dbs.sh start mongodb3.2 方式 B仅 MongoDB# 方式 B-1通过统一脚本指定服务 ./tests/database-dockers/run_test_dbs.sh start mongodb # 方式 B-2直接使用 mongodb 子目录的 compose 文件 cd tests/database-dockers/mongodb docker compose up -d说明原文档中引用的run_test_dbs.sh在仓库当前快照中未找到同名脚本仓库当前实际提供的跨服务辅助入口是 PowerShell 脚本 tests/database-dockers/test-dbs.ps1Windows 使用macOS/Linux 下各服务子目录自带独立的docker-compose.yml与start.sh详见 tests/database-dockers/README.md。因此 macOS/Linux 下最稳妥的启动方式是cd tests/database-dockers/mongodb docker compose up -d --build --wait3.3 运行测试与清理# 运行 MongoDB 测试套件 pytest tests/database-dockers/mongodb/ -v # 停止全部测试数据库 ./tests/database-dockers/run_test_dbs.sh stop # 或仅停 MongoDB子目录脚本 cd tests/database-dockers/mongodb docker compose down3.4 一条命令数据库 后端联调mongodb/start.sh 提供了测试库 Data Formulator 后端的一键联调脚本./tests/database-dockers/mongodb/start.sh # 启动 MongoDB DF 后端端口 5567 ./tests/database-dockers/mongodb/start.sh stop # 关闭 MongoDB 容器脚本内部逻辑若容器df-test-mongodb已在运行则复用否则执行docker compose up -d --build --wait等待健康检查通过随后导出测试环境变量并启动后端export MONGO_HOSTlocalhost MONGO_PORT${MONGO_PORT:-27018} export MONGO_USERNAMEtestuser MONGO_PASSWORDtestpass MONGO_DATABASEtestdb uv run data_formulator --port 5567 --dev前端需另行执行npx vite默认 http://localhost:5173。四、环境变量Env vars测试通过环境变量定位 MongoDB未设置时使用默认值与 docker-compose.yml 的容器端口映射宿主机27018→ 容器27017保持一致变量默认值说明MONGO_HOSTlocalhostMongoDB 主机地址MONGO_PORT27018宿主机映射端口非默认 27017MONGO_USERNAMEtestuser测试用户MONGO_PASSWORDtestpass测试密码MONGO_DATABASEtestdb目标数据库若 27018 端口被占用可在启动前覆盖MONGO_PORTexport MONGO_PORT37018 ./tests/database-dockers/mongodb/start.sh五、测试基础设施源码剖析5.1 Dockerfile 与 Compose 配置Dockerfile 基于mongo:7-jammy通过环境变量预设 root 账号admin/admin与初始数据库testdb并将 init_data.js 挂入/docker-entrypoint-initdb.d/实现首次启动自动建库、建用户、插种子数据FROM mongo:7-jammy ENV MONGO_INITDB_ROOT_USERNAMEadmin ENV MONGO_INITDB_ROOT_PASSWORDadmin ENV MONGO_INITDB_DATABASEtestdb COPY init_data.js /docker-entrypoint-initdb.d/ EXPOSE 27017docker-compose.yml 的关键点容器名固定为df-test-mongodb端口映射${MONGO_PORT:-27018}:27017可通过环境变量覆盖宿主机端口内置healthcheck用mongosh --quiet --eval db.adminCommand(ping)每 5 秒探测、3 秒超时、最多重试 10 次供docker compose up --wait同步等待就绪。5.2 种子数据设计init_data.jsinit_data.js完成两件事创建测试用户testuser / testpass角色为testdb上的readWrite与 README 的环境变量默认值一一对应插入四组集合其中products特意设计为嵌套对象specs 数组tags混合结构例如{ id: 1, name: Laptop Pro 15, category: Electronics, price: 1299.99, stock_quantity: 50, created_at: new Date(2024-01-01T10:00:00Z), specs: { weight_kg: 1.8, screen: 15.6 inch, ram_gb: 16 }, tags: [laptop, portable, pro] }这类文档正是 test_mongodb_loader.py 中嵌套文档扁平化与数组扁平化两条用例的数据来源。六、连接参数、认证与静态契约6.1 list_params完整的连接参数声明MongoDBDataLoader.list_params()静态方法无需连接即可测试声明了全部可配置参数及其类型、必填性、默认值与 UI 分组tier是前端连接表单与后端校验的单一事实来源参数类型必填默认值tier说明hoststring✅localhostconnection服务器地址portint–27017connectionadvanced服务器端口usernamestring–auth留空表示无认证passwordstring–authsensitive留空表示无认证databasestring✅connection数据库名collectionstring–filter留空则枚举全部集合authSourcestring–auth认证库默认指向目标库测试断言host为必填项requiredTrue这与基类validate_params()的校验逻辑吻合——缺必填参数会抛出ConnectorParamError并列出缺失项见 external_data_loader.py。password标记为sensitive在get_safe_params()落盘元数据时会被剔除避免凭据泄漏。6.2 auth_instructions面向用户的认证指引auth_instructions()返回一段 Markdown 格式的指导文本长度 100 字符涵盖本地与远程两种场景**Example:** host: localhost · port: 27017 · database: mydb · collection: users **Local setup:** 无认证时留空用户名密码 **Remote setup:** 从 DBA 处获取 host/port/username/password **Troubleshooting:** 用 mongosh --host host --port port 先行验证6.3 连接建立与认证语义构造函数__init__中用户名与密码同时非空时才启用认证并附带authSource默认取目标database名否则以匿名方式连接if self.username and self.password: self.mongo_client pymongo.MongoClient( hostself.host, portself.port, usernameself.username, passwordself.password, authSourceauth_source, ) else: self.mongo_client pymongo.MongoClient(hostself.host, portself.port)连接失败会被包装为RuntimeError(Failed to connect to MongoDB: ...)抛出同时记录 error 日志。此外还提供了test_connection()方法通过admin.command(ping)快速探测连通性。七、核心取数链路文档 → DataFrame → Arrow7.1 特殊类型转换_convert_special_typesMongoDB 文档含ObjectId、datetime、bytes等非 JSON 类型直接转 pandas 会出问题。_convert_special_types做递归归一化ObjectId→str(value)datetime→value.isoformat()如2024-01-15T10:30:00bytes→value.decode(utf-8, errorsignore)嵌套 dict / list 递归处理list 内元素同样归一化。测试 test_mongodb_loader.py 中test_convert_special_types用ObjectId()、datetime(2024, 1, 15, 10, 30, 0)、bbinary data三种类型断言转换结果为 str 且日期含2024-01-15。7.2 文档扁平化_flatten_document_flatten_document(doc, parent_key, sep_)用递归把嵌套文档与数组拍平成键_键式扁平列规则清晰嵌套 dictnested.level1→ 键nested_level1嵌套更深nested.deeper.level2→ 键nested_deeper_level2数组按下标从 1 开始编号array[1,2,3]→array_1/array_2/array_3空数组映射为键 None数组元素若为 dict继续递归item_key带上索引。测试test_flatten_document给出的对应关系原始文档扁平后键值name: TestnameTestnested: {level1: value1}nested_level1value1nested.deeper: {level2: value2}nested_deeper_level2value2array: [1,2,3]array_1 / array_2 / array_31 / 2 / 3_process_documents将两个静态方法串联先_convert_special_types再_flatten_document最后pd.DataFrame(processed_docs)。7.3 取数主方法fetch_data_as_arrow由于 MongoDB 没有原生 Arrow 支持实现路径为查询 cursor → 取回文档列表 →_process_documents→pa.Table.from_pandas(df, preserve_indexFalse)。关键细节size通过min(opts.get(size, MAX_IMPORT_ROWS), MAX_IMPORT_ROWS)双重限制MAX_IMPORT_ROWS 2_000_000定义于 external_data_loader.pysort_columns/sort_order支持排序desc方向映射为 -1直接作用于 cursor 的.sort(spec)结果经.limit(size)后转列表source_table支持database.collection全名取最后一个.后的片段作为集合名空集合返回pa.table({})并打 warning。测试覆盖test_fetch_data_as_arrow断言取回表非空且含name/category/price列test_fetch_data_respects_size断言size5时行数 ≤ 5。7.4 probe原生聚合管道下推MongoDB loader 重写了基类的probe(path, query)把前端 SPJQ 查询对象filters/columns/group_by/aggregates/order_by/limit编译为MongoDB 聚合管道$match / $group / $sort / $limit在服务端全量执行因此结果标记为exactTrue。_compile_probe_pipeline中的映射细节包括过滤器EQ/NEQ/GT/GTE/LT/LTE→$eq/$ne/$gt/$gte/$lt/$lteLIKE/ILIKE→$regex值经re.escape防注入$options: i忽略大小写IN/NOT_IN→$in/$ninIS_NULL/IS_NOT_NULL→$eq: None / $ne: NoneBETWEEN→$gte $lte聚合count无列时$sum:1有列时条件计数、count_distinct$addToSet后按$size取长度、sum/avg/min/max→ 对应$group操作符。八、集合枚举与目录树Catalog8.1 list_tables扁平枚举list_tables(table_filterNone)的行为配置了collection时只枚举该集合否则db.list_collection_names()全量枚举table_filter做大小写不敏感的子串过滤每个集合返回name、path[collection_name]与metadatarow_count、columns、sample_rowsrow_count用count_documents({})精确统计列信息来自最多 10 条样本文档经扁平化后的 DataFrame dtype样本行经df_to_safe_records转为可安全序列化的记录见 datalake/parquet_utils.py单个集合枚举失败仅打 debug 日志并跳过不影响其余集合。对应测试test_list_tables断言products/customers/orders/app_settings全部出现且元数据含columns与row_counttest_list_tables_with_filter用prod过滤test_list_tables_specific_collection在配置collectionproducts时断言只返回 1 张表test_list_tables_row_count精确断言 products 的row_count 12。8.2 目录树浏览catalog_hierarchy 与 lscatalog_hierarchy()声明两级结构Database → Collection。由于database是必填参数连接时即固定被钉住/pinned实际可浏览层级只剩 Collection 一级effective_hierarchy()会把已提供值的层级从浏览树中隐藏ls(path)在 collection 层返回CatalogNode(name集合名, node_typetable, path[集合名])支持filter子串过滤。get_metadata(path)则对单个集合做count_documents 5 条样本的列推断。九、落盘与工作区Workspace验证9.1 ingest_to_workspace 全流程ingest_to_workspace(workspace, table_name, source_table, import_options)由基类统一实现见 external_data_loader.py测试验证了完整链路workspace Workspace(test-identity-mongo, root_dir临时目录) meta loader.ingest_to_workspace( workspace, products_test, source_tableproducts, import_options{size: 100}, )测试断言meta.name products_test、file_type parquet、row_count 0且workspace.file_exists(meta.filename)为真随后workspace.read_data_as_df(products_test)读回 DataFrame行数为 12。9.2 表名清洗test_ingest_sanitizes_table_name用test-table-with-dashes这种含连字符的表名验证清洗逻辑落盘后表名出现在workspace.list_tables()中且可正常读回。底层依赖 datalake/table_names.py 的sanitize_external_loader_table_nameExternalDataLoader中的sanitize_table_name是其向后兼容别名。9.3 工作区元数据与 Parquet 结构test_get_table_info_from_datalake验证导入后workspace.list_tables()含新表workspace.get_table_metadata(my_table)返回元数据对象名称一致workspace.get_parquet_schema(my_table)返回含columns与num_rows的 schema 信息read_data_as_df可读回数据。9.4 嵌套与数组文档的落盘验证test_ingest_nested_documents_flattened导入 products 后DataFrame 中存在以specs_前缀开头的列如specs_weight_kgtest_ingest_with_arrays_flattened存在以tags_前缀开头的列如tags_1/tags_2/tags_3。这两条用例直接印证了第 7 节扁平化逻辑在真实取数 → 落盘 → 读回全链路中的正确性。十、连接生命周期管理MongoDB loader 实现了多重资源释放保障见 mongodb_data_loader.pyclose()关闭mongo_client并置None幂等安全__enter__ / __exit__支持with MongoDBDataLoader(config) as loader:语法退出即关闭__del__析构函数兜底关闭防止连接泄漏。对应测试test_connection_closeclose()后loader.mongo_client is Nonetest_context_managerwith块结束后断言客户端已关闭。测试基座本身同样注意清理每个 loader 经addCleanup(self._close_loader)注册关闭回调workspace 临时目录经shutil.rmtree(ignore_errorsTrue)清理。十一、测试矩阵速查将 test_mongodb_loader.py 的全部用例归纳如下作为执行pytest tests/database-dockers/mongodb/ -v时的预期清单类别用例验证点集合枚举test_list_tables/_with_filter/_specific_collection/_row_count全量枚举、子串过滤、单集合、精确行数12取数test_fetch_data_as_arrow/_respects_sizeArrow 表非空、列存在、size 限制生效落盘test_ingest_table_to_workspaceParquet 写入、读回、行数一致扁平化test_ingest_nested_documents_flattened/_with_arrays_flattenedspecs_*、tags_*前缀列生成表名清洗test_ingest_sanitizes_table_name连字符表名可安全落盘元数据test_get_table_info_from_datalakeget_table_metadata、get_parquet_schema、读回连接生命周期test_connection_close/test_context_manager显式关闭与 with 语法关闭静态转换test_flatten_document/test_convert_special_types扁平化键规则、特殊类型归一化静态契约test_list_params/test_auth_instructions参数声明完整性、认证指引内容十二、常见问题与排查连接被拒 / 测试跳过确认容器已启动且健康docker compose ps确认MONGO_PORT与 compose 映射一致默认 27018可用mongosh --host localhost --port 27018 -u testuser -p testpass --authenticationDatabase testdb手动验证。端口冲突覆盖MONGO_PORT环境变量后重启容器。容器名冲突若df-test-mongodb已被占用先docker compose down或删除旧容器参考 tests/database-dockers/README.md 的提示。使用统一入口Windows/PowerShell 用户可用 tests/database-dockers/test-dbs.ps1 执行.\tests\database-dockers\test-dbs.ps1 test mongodb脚本会自动启库并运行对应 pytest 目录。验证无认证环境本地开发库未开启认证时将MONGO_USERNAME/MONGO_PASSWORD留空即可loader 会自动走匿名连接分支。十三、总结围绕 tests/database-dockers/mongodb/README.md 描述的测试方案本文还原了其完整的工程落地Docker 化的 MongoDB 7 测试实例docker-compose.yml、Dockerfile、init_data.js、覆盖连接/枚举/取数/扁平化/落盘/资源管理的测试矩阵test_mongodb_loader.py以及底层加载器 mongodb_data_loader.py 的参数契约、特殊类型转换、递归扁平化、Arrow 取数与聚合管道下推等实现细节。这套真实数据库 workspace/datalake的测试范式为 Data Formulator 连接任意外部数据源提供了可复制的集成验证模板。【免费下载链接】data-formulator Data Formulator is an interactive AI-powered data analysis system makes it easy to connect, explore and visualize data.项目地址: https://gitcode.com/GitHub_Trending/da/data-formulator创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
MongoDB 数据加载器集成测试指南：基于 Docker 与 workspace/datalake 架构的 Data Formulator 验证方案
MongoDB 数据加载器集成测试指南基于 Docker 与 workspace/datalake 架构的 Data Formulator 验证方案【免费下载链接】data-formulator Data Formulator is an interactive AI-powered data analysis system makes it easy to connect, explore and visualize data.项目地址: https://gitcode.com/GitHub_Trending/da/data-formulator本篇指南围绕 Data Formulator 项目中 MongoDB 数据加载器MongoDBDataLoader的集成测试展开通过 Docker 快速拉起一个真实 MongoDB 7 实例并以项目自研的 workspace/datalake 落盘架构为验证底座覆盖连接、集合枚举、文档扁平化、Parquet 落盘、连接生命周期管理等一系列可重复执行的测试场景。读完本文你将掌握从零启动测试数据库、运行完整测试套件、理解 MongoDB 文档到 DataFrame/Arrow 的转换原理以及如何用环境变量控制测试目标的完整能力。一、测试目标与架构背景1.1 被测对象MongoDBDataLoader被测核心是 py-src/data_formulator/data_loader/mongodb_data_loader.py 中定义的MongoDBDataLoader类它继承自 external_data_loader.py 中的ExternalDataLoader抽象基类。该类在项目中的定位是Connect to a MongoDB database and load documents from collections.即负责建立到 MongoDB 的连接、枚举集合collection、把文档取回并转换为 PyArrow Table最终以 Parquet 形式写入工作区workspace。整条链路遵循基类定义的约定外部数据源MongoDB Collection → PyArrow Table → Parquetworkspace/datalakefetch_data_as_arrow()是每个加载器必须实现的主取数方法ingest_to_workspace()则由基类统一实现先通过fetch_data_as_arrow()取回 Arrow 表再调用Workspace.write_parquet_from_arrow()落盘全程避免 pandas 中转带来的额外开销见 external_data_loader.py 中ingest_to_workspace的实现与文档注释。1.2 测试设计真实数据库 workspace/datalake本测试套件tests/database-dockers/mongodb/README.md并非单元级 mock而是要求真实的 MongoDB 实例配合workspace/datalake 设计进行端到端验证用 Docker 运行 MongoDB 7 容器内置种子数据测试过程中创建临时 workspacetempfile.mkdtemp前缀df_test_mongo_验证文档数据被正确写入 Parquet 文件并可读回。测试还默认跳过式设计mongo_available()会先尝试client.admin.command(ping)探测连接连接不可用时跳过运行时用例静态用例list_params、auth_instructions则始终可跑。二、环境准备Prerequisites依赖说明Docker运行 MongoDB 测试容器镜像mongo:7-jammyPython 依赖pymongo、pyarrow、pandas、bson等项目环境见 requirements.txt项目环境使用仓库根目录的uv run/ pytest 运行测试种子数据脚本 init_data.js 会在容器首次启动时自动执行创建products12 条每条含嵌套对象specs与数组tags专门用于验证文档扁平化逻辑customers10 条含signup_date日期字段orders10 条含order_date与金额字段app_settings4 条轻量配置型集合。其中products的 12 条数据与 MySQL/Postgres 种子数据保持一致便于跨数据库加载器对拍见 init_data.js 顶部注释。三、快速开始Quick Start以下命令均从仓库根目录执行。3.1 方式 A统一脚本启动# 一次性启动全部测试数据库 ./tests/database-dockers/run_test_dbs.sh start # 仅启动 MongoDB ./tests/database-dockers/run_test_dbs.sh start mongodb3.2 方式 B仅 MongoDB# 方式 B-1通过统一脚本指定服务 ./tests/database-dockers/run_test_dbs.sh start mongodb # 方式 B-2直接使用 mongodb 子目录的 compose 文件 cd tests/database-dockers/mongodb docker compose up -d说明原文档中引用的run_test_dbs.sh在仓库当前快照中未找到同名脚本仓库当前实际提供的跨服务辅助入口是 PowerShell 脚本 tests/database-dockers/test-dbs.ps1Windows 使用macOS/Linux 下各服务子目录自带独立的docker-compose.yml与start.sh详见 tests/database-dockers/README.md。因此 macOS/Linux 下最稳妥的启动方式是cd tests/database-dockers/mongodb docker compose up -d --build --wait3.3 运行测试与清理# 运行 MongoDB 测试套件 pytest tests/database-dockers/mongodb/ -v # 停止全部测试数据库 ./tests/database-dockers/run_test_dbs.sh stop # 或仅停 MongoDB子目录脚本 cd tests/database-dockers/mongodb docker compose down3.4 一条命令数据库 后端联调mongodb/start.sh 提供了测试库 Data Formulator 后端的一键联调脚本./tests/database-dockers/mongodb/start.sh # 启动 MongoDB DF 后端端口 5567 ./tests/database-dockers/mongodb/start.sh stop # 关闭 MongoDB 容器脚本内部逻辑若容器df-test-mongodb已在运行则复用否则执行docker compose up -d --build --wait等待健康检查通过随后导出测试环境变量并启动后端export MONGO_HOSTlocalhost MONGO_PORT${MONGO_PORT:-27018} export MONGO_USERNAMEtestuser MONGO_PASSWORDtestpass MONGO_DATABASEtestdb uv run data_formulator --port 5567 --dev前端需另行执行npx vite默认 http://localhost:5173。四、环境变量Env vars测试通过环境变量定位 MongoDB未设置时使用默认值与 docker-compose.yml 的容器端口映射宿主机27018→ 容器27017保持一致变量默认值说明MONGO_HOSTlocalhostMongoDB 主机地址MONGO_PORT27018宿主机映射端口非默认 27017MONGO_USERNAMEtestuser测试用户MONGO_PASSWORDtestpass测试密码MONGO_DATABASEtestdb目标数据库若 27018 端口被占用可在启动前覆盖MONGO_PORTexport MONGO_PORT37018 ./tests/database-dockers/mongodb/start.sh五、测试基础设施源码剖析5.1 Dockerfile 与 Compose 配置Dockerfile 基于mongo:7-jammy通过环境变量预设 root 账号admin/admin与初始数据库testdb并将 init_data.js 挂入/docker-entrypoint-initdb.d/实现首次启动自动建库、建用户、插种子数据FROM mongo:7-jammy ENV MONGO_INITDB_ROOT_USERNAMEadmin ENV MONGO_INITDB_ROOT_PASSWORDadmin ENV MONGO_INITDB_DATABASEtestdb COPY init_data.js /docker-entrypoint-initdb.d/ EXPOSE 27017docker-compose.yml 的关键点容器名固定为df-test-mongodb端口映射${MONGO_PORT:-27018}:27017可通过环境变量覆盖宿主机端口内置healthcheck用mongosh --quiet --eval db.adminCommand(ping)每 5 秒探测、3 秒超时、最多重试 10 次供docker compose up --wait同步等待就绪。5.2 种子数据设计init_data.jsinit_data.js完成两件事创建测试用户testuser / testpass角色为testdb上的readWrite与 README 的环境变量默认值一一对应插入四组集合其中products特意设计为嵌套对象specs 数组tags混合结构例如{ id: 1, name: Laptop Pro 15, category: Electronics, price: 1299.99, stock_quantity: 50, created_at: new Date(2024-01-01T10:00:00Z), specs: { weight_kg: 1.8, screen: 15.6 inch, ram_gb: 16 }, tags: [laptop, portable, pro] }这类文档正是 test_mongodb_loader.py 中嵌套文档扁平化与数组扁平化两条用例的数据来源。六、连接参数、认证与静态契约6.1 list_params完整的连接参数声明MongoDBDataLoader.list_params()静态方法无需连接即可测试声明了全部可配置参数及其类型、必填性、默认值与 UI 分组tier是前端连接表单与后端校验的单一事实来源参数类型必填默认值tier说明hoststring✅localhostconnection服务器地址portint–27017connectionadvanced服务器端口usernamestring–auth留空表示无认证passwordstring–authsensitive留空表示无认证databasestring✅connection数据库名collectionstring–filter留空则枚举全部集合authSourcestring–auth认证库默认指向目标库测试断言host为必填项requiredTrue这与基类validate_params()的校验逻辑吻合——缺必填参数会抛出ConnectorParamError并列出缺失项见 external_data_loader.py。password标记为sensitive在get_safe_params()落盘元数据时会被剔除避免凭据泄漏。6.2 auth_instructions面向用户的认证指引auth_instructions()返回一段 Markdown 格式的指导文本长度 100 字符涵盖本地与远程两种场景**Example:** host: localhost · port: 27017 · database: mydb · collection: users **Local setup:** 无认证时留空用户名密码 **Remote setup:** 从 DBA 处获取 host/port/username/password **Troubleshooting:** 用 mongosh --host host --port port 先行验证6.3 连接建立与认证语义构造函数__init__中用户名与密码同时非空时才启用认证并附带authSource默认取目标database名否则以匿名方式连接if self.username and self.password: self.mongo_client pymongo.MongoClient( hostself.host, portself.port, usernameself.username, passwordself.password, authSourceauth_source, ) else: self.mongo_client pymongo.MongoClient(hostself.host, portself.port)连接失败会被包装为RuntimeError(Failed to connect to MongoDB: ...)抛出同时记录 error 日志。此外还提供了test_connection()方法通过admin.command(ping)快速探测连通性。七、核心取数链路文档 → DataFrame → Arrow7.1 特殊类型转换_convert_special_typesMongoDB 文档含ObjectId、datetime、bytes等非 JSON 类型直接转 pandas 会出问题。_convert_special_types做递归归一化ObjectId→str(value)datetime→value.isoformat()如2024-01-15T10:30:00bytes→value.decode(utf-8, errorsignore)嵌套 dict / list 递归处理list 内元素同样归一化。测试 test_mongodb_loader.py 中test_convert_special_types用ObjectId()、datetime(2024, 1, 15, 10, 30, 0)、bbinary data三种类型断言转换结果为 str 且日期含2024-01-15。7.2 文档扁平化_flatten_document_flatten_document(doc, parent_key, sep_)用递归把嵌套文档与数组拍平成键_键式扁平列规则清晰嵌套 dictnested.level1→ 键nested_level1嵌套更深nested.deeper.level2→ 键nested_deeper_level2数组按下标从 1 开始编号array[1,2,3]→array_1/array_2/array_3空数组映射为键 None数组元素若为 dict继续递归item_key带上索引。测试test_flatten_document给出的对应关系原始文档扁平后键值name: TestnameTestnested: {level1: value1}nested_level1value1nested.deeper: {level2: value2}nested_deeper_level2value2array: [1,2,3]array_1 / array_2 / array_31 / 2 / 3_process_documents将两个静态方法串联先_convert_special_types再_flatten_document最后pd.DataFrame(processed_docs)。7.3 取数主方法fetch_data_as_arrow由于 MongoDB 没有原生 Arrow 支持实现路径为查询 cursor → 取回文档列表 →_process_documents→pa.Table.from_pandas(df, preserve_indexFalse)。关键细节size通过min(opts.get(size, MAX_IMPORT_ROWS), MAX_IMPORT_ROWS)双重限制MAX_IMPORT_ROWS 2_000_000定义于 external_data_loader.pysort_columns/sort_order支持排序desc方向映射为 -1直接作用于 cursor 的.sort(spec)结果经.limit(size)后转列表source_table支持database.collection全名取最后一个.后的片段作为集合名空集合返回pa.table({})并打 warning。测试覆盖test_fetch_data_as_arrow断言取回表非空且含name/category/price列test_fetch_data_respects_size断言size5时行数 ≤ 5。7.4 probe原生聚合管道下推MongoDB loader 重写了基类的probe(path, query)把前端 SPJQ 查询对象filters/columns/group_by/aggregates/order_by/limit编译为MongoDB 聚合管道$match / $group / $sort / $limit在服务端全量执行因此结果标记为exactTrue。_compile_probe_pipeline中的映射细节包括过滤器EQ/NEQ/GT/GTE/LT/LTE→$eq/$ne/$gt/$gte/$lt/$lteLIKE/ILIKE→$regex值经re.escape防注入$options: i忽略大小写IN/NOT_IN→$in/$ninIS_NULL/IS_NOT_NULL→$eq: None / $ne: NoneBETWEEN→$gte $lte聚合count无列时$sum:1有列时条件计数、count_distinct$addToSet后按$size取长度、sum/avg/min/max→ 对应$group操作符。八、集合枚举与目录树Catalog8.1 list_tables扁平枚举list_tables(table_filterNone)的行为配置了collection时只枚举该集合否则db.list_collection_names()全量枚举table_filter做大小写不敏感的子串过滤每个集合返回name、path[collection_name]与metadatarow_count、columns、sample_rowsrow_count用count_documents({})精确统计列信息来自最多 10 条样本文档经扁平化后的 DataFrame dtype样本行经df_to_safe_records转为可安全序列化的记录见 datalake/parquet_utils.py单个集合枚举失败仅打 debug 日志并跳过不影响其余集合。对应测试test_list_tables断言products/customers/orders/app_settings全部出现且元数据含columns与row_counttest_list_tables_with_filter用prod过滤test_list_tables_specific_collection在配置collectionproducts时断言只返回 1 张表test_list_tables_row_count精确断言 products 的row_count 12。8.2 目录树浏览catalog_hierarchy 与 lscatalog_hierarchy()声明两级结构Database → Collection。由于database是必填参数连接时即固定被钉住/pinned实际可浏览层级只剩 Collection 一级effective_hierarchy()会把已提供值的层级从浏览树中隐藏ls(path)在 collection 层返回CatalogNode(name集合名, node_typetable, path[集合名])支持filter子串过滤。get_metadata(path)则对单个集合做count_documents 5 条样本的列推断。九、落盘与工作区Workspace验证9.1 ingest_to_workspace 全流程ingest_to_workspace(workspace, table_name, source_table, import_options)由基类统一实现见 external_data_loader.py测试验证了完整链路workspace Workspace(test-identity-mongo, root_dir临时目录) meta loader.ingest_to_workspace( workspace, products_test, source_tableproducts, import_options{size: 100}, )测试断言meta.name products_test、file_type parquet、row_count 0且workspace.file_exists(meta.filename)为真随后workspace.read_data_as_df(products_test)读回 DataFrame行数为 12。9.2 表名清洗test_ingest_sanitizes_table_name用test-table-with-dashes这种含连字符的表名验证清洗逻辑落盘后表名出现在workspace.list_tables()中且可正常读回。底层依赖 datalake/table_names.py 的sanitize_external_loader_table_nameExternalDataLoader中的sanitize_table_name是其向后兼容别名。9.3 工作区元数据与 Parquet 结构test_get_table_info_from_datalake验证导入后workspace.list_tables()含新表workspace.get_table_metadata(my_table)返回元数据对象名称一致workspace.get_parquet_schema(my_table)返回含columns与num_rows的 schema 信息read_data_as_df可读回数据。9.4 嵌套与数组文档的落盘验证test_ingest_nested_documents_flattened导入 products 后DataFrame 中存在以specs_前缀开头的列如specs_weight_kgtest_ingest_with_arrays_flattened存在以tags_前缀开头的列如tags_1/tags_2/tags_3。这两条用例直接印证了第 7 节扁平化逻辑在真实取数 → 落盘 → 读回全链路中的正确性。十、连接生命周期管理MongoDB loader 实现了多重资源释放保障见 mongodb_data_loader.pyclose()关闭mongo_client并置None幂等安全__enter__ / __exit__支持with MongoDBDataLoader(config) as loader:语法退出即关闭__del__析构函数兜底关闭防止连接泄漏。对应测试test_connection_closeclose()后loader.mongo_client is Nonetest_context_managerwith块结束后断言客户端已关闭。测试基座本身同样注意清理每个 loader 经addCleanup(self._close_loader)注册关闭回调workspace 临时目录经shutil.rmtree(ignore_errorsTrue)清理。十一、测试矩阵速查将 test_mongodb_loader.py 的全部用例归纳如下作为执行pytest tests/database-dockers/mongodb/ -v时的预期清单类别用例验证点集合枚举test_list_tables/_with_filter/_specific_collection/_row_count全量枚举、子串过滤、单集合、精确行数12取数test_fetch_data_as_arrow/_respects_sizeArrow 表非空、列存在、size 限制生效落盘test_ingest_table_to_workspaceParquet 写入、读回、行数一致扁平化test_ingest_nested_documents_flattened/_with_arrays_flattenedspecs_*、tags_*前缀列生成表名清洗test_ingest_sanitizes_table_name连字符表名可安全落盘元数据test_get_table_info_from_datalakeget_table_metadata、get_parquet_schema、读回连接生命周期test_connection_close/test_context_manager显式关闭与 with 语法关闭静态转换test_flatten_document/test_convert_special_types扁平化键规则、特殊类型归一化静态契约test_list_params/test_auth_instructions参数声明完整性、认证指引内容十二、常见问题与排查连接被拒 / 测试跳过确认容器已启动且健康docker compose ps确认MONGO_PORT与 compose 映射一致默认 27018可用mongosh --host localhost --port 27018 -u testuser -p testpass --authenticationDatabase testdb手动验证。端口冲突覆盖MONGO_PORT环境变量后重启容器。容器名冲突若df-test-mongodb已被占用先docker compose down或删除旧容器参考 tests/database-dockers/README.md 的提示。使用统一入口Windows/PowerShell 用户可用 tests/database-dockers/test-dbs.ps1 执行.\tests\database-dockers\test-dbs.ps1 test mongodb脚本会自动启库并运行对应 pytest 目录。验证无认证环境本地开发库未开启认证时将MONGO_USERNAME/MONGO_PASSWORD留空即可loader 会自动走匿名连接分支。十三、总结围绕 tests/database-dockers/mongodb/README.md 描述的测试方案本文还原了其完整的工程落地Docker 化的 MongoDB 7 测试实例docker-compose.yml、Dockerfile、init_data.js、覆盖连接/枚举/取数/扁平化/落盘/资源管理的测试矩阵test_mongodb_loader.py以及底层加载器 mongodb_data_loader.py 的参数契约、特殊类型转换、递归扁平化、Arrow 取数与聚合管道下推等实现细节。这套真实数据库 workspace/datalake的测试范式为 Data Formulator 连接任意外部数据源提供了可复制的集成验证模板。【免费下载链接】data-formulator Data Formulator is an interactive AI-powered data analysis system makes it easy to connect, explore and visualize data.项目地址: https://gitcode.com/GitHub_Trending/da/data-formulator创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考