實作伺服器端授權

傳送至 Gmail API 的要求必須使用 OAuth 2.0 憑證授權。如果應用程式需要代表使用者存取 Google API (例如使用者離線時),您就應使用伺服器端流程。此方法需要將一次性授權碼從用戶端傳遞至伺服器,這段程式碼可用於取得伺服器的存取權杖和更新權杖。

如要進一步瞭解伺服器端 Google OAuth 2.0 實作方式,請參閱「針對網路伺服器應用程式使用 OAuth 2.0」。

目錄

建立用戶端 ID 和用戶端密鑰

如要開始使用 Gmail API,請使用設定工具;這項工具會逐步引導您在 Google API 控制台中建立專案、啟用 API,並建立憑證。

  1. 在「憑證」頁面中,依序按一下「建立憑證」>「OAuth 用戶端 ID」建立 OAuth 2.0 憑證,或依序按一下「建立憑證」>「服務帳戶金鑰」建立服務帳戶。
  2. 如果您已建立 OAuth 用戶端 ID,請選取應用程式類型。
  3. 填寫表單,然後按一下「建立」

應用程式的用戶端 ID 和服務帳戶金鑰現在會列在「憑證」頁面。如需詳細資訊,請按一下用戶端 ID。參數會因 ID 類型而異,但可能包括電子郵件地址、用戶端密鑰、JavaScript 來源或重新導向 URI。

請記下用戶端 ID,因為您稍後需要將其加入程式碼。

處理授權要求

使用者首次載入應用程式時,系統會顯示對話方塊,要求他們授予應用程式存取其 Gmail 帳戶的權限,並提供所需的權限範圍。在初始授權完成後,只有在應用程式的用戶端 ID 或要求的權限範圍有所變動時,使用者才會看到權限對話方塊。

驗證使用者

這項初始登入作業會傳回授權結果物件,如果成功,則會包含授權碼

將授權碼換成存取權杖

授權碼是一次性程式碼,可讓伺服器交換存取權杖。這個存取權權杖會傳遞至 Gmail API,授予應用程式對使用者資料的限時存取權。

如果應用程式需要 offline 存取權,則在應用程式首次交換授權碼時,也會收到重新整理權杖,可用於在先前權杖到期後,接收新的存取權杖。應用程式會儲存這項重新整理權杖 (通常儲存在伺服器上的資料庫中),以供日後使用。

以下程式碼範例示範如何使用 offline 存取權,將授權碼換成存取權杖,並儲存更新權杖。

Python

CLIENTSECRETS_LOCATION 值替換為 credentials.json 檔案的位置。

import logging
from oauth2client.client import flow_from_clientsecrets
from oauth2client.client import FlowExchangeError
from oauth2client.client import Credentials # Needed for type hinting/usage in comments
from googleapiclient.discovery import build
from googleapiclient import errors as google_api_errors
import httplib2

# Path to credentials.json which should contain a JSON document such as:
#   {
#     "web": {
#       "client_id": "[[YOUR_CLIENT_ID]]",
#       "client_secret": "[[YOUR_CLIENT_SECRET]]",
#       "redirect_uris": [],
#       "auth_uri": "https://accounts.google.com/o/oauth2/auth",
#       "token_uri": "https://accounts.google.com/o/oauth2/token"
#     }
#   }
CLIENTSECRETS_LOCATION = '<PATH/TO/CLIENT_SECRETS.JSON>'
REDIRECT_URI = '<YOUR_REGISTERED_REDIRECT_URI>'
SCOPES = [
    'https://www.googleapis.com/auth/gmail.readonly',
    'https://www.googleapis.com/auth/userinfo.email',
    'https://www.googleapis.com/auth/userinfo.profile',
    # Add other requested scopes.
]

class GetCredentialsException(Exception):
  """Error raised when an error occurred while retrieving credentials.

  Attributes:
    authorization_url: Authorization URL to redirect the user to in order to
                      request offline access.
  """
  def __init__(self, authorization_url):
    """Construct a GetCredentialsException."""
    super().__init__(f"Authorization URL: {authorization_url}")
    self.authorization_url = authorization_url

