使用Python调用Google Drive API V3转移文件所有权时遭遇403(pendingOwnerWriterRequired)权限错误的问题咨询

开发者阿泽

2026-4-28

解决Google Drive跨账户文件转移的403 pendingOwnerWriterRequired错误

我最近在写Python脚本，想把一批文件从我的Google Drive转移到另一个Google账户，结果每次执行都返回403错误。我明明已经给目标账户开了Editor权限，但还是报这个错，实在搞不懂哪里出问题了。

错误日志

2026-04-17 17:22:45,421 INFO file_id=...<file-id>....- status=failed name=Graph.JPG message=<HttpError 403 when requesting https://www.googleapis.com/drive/v3/files/....<file-id>..../permissions/04862162013125427771?supportsAllDrives=true&fields=id%2CemailAddress%2Crole%2CpendingOwner&alt=json returned "The target user cannot be a pending owner because the target user does not have a writer role for the file.". Details: "[{'message': 'The target user cannot be a pending owner because the target user does not have a writer role for the file.', 'domain': 'global', 'reason': 'pendingOwnerWriterRequired'}]">

已尝试操作

手动给目标Google账户授予了目标文件夹的Editor（编辑）权限
确认OAuth/服务账号的权限范围包含https://www.googleapis.com/auth/drive

使用的代码

import argparse
import json
import logging
import os
import random
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence
from urllib.parse import parse_qs
from google.auth.transport.requests import Request
from google.oauth2 import service_account
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError

SCOPES = ["https://www.googleapis.com/auth/drive"]
FOLDER_MIME = "application/vnd.google-apps.folder"

RETRYABLE_REASONS = {
    "rateLimitExceeded",
    "userRateLimitExceeded",
    "sharingRateLimitExceeded",
    "backendError",
    "internalError",
}
NON_RETRYABLE_REASONS = {"pendingOwnerWriterRequired"}
RETRYABLE_STATUS = {403, 429, 500, 502, 503, 504}

def parse_args() -> argparse.Namespace:
    base_dir = Path(__file__).resolve().parent
    parser = argparse.ArgumentParser()
    parser.add_argument("--auth-mode", choices=["oauth", "service-account"], default="oauth")
    parser.add_argument("--credentials-file")
    parser.add_argument("--config-file", default=str(base_dir / "request_config.json"))
    parser.add_argument("--token-file", default=str(base_dir / "request_token.json"))
    parser.add_argument("--oauth-redirect-uri")
    parser.add_argument("--subject-email")
    parser.add_argument("--folder-id")
    parser.add_argument("--target-owner-email")
    parser.add_argument("--page-size", type=int, default=200)
    parser.add_argument("--batch-size", type=int, default=100)
    parser.add_argument("--max-files", type=int)
    parser.add_argument("--max-retries", type=int, default=8)
    parser.add_argument("--min-backoff", type=float, default=1.0)
    parser.add_argument("--max-backoff", type=float, default=32.0)
    parser.add_argument("--send-notification-email", action="store_true", default=True)
    parser.add_argument("--no-send-notification-email", dest="send_notification_email", action="store_false")
    parser.add_argument("--log-file", default=str(base_dir / "request.log"))
    parser.add_argument("--result-file", default=str(base_dir / "request_results.jsonl"))
    parser.add_argument("--dry-run", action="store_true")
    return parser.parse_args()

def guess_credentials_file(auth_mode: str) -> Optional[str]:
    base_dir = Path(__file__).resolve().parent
    if auth_mode == "service-account":
        patterns = [
            "service-account.json",
            "*service-account*.json",
            "*.service-account.json",
        ]
    else:
        patterns = [
            "cred.json",
            "oauth-client.json",
            "credentials.json",
            "client_secret*.json",
            "*oauth*.json",
        ]
    for pattern in patterns:
        matches = sorted(base_dir.glob(pattern))
        if matches:
            return str(matches[0])
    return None

def prompt_value(label: str, current_value: Optional[str], default: Optional[str] = None) -> str:
    if current_value:
        return current_value
    if default:
        entered = input(f"{label} [{default}]: ").strip()
        return entered or default
    while True:
        entered = input(f"{label}: ").strip()
        if entered:
            return entered

def load_json_file(path: str) -> Dict[str, Any]:
    try:
        payload = json.loads(Path(path).read_text(encoding="utf-8"))
    except Exception:
        return {}
    return payload if isinstance(payload, dict) else {}

def save_json_file(path: str, payload: Dict[str, Any]) -> None:
    config_path = Path(path)
    config_path.parent.mkdir(parents=True, exist_ok=True)
    config_path.write_text(json.dumps(payload, ensure_ascii=True, indent=2), encoding="utf-8")

