数据工程数据编排ETL任务调度批处理流处理数据集成后端【免费下载链接】mage-ai Build, run, and manage data pipelines for integrating and transforming data.项目地址https://gitcode.com/gh_mirrors/ma/mage-ai点击查看免费下载Commercetools 是电商领域常用的 Headless Commerce 平台本指南聚焦 Mage 开源仓库中mage_integrations子项目提供的 Commercetools 数据源Source集成完整讲解其五项必填配置、基于 OAuth 2.0 Client Credentials 的认证流程、六大可同步数据流Stream及其 JSON Schema 结构并结合仓库源码说明底层实现原理。读完本文你将能够在 Mage 中独立完成 Commercetools 数据源的配置、连接测试与数据管道搭建。数据源定位与配置清单在 Mage 项目中Commercetools 数据源位于 mage_integrations/mage_integrations/sources/commercetools其官方说明文档即 README.md仓库内另有面向使用者的镜像文档 docs/data-integrations/sources/commercetools.mdx。配置该数据源时必须提供以下五个凭据参数Key说明示例值client_idAPI 客户端的client_id用于获取访问令牌access tokenabcdefgclient_secretAPI 客户端的client_secret用于获取访问令牌abcdefghijklmnopqrstuvwxyzhostCommercetools URL 中的主机段例如aws、gcpgcpproject_key与你的项目关联的项目键project keyprojectkeyregionCommercetools URL 中的区域段例如us-central1、us-east-2us-central1仓库为配置提供了可直接引用的 JSON 模板 templates/config.json结构如下{ region: region, host: host, project_key: project_key, client_id: client_id, client_secret: client_secret }从模板可以看出五个键均为字符串类型且模板中的占位值与 README 中的示例一一对应实际使用时只需将占位符替换为你在 Commercetools 控制台获得的真实值。如何获取client_id与client_secret按照 Commercetools 官方的授权文档docs.commercetools.com 的 Authorization 章节执行以下两步即可创建 API 客户端Create an API client登录 Commercetools Merchant Center在项目设置中进入开发者工具创建一个新的 API Client创建时可按需选择 scopes权限范围。从 API 客户端凭据页面获取三项信息创建完成后凭据页面会展示client_id、client_secret以及project_key将这三项分别填入上文配置的对应字段。其中client_secret属于高敏感凭据创建后通常仅展示一次建议直接存入项目的环境变量或密钥管理Secrets机制中避免明文写入代码仓库。底层认证与请求实现OAuth 2.0 Client Credentials配置项之所以是这五键组合可以在 client.py 的源码中得到完整印证。该文件定义Client类继承自mage_integrations.sources.http.client.Client其职责是完成令牌获取、URL 拼接与 HTTP 请求。令牌获取get_access_token实现了标准的 OAuth 2.0 Client Credentials 流程通过requests.post向认证端点发送请求端点 URL 由 region 与 host 拼接而成https://auth.{region}.{host}.commercetools.com/oauth/token?grant_typeclient_credentialsscopemanage_project:{project_key}请求时以client_id与client_secret作为 HTTP Basic Auth 凭据auth(client_id, client_secret)响应 JSON 中携带access_token随后被封装为Bearer {access_token}形式的令牌保存在实例中供后续所有 API 请求的Authorization请求头使用def get_headers(self): return { Authorization: self.token, Content-Type: application/json }业务 API 端点base_url同样由配置动态拼接https://api.{region}.{host}.commercetools.com/{project_key}由此可以推断host与region两个配置项的本质作用它们共同决定认证域名与 API 域名的地理分布不同云厂商aws/gcp与不同区域如us-central1、us-east-2会指向不同的 Commercetools 基础设施这也是这两个字段必须与你在 Commercetools 项目中实际分配到的部署位置严格一致的原因。project_key则作为路径的一部分用于隔离不同项目的数据。request()方法为上层流Stream提供统一的请求入口支持get等方法可传入path自动拼接base_url或完整url最终调用父类的make_request完成实际网络请求。数据同步的入口Source 与六大 StreamCommercetools 数据源的入口定义在init.py其中Commercetools类继承自mage_integrations.sources.base.Source构造时即创建Client实例这意味着初始化数据源时就会触发令牌获取load_data(stream, bookmarks, query)按tap_stream_id从STREAMS映射表中查找对应的 Stream 类并委托其load_data(bookmarks)拉取数据get_table_key_properties与get_valid_replication_keys均返回该 Stream 的KEY_PROPERTIES即主键字段文件末尾的if __name__ __main__: main(Commercetools)提供了标准的命令行入口可直接以脚本方式运行该数据源。Stream 的注册表定义在 schemas.py 中共六个数据流Stream ID类名API 路径主键KEY_PROPERTIEScustomersCustomersStream/customersiddiscount_codesDiscountCodesStream/discount-codesidinventoryInventoryStream/inventoryidordersOrdersStream/ordersidpaymentsPaymentsStream/paymentsidproductsProductsStream/productsid以 orders.py 和 products.py 为例每个 Stream 类只需声明KEY_PROPERTIES与URL_PATH两个类属性即可完成定义其余行为继承自 streams/base.py 中的BaseStream。分页拉取机制BaseStream.load_data是数据抽取的核心采用基于 offset 的游标分页实现要点如下从offset 0开始每次请求携带params {offset: offset}调用self.client.request(pathself.URL_PATH, params...)获取响应从响应 JSON 的results字段取出本页数据并yield以生成器方式流式产出避免一次性加载全量数据占用内存将offset累加本页返回的count即本次实际返回的记录数继续循环直至results为空拉取完成后通过self.logger.info输出完成日志。这一实现与 Commercetools API 的查询响应结构resultscount严格对应count字段同时承担“本页大小”与“下一页游标步长”的双重职责。bookmarks参数为增量同步incremental sync预留了书签接口但当前实现中未对 bookmarks 做额外过滤逻辑实际同步以全量分页为主。数据 Schema 与共享结构数据源的输出结构由 schemas 目录下的 JSON Schemadraft-07定义包含六个顶层 schema 文件customers.json、discount_codes.json、inventory.json、orders.json、payments.json、products.json以及shared/子目录中的 19 个共享子结构。顶层 schema 直接对应各 API 资源的字段。以 customers.json 为例包含id、customerNumber、key、version、createdAt、lastModifiedAt、email、firstName、lastName、middleName、title、salutation等字段其中createdAt、lastModifiedAt声明为format: date-time的时间戳字段可用于后续的时间分区或增量判断createdBy、lastModifiedBy等审计字段则引用共享结构。shared/目录中的共享结构覆盖了电商场景下的典型嵌套对象例如money.json金额结构包含type、currencyCode、centAmount以分为单位的金额与fractionDigits货币小数位address.json地址结构price.json/price_scoped.json/discounted_price_per_quantity.json商品价格相关结构product_data.json/product_variant.json/dimension.json商品数据与变体结构taxrate.json/taxed_price.json/taxed_item_price.json税务相关结构custom.json/field_type.json自定义字段与字段类型localized_string.json/state.json/reference.json/key_reference.json本地化字符串、状态机与引用类型shipping_details.json/created_by.json/last_modified_by.json发货详情与审计信息。这些共享结构通过$ref被各顶层 schema 引用保证了跨流之间同类字段的类型定义一致。将 Commercetools 接入 Mage 数据管道在 Mage 中实际使用该数据源时通常遵循以下流程准备凭据按上文“获取 client_id 与 client_secret”一节创建 API Client记录client_id、client_secret与project_key同时确认项目部署的host如aws、gcp与region如us-central1、us-east-2。在数据集成管道中配置 Source新建 Data Integration 类型的管道将 Source 类型选为 Commercetools依次填写五键配置。执行 Discovery触发 schema 发现Mage 会读取上文描述的 JSON Schema 生成可选的输出目录catalog你可以选择要同步的一个或多个 Stream如仅同步customers与orders。连接测试Mage 会使用配置尝试初始化Client并验证令牌获取链路是否可用。需要说明的是Commercetools类并未覆写Source基类中的test_connection方法而基类的默认实现会抛出“不支持连接测试”的异常因此该数据源目前以真实拉取数据时的成败作为连通性验证手段。运行管道管道运行时load_data会按 Stream 依次请求/customers、/orders等端点以 offset 分页逐批拉取数据并写入目标端。配置过程中如果遇到认证失败优先检查client_id/client_secret是否与 API Client 凭据页面一致、scope是否包含manage_project:{project_key}所需的权限以及host/region是否与项目实际部署位置匹配——后两项拼写错误会直接导致认证端点或 API 端点 DNS 解析失败这一点可以从 client.py 中 URL 的拼接逻辑直观定位。小结本文以仓库中的 Commercetools README 为骨架结合 client.py、schemas.py、streams/base.py 与 schemas 目录中的 JSON Schema完整还原了 Commercetools 数据源从配置、认证到分页同步的整条实现链路。对于需要将 Commercetools 的客户、订单、商品、库存、支付与折扣码数据汇入数据仓库的团队按上述步骤配置即可快速跑通数据集成管道。赞分享数据工程数据编排ETL任务调度批处理流处理数据集成后端【免费下载链接】mage-ai Build, run, and manage data pipelines for integrating and transforming data.项目地址https://gitcode.com/gh_mirrors/ma/mage-ai点击查看免费下载相关推荐从混用尺寸到统一 A4PDF补丁丁页面大小调整的 3 个实操任务从混用尺寸到统一 A4PDF补丁丁页面大小调整的 3 个实操任务 扫描文档一半 A4 一半 A5合并后的文件页面尺寸各不相等打印排队时最头疼。PDF补丁丁数据工程数据编排ETL任务调度批处理流处理数据集成后端前端Mage 数据集成中接入 Outreach 数据源OAuth 认证配置、参数详解与增量同步原理Mage 数据集成中接入 Outreach 数据源OAuth 认证配置、参数详解与增量同步原理 Outreach 是销售参与Sales Engagement数据工程数据编排ETL任务调度批处理流处理数据集成后端前端Mage AI 数据集成深入解析 Mode 分析平台数据源Mode Source的配置、认证与同步原理Mage AI 数据集成深入解析 Mode 分析平台数据源Mode Source的配置、认证与同步原理 导读 本文围绕 Mage AI 开源项目中内置的数据工程数据编排ETL任务调度批处理流处理数据集成后端前端上一篇Teleport Machine ID 主机证书支持RFD 83实践指南用 tbot 自动化 OpenSSH 主机证书签发下一篇超全PrimeVue零基础入门从安装到实战创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考