使用Google服务账号域范围委托传照片遇401错误求助

我正尝试使用具备域范围委派权限的Google服务账号，以user@domain.com账号身份向Google Photos服务上传图片。但执行代码时，始终收到401错误提示：Authentication session is not defined。请问哪里操作有误？

import os
import requests
from google.oauth2 import service_account
from googleapiclient import discovery

class GooglePhotosApi:
    def __init__(self):
        self.SERVICE_ACCOUNT_FILE = (
            os.path.dirname(os.path.abspath(__file__)) + "/service_account.json"
        )
        self.SCOPES = [
            "https://www.googleapis.com/auth/photoslibrary",
            "https://www.googleapis.com/auth/photoslibrary.sharing",
            "https://www.googleapis.com/auth/photoslibrary.appendonly",
            "https://www.googleapis.com/auth/iam",
            "https://www.googleapis.com/auth/cloud-platform",
        ]

def auth_service_account(self):
        self.credentials = service_account.Credentials.from_service_account_file(
            self.SERVICE_ACCOUNT_FILE,
            scopes=self.SCOPES,
            subject="*user@domain.com*",
        )

        self.access_token = self.credentials._make_authorization_grant_assertion()
        return self.credentials

def upload_media(self, img):
    self.service = discovery.build(
        "photoslibrary", "v1", credentials=self.credentials, static_discovery=False
    )

    # step 1: Upload byte data to Google Server
    image_dir = os.path.dirname(os.path.abspath(__file__)) + "/"
    upload_url = "https://photoslibrary.googleapis.com/v1/uploads"

    headers = {
        # "Authorization": "Bearer {}".format(self.access_token.decode("utf-8")),
        "Authorization": "Bearer " + str(self.access_token),
        "Content-type": "application/octet-stream",
        "X-Goog-Upload-Protocol": "raw",
    }

    image_file = os.path.join(image_dir, "img.jpg")
    headers["X-Goog-Upload-File-Name"] = "img.jpg"

    img = open(image_file, "rb").read()
    response = requests.post(upload_url, data=img, headers=headers)

    request_body = {
        "newMediaItems": [
            {
                "description": "Static Name",
                "simpleMediaItem": {
                    "uploadToken": response.content.decode("utf-8")
                },
            }
        ]
    }

    upload_response = (
        self.service.mediaItems().batchCreate(body=request_body).execute()
    )

    return upload_response  

• https://www.googleapis.com/auth/drive
• https://www.googleapis.com/auth/calendar
• https://www.googleapis.com/auth/photoslibrary
• https://www.googleapis.com/auth/photoslibrary.sharing
• https://www.googleapis.com/auth/photoslibrary.appendonly
• https://www.googleapis.com/auth/cloud-platform
• https://www.googleapis.com/auth/iam

1. 手动生成token的方式错误 #

原代码调用_make_authorization_grant_assertion()返回的是JWT断言，并非可直接使用的access_token。Google服务需要的是经过OAuth2流程生成的有效access_token，正确做法是调用credentials.refresh(requests.Request())让官方库自动处理token的生成与刷新。

2. Authorization头使用错误 #

上传文件时的Authorization头应使用self.credentials.token，这是刷新后得到的有效token，而非手动生成的JWT断言。

3. 类方法缩进错误 #

原代码中auth_service_account和upload_media方法未缩进在GooglePhotosApi类内部，导致无法正确调用类属性，修正后需将方法缩进至类范围内。

4. 多余字符干扰 #

subject="*user@domain.com*"中的星号是多余的，直接填写目标用户邮箱即可。

修正后的代码 #

import os
import requests
from google.oauth2 import service_account
from googleapiclient import discovery

class GooglePhotosApi:
    def __init__(self):
        self.SERVICE_ACCOUNT_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "service_account.json")
        self.SCOPES = [
            "https://www.googleapis.com/auth/photoslibrary",
            "https://www.googleapis.com/auth/photoslibrary.appendonly",
        ]
        self.credentials = None
        self.service = None

    def auth_service_account(self):
        self.credentials = service_account.Credentials.from_service_account_file(
            self.SERVICE_ACCOUNT_FILE,
            scopes=self.SCOPES,
            subject="user@domain.com",
        )
        # 刷新获取有效access_token
        self.credentials.refresh(requests.Request())
        return self.credentials

    def upload_media(self):
        if not self.credentials:
            self.auth_service_account()
            
        self.service = discovery.build(
            "photoslibrary", "v1", credentials=self.credentials, static_discovery=False
        )

        upload_url = "https://photoslibrary.googleapis.com/v1/uploads"
        image_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), "img.jpg")

        headers = {
            "Authorization": f"Bearer {self.credentials.token}",
            "Content-Type": "application/octet-stream",
            "X-Goog-Upload-Protocol": "raw",
            "X-Goog-Upload-File-Name": "img.jpg"
        }

        with open(image_file, "rb") as f:
            img_data = f.read()
            
        response = requests.post(upload_url, data=img_data, headers=headers)
        if response.status_code != 200:
            raise Exception(f"文件上传失败: {response.text}")

        upload_token = response.content.decode("utf-8")
        request_body = {
            "newMediaItems": [
                {
                    "description": "Static Name",
                    "simpleMediaItem": {"uploadToken": upload_token}
                }
            ]
        }

        return self.service.mediaItems().batchCreate(body=request_body).execute()

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