def normalize_args(args: argparse.Namespace) -> argparse.Namespace:
    saved_config = load_json_file(args.config_file)
    guessed_credentials = guess_credentials_file(args.auth_mode)
    saved_credentials = saved_config.get("credentials_file")
    saved_folder_id = saved_config.get("folder_id")
    saved_target_owner_email = saved_config.get("target_owner_email")
    saved_subject_email = saved_config.get("subject_email")
    saved_redirect_uri = saved_config.get("oauth_redirect_uri")
    args.credentials_file = prompt_value(
        "Path credentials file", args.credentials_file, saved_credentials or guessed_credentials,
    )
    if args.auth_mode == "service-account":
        args.subject_email = prompt_value(
            "Email akun sumber untuk impersonation", args.subject_email, saved_subject_email
        )
    args.folder_id = prompt_value("Folder ID to process", args.folder_id, saved_folder_id)
    args.target_owner_email = prompt_value(
        "Target owner email", args.target_owner_email, saved_target_owner_email
    )
    if args.auth_mode == "oauth" and not args.oauth_redirect_uri:
        args.oauth_redirect_uri = prompt_value(
            "OAuth redirect URI", args.oauth_redirect_uri, saved_redirect_uri or "http://localhost:8080"
        )
    save_json_file(
        args.config_file,
        {
            "credentials_file": args.credentials_file,
            "folder_id": args.folder_id,
            "target_owner_email": args.target_owner_email,
            "subject_email": args.subject_email,
            "oauth_redirect_uri": args.oauth_redirect_uri,
        },
    )
    return args

def get_credentials(args: argparse.Namespace) -> Credentials:
    creds = None
    if args.auth_mode == "service-account":
        creds = service_account.Credentials.from_service_account_file(
            args.credentials_file, scopes=SCOPES, subject=args.subject_email
        )
    else:
        if os.path.exists(args.token_file):
            creds = Credentials.from_authorized_user_file(args.token_file, SCOPES)
        if not creds or not creds.valid:
            if creds and creds.expired and creds.refresh_token:
                creds.refresh(Request())
            else:
                flow = InstalledAppFlow.from_client_secrets_file(
                    args.credentials_file, SCOPES, redirect_uri=args.oauth_redirect_uri
                )
                creds = flow.run_local_server(port=0)
            with open(args.token_file, "w") as token:
                token.write(creds.to_json())
    return creds

def get_files(service, folder_id: str, page_size: int) -> Iterable[Dict[str, Any]]:
    query = f"'{folder_id}' in parents and trashed = false"
    page_token = None
    while True:
        response = (
            service.files()
            .list(
                q=query,
                pageSize=page_size,
                fields="nextPageToken, files(id, name, mimeType, parents)",
                supportsAllDrives=True,
                includeItemsFromAllDrives=True,
            )
            .execute()
        )
        yield from response.get("files", [])
        page_token = response.get("nextPageToken")
        if not page_token:
            break

def retry(func: Callable[..., Any]) -> Callable[..., Any]:
    def wrapper(*args, **kwargs):
        args_obj = args[0] if args else None
        max_retries = args_obj.max_retries if args_obj else 8
        min_backoff = args_obj.min_backoff if args_obj else 1.0
        max_backoff = args_obj.max_backoff if args_obj else 32.0
        delay = min_backoff
        for attempt in range(max_retries + 1):
            try:
                return func(*args, **kwargs)
            except HttpError as e:
                error_details = e.error_details[0] if e.error_details else {}
                reason = error_details.get("reason")
                status_code = e.status_code
                if reason in NON_RETRYABLE_REASONS:
                    raise
                if status_code not in RETRYABLE_STATUS and reason not in RETRYABLE_REASONS:
                    raise
                if attempt == max_retries:
                    raise
                logging.warning(
                    f"Retryable error {status_code} ({reason}) on attempt {attempt + 1}/{max_retries + 1}, backing off {delay}s"
                )
                time.sleep(delay + random.uniform(0, 0.5))
                delay = min(delay * 2, max_backoff)
    return wrapper

@retry
def add_permission(service, file_id: str, email: str, role: str, send_notifications: bool) -> Dict[str, Any]:
    if role == "pendingOwner":
        permission = {"type": "user", "role": role, "emailAddress": email}
    else:
        permission = {"type": "user", "role": role, "emailAddress": email}
    return (
        service.permissions()
        .create(
            fileId=file_id,
            body=permission,
            supportsAllDrives=True,
            sendNotificationEmail=send_notifications,
            transferOwnership=(role == "pendingOwner"),
        )
        .execute()
    )

def process_file(args, service, file: Dict[str, Any]) -> Dict[str, Any]:
    file_id = file["id"]
    file_name = file["name"]
    try:
        logging.info(f"Processing file {file_id} {file_name}")
        if args.dry_run:
            return {"file_id": file_id, "name": file_name, "status": "dry-run", "message": "Dry run skipped"}
        # First add editor permission
        add_permission(args, service, file_id, args.target_owner_email, "writer", args.send_notification_email)
        # Then transfer ownership
        add_permission(args, service, file_id, args.target_owner_email, "pendingOwner", args.send_notification_email)
        logging.info(f"Successfully processed file {file_id} {file_name}")
        return {"file_id": file_id, "name": file_name, "status": "success", "message": "Ownership transferred"}
    except HttpError as e:
        error_msg = str(e)
        logging.error(f"Failed to process file {file_id} {file_name}: {error_msg}")
        return {"file_id": file_id, "name": file_name, "status": "failed", "message": error_msg}