class CodeExchangeException(GetCredentialsException):
  """Error raised when a code exchange has failed."""
  pass

class NoRefreshTokenException(GetCredentialsException):
  """Error raised when no refresh token has been found."""
  pass

class NoUserIdException(Exception):
  """Error raised when no user ID could be retrieved."""
  pass

def get_stored_credentials(user_id):
  """Retrieved stored credentials for the provided user ID.

  Args:
    user_id: User's ID.

  Returns:
    Stored oauth2client.client.OAuth2Credentials if found, None otherwise.

  Raises:
    NotImplementedError: This function has not been implemented.
  """
  # TODO: Implement this function to work with your database.
  #       To instantiate an OAuth2Credentials instance from a Json
  #       representation, use the oauth2client.client.Credentials.new_from_json
  #       class method. (oauth2client.client.Credentials needs to be imported)
  #       Example:
  #       from oauth2client.client import Credentials
  #       json_creds = load_from_db(user_id)
  #       if json_creds:
  #           return Credentials.new_from_json(json_creds)
  #       return None
  raise NotImplementedError()

def store_credentials(user_id, credentials):
  """Store OAuth 2.0 credentials in the application's database.

  This function stores the provided OAuth 2.0 credentials using the user ID as
  key.

  Args:
    user_id: User's ID.
    credentials: OAuth 2.0 credentials to store.

  Raises:
    NotImplementedError: This function has not been implemented.
  """
  # TODO: Implement this function to work with your database.
  #       To retrieve a Json representation of the credentials instance, call the
  #       credentials.to_json() method.
  #       Example:
  #       save_to_db(user_id, credentials.to_json())
  raise NotImplementedError()

def exchange_code(authorization_code):
  """Exchange an authorization code for OAuth 2.0 credentials.

  Args:
    authorization_code: Authorization code to exchange for OAuth 2.0
                        credentials.

  Returns:
    oauth2client.client.OAuth2Credentials instance.

  Raises:
    CodeExchangeException: an error occurred.
  """
  flow = flow_from_clientsecrets(CLIENTSECRETS_LOCATION, ' '.join(SCOPES))
  flow.redirect_uri = REDIRECT_URI
  try:
    credentials = flow.step2_exchange(authorization_code)
    return credentials
  except FlowExchangeError as error:
    logging.error('An error occurred: %s', error)
    raise CodeExchangeException(None)

def get_user_info(credentials):
  """Send a request to the UserInfo API to retrieve the user's information.

  Args:
    credentials: oauth2client.client.OAuth2Credentials instance to authorize the
              request.

  Returns:
    User information as a dict.
  """
  user_info_service = build(
      serviceName='oauth2', version='v2',
      http=credentials.authorize(httplib2.Http()))
  user_info = None
  try:
    user_info = user_info_service.userinfo().get().execute()
  except google_api_errors.HttpError as e:
    logging.error('An error occurred: %s', e)
  if user_info and user_info.get('id'):
    return user_info
  else:
    raise NoUserIdException()

def get_authorization_url(email_address, state):
  """Retrieve the authorization URL.

  Args:
    email_address: User's e-mail address.
    state: State for the authorization URL.

  Returns:
    Authorization URL to redirect the user to.
  """
  flow = flow_from_clientsecrets(CLIENTSECRETS_LOCATION, ' '.join(SCOPES))
  flow.params['access_type'] = 'offline'
  flow.params['approval_prompt'] = 'force'
  flow.params['user_id'] = email_address
  flow.params['state'] = state
  # The step1_get_authorize_url method uses the flow.redirect_uri attribute.
  flow.redirect_uri = REDIRECT_URI
  return flow.step1_get_authorize_url()

