多Colab账户在共享Google Drive创建同名文件夹的一致性问题求解

问题背景 #

存在一个名为shared_folder的Google Drive共享文件夹，两个不同账户可通过Colab的Jupyter笔记本运行Python代码创建子文件夹：

• 单账户重复创建同名文件夹时，系统会正常触发错误：

import os
os.mkdir('/content/drive/MyDrive/shared_folder/test_folder')
os.mkdir('/content/drive/MyDrive/shared_folder/test_folder')

FileExistsError: [Errno 17] File exists: '/content/drive/MyDrive/shared_folder/test_folder'

• 但两个账户短时间内同时创建同名文件夹时，会出现异常：
  • Google Drive云端出现两个同名文件夹；
  • Colab的Files面板中显示两个路径：/content/drive/MyDrive/shared_folder/test_folder 和 /content/drive/MyDrive/shared_folder/test_folder (1)；
  • 向两个文件夹分别添加first.txt和second.txt后，不同账户看到的(1)后缀对应内容不一致，比如甲账户看到first.txt在test_folder，乙账户看到它在test_folder (1)。

该问题源于Google Drive账户间的文件列表同步延迟，导致本地挂载的缓存无法及时获取其他账户的操作结果。

解决方案 #

1. 用Google Drive API实现原子性检查与创建 #

放弃依赖本地挂载缓存的os.mkdir，直接调用Google Drive云端API做查询和创建，利用API的原子性避免冲突。以下是基于pydrive2库的实现：

from pydrive2.auth import GoogleAuth
from pydrive2.drive import GoogleDrive

# Colab环境下认证（本地环境可替换为LocalWebserverAuth）
gauth = GoogleAuth()
gauth.ColabAuth()
drive = GoogleDrive(gauth)

# 替换为你的shared_folder云端ID（可从Drive文件夹URL中获取）
shared_folder_id = "替换为实际文件夹ID"
target_folder_name = "test_folder"

# 云端查询是否存在同名文件夹
query = f"title='{target_folder_name}' and '{shared_folder_id}' in parents and mimeType='application/vnd.google-apps.folder' and trashed=false"
existing_folders = drive.ListFile({'q': query}).GetList()

if len(existing_folders) > 0:
    raise FileExistsError(f"文件夹 {target_folder_name} 已存在")
else:
    # 创建新文件夹
    folder_metadata = {
        'title': target_folder_name,
        'parents': [{'id': shared_folder_id}],
        'mimeType': 'application/vnd.google-apps.folder'
    }
    new_folder = drive.CreateFile(folder_metadata)
    new_folder.Upload()
    print(f"文件夹 {target_folder_name} 创建成功")

说明：直接查询云端文件状态，完全绕过本地挂载的缓存延迟，查询与创建逻辑在云端层面具备原子性，能准确检测到其他账户刚创建的文件夹。

2. 基于锁文件的互斥创建机制 #

通过在共享文件夹中创建临时锁文件，确保同一时间只有一个账户执行文件夹创建操作，避免并行冲突：

import os
import time
from pathlib import Path

target_folder = Path('/content/drive/MyDrive/shared_folder/test_folder')
lock_file = Path('/content/drive/MyDrive/shared_folder/.create_lock')

# 原子性创建锁文件（O_EXCL确保只有一个进程能创建成功）
try:
    fd = os.open(lock_file, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
    os.close(fd)
except FileExistsError:
    # 锁存在时等待后重试检查
    time.sleep(2)
    if target_folder.exists():
        raise FileExistsError(f"文件夹 {target_folder} 已存在")
    # 处理异常遗留的锁文件（谨慎操作，确保无其他进程在使用）
    os.remove(lock_file)
    fd = os.open(lock_file, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
    os.close(fd)

try:
    # 再次检查目标文件夹是否存在
    if target_folder.exists():
        raise FileExistsError(f"文件夹 {target_folder} 已存在")
    # 创建文件夹
    os.mkdir(target_folder)
finally:
    # 释放锁文件
    if lock_file.exists():
        os.remove(lock_file)

说明：利用os.open的O_EXCL参数实现锁文件的原子性创建，确保同一时间只有一个进程进入创建流程，再结合本地检查，减少同步延迟带来的冲突。

3. 强制刷新Drive挂载缓存 #

在创建文件夹前重新挂载Drive，强制拉取最新的云端文件列表，减少本地缓存与云端的不一致：

import os
from google.colab import drive

# 强制重新挂载Drive，刷新文件列表
drive.mount('/content/drive', force_remount=True)

target_folder = '/content/drive/MyDrive/shared_folder/test_folder'
if os.path.exists(target_folder):
    raise FileExistsError(f"文件夹 {target_folder} 已存在")
os.mkdir(target_folder)

说明：该方法简单但可靠性略低，若两个账户几乎同时刷新并创建，仍可能出现冲突，适合冲突概率较低的场景。

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