def main():
    args = parse_args()
    args = normalize_args(args)
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(message)s",
        handlers=[logging.FileHandler(args.log_file), logging.StreamHandler()],
    )
    creds = get_credentials(args)
    service = build("drive", "v3", credentials=creds)
    files = get_files(service, args.folder_id, args.page_size)
    if args.max_files:
        files = list(files)[:args.max_files]
    with open(args.result_file, "a", encoding="utf-8") as f:
        for file in files:
            result = process_file(args, service, file)
            f.write(json.dumps(result, ensure_ascii=False) + "\n")

if __name__ == "__main__":
    main()

问题根源分析

这个403错误的核心原因是：在将目标用户设置为待处理所有者（pendingOwner）之前，该用户必须已经拥有文件的writer（编辑）权限。你虽然给文件夹加了Editor权限，但可能存在以下情况：

文件夹内的文件没有继承父文件夹的权限（比如部分文件是手动上传的私有文件，或者权限继承被关闭）
代码中尝试设置pendingOwner的时机过早，writer权限还没来得及在Google服务器生效
代码逻辑中可能遗漏了对嵌套文件夹内文件的权限处理

修复步骤

1. 确保每个文件都有明确的writer权限

不要依赖文件夹权限继承，要对每个要转移的文件单独添加Editor权限，并且增加检查逻辑确认权限已生效：

@retry
def check_permission_exists(service, file_id: str, email: str, role: str) -> bool:
    permissions = service.permissions().list(
        fileId=file_id,
        supportsAllDrives=True,
        fields="permissions(id,emailAddress,role)"
    ).execute().get("permissions", [])
    for perm in permissions:
        if perm.get("emailAddress") == email and perm.get("role") == role:
            return True
    return False

# 修改process_file中的逻辑：
def process_file(args, service, file: Dict[str, Any]) -> Dict[str, Any]:
    file_id = file["id"]
    file_name = file["name"]
    try:
        logging.info(f"Processing file {file_id} {file_name}")
        if args.dry_run:
            return {"file_id": file_id, "name": file_name, "status": "dry-run", "message": "Dry run skipped"}
        
        # 先检查是否已有writer权限，没有再添加
        if not check_permission_exists(service, file_id, args.target_owner_email, "writer"):
            add_permission(args, service, file_id, args.target_owner_email, "writer", args.send_notification_email)
            # 增加延迟确保权限生效
            time.sleep(2)
        
        # 再执行所有权转移
        add_permission(args, service, file_id, args.target_owner_email, "pendingOwner", args.send_notification_email)
        logging.info(f"Successfully processed file {file_id} {file_name}")
        return {"file_id": file_id, "name": file_name, "status": "success", "message": "Ownership transferred"}
    except HttpError as e:
        error_msg = str(e)
        logging.error(f"Failed to process file {file_id} {file_name}: {error_msg}")
        return {"file_id": file_id, "name": file_name, "status": "failed", "message": error_msg}

2. 递归处理嵌套文件夹

当前代码只处理目标文件夹下的一级文件，需要修改get_files函数来遍历所有子文件夹：

def get_files(service, folder_id: str, page_size: int) -> Iterable[Dict[str, Any]]:
    query = f"'{folder_id}' in parents and trashed = false"
    page_token = None
    while True:
        response = (
            service.files()
            .list(
                q=query,
                pageSize=page_size,
                fields="nextPageToken, files(id, name, mimeType, parents)",
                supportsAllDrives=True,
                includeItemsFromAllDrives=True,
            )
            .execute()
        )
        for file in response.get("files", []):
            yield file
            # 如果是文件夹，递归遍历内部文件
            if file["mimeType"] == FOLDER_MIME:
                yield from get_files(service, file["id"], page_size)
        page_token = response.get("nextPageToken")
        if not page_token:
            break

3. 调整重试逻辑

把pendingOwnerWriterRequired从不可重试列表中移除，让代码可以在权限生效后自动重试：

NON_RETRYABLE_REASONS = set()  # 移除原有的pendingOwnerWriterRequired

验证方法

先手动对单个文件执行"添加Editor权限→转移所有权"的流程，确认操作可行
用--dry-run模式运行代码，检查要处理的文件列表是否完整
查看日志，确认每个文件都先成功添加了writer权限，再进行所有权转移

内容的提问来源于stack exchange，提问作者bruh

使用Python调用Google Drive API V3转移文件所有权时遭遇403(pendingOwnerWriterRequired)权限错误的问题咨询

错误日志 #

已尝试操作 #

使用的代码 #

问题根源分析 #

修复步骤 #

1. 确保每个文件都有明确的writer权限 #

2. 递归处理嵌套文件夹 #

3. 调整重试逻辑 #

验证方法 #