def get_credentials(authorization_code, state):
  """Retrieve credentials using the provided authorization code.

  This function exchanges the authorization code for an access token and queries
  the UserInfo API to retrieve the user's e-mail address.

  If a refresh token has been retrieved along with an access token, it is stored
  in the application database using the user's e-mail address as key.

  If no refresh token has been retrieved, the function checks in the application
  database for one and returns it if found or raises a NoRefreshTokenException
  with the authorization URL to redirect the user to.

  Args:
    authorization_code: Authorization code to use to retrieve an access token.
    state: State to set to the authorization URL in case of error.

  Returns:
    oauth2client.client.OAuth2Credentials instance containing an access and
    refresh token.

  Raises:
    CodeExchangeError: Could not exchange the authorization code.
    NoRefreshTokenException: No refresh token could be retrieved from the
                          available sources.
  """
  email_address = ''
  try:
    credentials = exchange_code(authorization_code)
    user_info = get_user_info(credentials) # Can raise NoUserIdException or google_api_errors.HttpError
    email_address = user_info.get('email')
    user_id = user_info.get('id')
    if credentials.refresh_token is not None:
      store_credentials(user_id, credentials)
      return credentials
    else:
      credentials = get_stored_credentials(user_id)
      if credentials and credentials.refresh_token is not None:
        return credentials
  except CodeExchangeException as error:
    logging.error('An error occurred during code exchange.')
    # Drive apps should try to retrieve the user and credentials for the current
    # session.
    # If none is available, redirect the user to the authorization URL.
    error.authorization_url = get_authorization_url(email_address, state)
    raise error
  except NoUserIdException:
    logging.error('No user ID could be retrieved.')
  # No refresh token has been retrieved.
  authorization_url = get_authorization_url(email_address, state)
  raise NoRefreshTokenException(authorization_url)

使用儲存的憑證進行授權

當使用者在首次授權流程順利完成後造訪您的應用程式時,應用程式可以使用儲存的更新權杖來授權要求,而無須再次提示使用者。

如果您已為使用者進行驗證,應用程式就能從資料庫中擷取更新權杖,並將權杖儲存在伺服器端工作階段中。如果重新整理權杖遭到撤銷或無效,您就需要擷取這項資訊並採取適當行動。

使用 OAuth 2.0 憑證

擷取 OAuth 2.0 憑證後 (如上一節所示),即可用於授權 Gmail 服務物件,並將要求傳送至 API。

將服務物件例項化

這個程式碼範例說明如何建立服務物件例項,然後授權該例項提出 API 要求。

Python

from apiclient.discovery import build
# ...

def build_service(credentials):
  """Build a Gmail service object.

  Args:
    credentials: OAuth 2.0 credentials.

  Returns:
    Gmail service object.
  """
  http = httplib2.Http()
  http = credentials.authorize(http)
  return build('gmail', 'v1', http=http)

傳送授權要求,並檢查是否有遭到撤銷的憑證

以下程式碼片段會使用已授權的 Gmail 服務執行個體,擷取郵件清單。

如果發生錯誤,程式碼會檢查 HTTP 401 狀態碼,並將使用者重新導向至授權網址。

如需更多 Gmail API 作業,請參閱 API 參考資料

Python

from googleapiclient import errors

# ...

def ListMessages(service, user, query=''):
  """Gets a list of messages.

  Args:
    service: Authorized Gmail API service instance.
    user: The email address of the account.
    query: String used to filter messages returned.
          Eg.- 'label:UNREAD' for unread Messages only.

  Returns:
    List of messages that match the criteria of the query. Note that the
    returned list contains Message IDs, you must use get with the
    appropriate id to get the details of a Message.
  """
  try:
    response = service.users().messages().list(userId=user, q=query).execute()
    messages = []
    if 'messages' in response:
      messages.extend(response['messages'])

    while 'nextPageToken' in response:
      page_token = response['nextPageToken']
      response = service.users().messages().list(userId=user, q=query,
                                        pageToken=page_token).execute()
      if 'messages' in response:
        messages.extend(response['messages'])

    return messages
  except errors.HttpError as error:
    print('An error occurred: %s' % error)
    if error.resp.status == 401:
      # Credentials have been revoked.
      # TODO: Redirect the user to the authorization URL.
      raise NotImplementedError()

後續步驟

授權 Gmail API 要求後,您就可以開始處理郵件、會話串和標籤,如開發人員指南章節所述。

如要進一步瞭解可用的 API 方法,請參閱 API 參考資料