批量操作Google Ads不卡顿：google-ads-python的add_mutation操作与Batch Job批量任务实战

批量操作Google Ads不卡顿：google-ads-python的add_mutation操作与Batch Job批量任务实战

【免费下载链接】google-ads-pythonGoogle Ads API Client Library for Python项目地址: https://gitcode.com/gh_mirrors/go/google-ads-python

google-ads-python是 Google Ads API 的官方 Python 客户端库，也是用 Python 批量管理广告计划、广告组、关键词的首选工具。本文面向新手，讲清楚两件事：如何用 Mutation（变更操作）一次性提交大批量增删改请求，以及在操作量超出单次调用上限时，如何用 Batch Job 批量任务让服务器排队慢慢跑，彻底告别"请求卡顿、限流报错"。

为什么批量操作 Google Ads 容易"卡顿"

很多新手的第一反应是：循环 1 万次，每次调用 API 建一个关键词。结果很快遇到两个问题：

问题	原因	后果
单次调用被拒	每次mutate_*调用最多携带10000 个操作	批量任务中途失败
RateExceededError 限流	单位时间内请求量超出账号配额	脚本卡顿、反复重试
超时断开	一次性同步请求耗时过长	本地网络超时

核心思路：不要"多次小请求"，而要"一次大请求"；实在太大，就交给 Batch Job 在服务器端排队执行。

最快写法：一次 mutate 调用携带上千个操作

在 google-ads-python 中，所有增删改操作统一叫Mutation，提交方式是各服务客户端的mutate_*方法（这就是"add_mutation"操作在代码中的真实形态）。关键点只有一个：它接收的是 Operation 列表，一次可以打包几千个操作。

以 examples/basic_operations/add_campaigns.py 为例，先建预算、再建广告计划：

campaign_budget_response = campaign_budget_service.mutate_campaign_budgets( customer_id=customer_id, operations=[campaign_budget_operation], # 列表：可一次放多个 )

想批量创建 500 个广告组？把 500 个AdGroupOperation全部放进operations列表，一次调用搞定。这就是批量操作不卡顿的第一层方案。

💡 每次调用携带 10000 个操作，批量 5 万条数据也只需 5 次调用，远比 5 万次调用安全。

操作量超限时：Batch Job 批量任务登场

当操作数超过 10000/次，甚至需要几十万条时，就该上BatchJobService了。它把整批操作交给 Google 服务器排队执行，你只管轮询进度、最后取结果。

完整参考示例：examples/campaign_management/add_complete_campaigns_using_batch_job.py，它演示了如何用 Batch Job 一次性创建"预算 + 计划 + 广告组 + 关键词 + 广告"的完整广告结构。

Batch Job 五步流程

第 1 步：创建批量任务—— 调用mutate_batch_job，拿到资源名（resource name），后续所有步骤都靠它定位任务。

第 2 步：上传操作—— 调用add_batch_job_operations。首次调用sequence_token传空，响应的next_sequence_token用于下一批上传，防止重复提交：

response = batch_job_service.add_batch_job_operations( resource_name=resource_name, sequence_token=None, # 首次传空 mutate_operations=operations, ) print(response.next_sequence_token) # 记住它，下次上传用

单个 Batch Job 最多可容纳100000 个操作，是常规 mutate 调用的 10 倍容量。

第 3 步：运行任务——run_batch_job返回一个长时操作（LRO）对象。

第 4 步：轮询等待—— 通过回调等待任务完成（示例中用asyncio.Event优雅地挂起主流程）。

第 5 步：获取结果—— 任务完成后分页拉取每条操作的执行状态：

for result in batch_job_service.list_batch_job_results(request=request): print(result.operation_index, result.status.message)

临时 ID：如何引用"还没创建成功"的实体

批量创建时，广告组要挂在广告计划下，可计划本身也在这批操作里还没建出来。解决办法是用负数临时 ID（如-1、-2）占位，任务执行时 Google 会自动替换成真实 ID。示例中get_next_temporary_id()函数就是干这个的：

_temporary_id: int = 0 def get_next_temporary_id() -> int: _temporary_id -= 1 return _temporary_id # 依次返回 -1、-2、-3...

⚠️ 注意：操作列表的顺序要符合依赖关系——预算在前、计划在中、广告组和关键词在后，临时 ID 引用才能正确解析。

批量操作的错误处理：部分失败与限流

批量操作最怕"全部回滚"，好在 Google Ads API 采用**部分失败（Partial Failure）**机制：某一条关键词写错，只影响那一条，其余照常创建。Batch Job 结果中每条操作都带独立状态，正好用来定位坏数据。

两个实战示例值得收藏：

• 部分失败：examples/error_handling/handle_partial_failure.py —— 故意创建含非法字符的广告组，演示如何从响应中提取partial_failure.errors并打印具体位置。
• 限流重试：examples/error_handling/handle_rate_exceeded_error.py —— 捕获RateExceededError后指数退避重试（示例配置了 3 次重试 + 递增等待时间），这是批量脚本的标配兜底逻辑。

相关错误类型定义可参考 google/ads/googleads/v24/errors/types/quota_error.py。

快速开始：安装与配置 google-ads-python

一键安装步骤（三种方式任选）：

pip install google-ads # 方式一：PyPI 安装 git clone https://gitcode.com/gh_mirrors/go/google-ads-python # 方式二：获取源码 pip install ./google-ads-python

配置方法：在用户主目录创建google-ads.yaml，填入开发者令牌、客户 ID 与 OAuth2 凭据；GoogleAdsClient.load_from_storage()会自动读取。配置模板见项目根目录的 google-ads.yaml，初始化细节在 examples/utils/example_helpers.py 中都有演示。

from google.ads.googleads.client import GoogleAdsClient client = GoogleAdsClient.load_from_storage(version="v24")

总结：该用哪种方案？

操作量级	推荐方案	关键 API
≤ 1 万条	单次 mutate 调用打包	mutate_*+ Operation 列表
1 万 ~ 10 万条	分批多次 mutate（每批 ≤1 万）	同上
> 10 万条	Batch Job 批量任务	BatchJobService五步流程

记住三个关键词：Operation 列表打包、sequence_token 防重、临时 ID 解依赖，再配上部分失败与限流重试兜底，你的批量脚本就能又快又稳。动手改改add_complete_campaigns_using_batch_job.py里的示例数字，跑通你的第一个 Batch Job 吧！

【免费下载链接】google-ads-pythonGoogle Ads API Client Library for Python项目地址: https://gitcode.com/gh_mirrors/go/google-ads-python