authentication.py

import pickle
import os.path
import glob
from typing import List

from googleapiclient import discovery
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
from google.auth.credentials import Credentials


class AuthenticatedService:
    """Handles authentication of the `GoogleCalendar`"""

    _READ_WRITE_SCOPES = 'https://www.googleapis.com/auth/calendar'
    _LIST_ORDERS = ("startTime", "updated")

    def __init__(
            self,
            *,
            credentials: Credentials = None,
            credentials_path: str = None,
            token_path: str = None,
            save_token: bool = True,
            read_only: bool = False,
            authentication_flow_host: str = 'localhost',
            authentication_flow_port: int = 8080,
            authentication_flow_bind_addr: str = None
    ):
        """
        Specify ``credentials`` to use in requests or ``credentials_path`` and ``token_path`` to get credentials from.

        :param credentials:
                Credentials with token and refresh token.
                If specified, ``credentials_path``, ``token_path``, and ``save_token`` are ignored.
                If not specified, credentials are retrieved from "token.pickle" file (specified in ``token_path`` or
                default path) or with authentication flow using secret from "credentials.json" ("client_secret_*.json")
                (specified in ``credentials_path`` or default path)
        :param credentials_path:
                Path to "credentials.json" ("client_secret_*.json") file.
                Default: ~/.credentials/credentials.json or ~/.credentials/client_secret*.json
        :param token_path:
                Existing path to load the token from, or path to save the token after initial authentication flow.
                Default: "token.pickle" in the same directory as the credentials_path
        :param save_token:
                Whether to pickle token after authentication flow for future uses
        :param read_only:
                If require read only access. Default: False
        :param authentication_flow_host:
                Host to receive response during authentication flow
        :param authentication_flow_port:
                Port to receive response during authentication flow
        :param authentication_flow_bind_addr:
                Optional IP address for the redirect server to listen on when it is not the same as host
                (e.g. in a container)
        """

        if credentials:
            self.credentials = self._ensure_refreshed(credentials)
        else:
            credentials_path = credentials_path or self._get_default_credentials_path()
            credentials_dir, credentials_file = os.path.split(credentials_path)
            token_path = token_path or os.path.join(credentials_dir, 'token.pickle')
            scopes = [self._READ_WRITE_SCOPES + ('.readonly' if read_only else '')]

            self.credentials = self._get_credentials(
                token_path,
                credentials_dir,
                credentials_file,
                scopes,
                save_token,
                authentication_flow_host,
                authentication_flow_port,
                authentication_flow_bind_addr
            )

        self.service = discovery.build('calendar', 'v3', credentials=self.credentials)

    @staticmethod
    def _ensure_refreshed(
            credentials: Credentials
    ) -> Credentials:
        if not credentials.valid and credentials.expired:
            credentials.refresh(Request())
        return credentials

    @staticmethod
    def _get_credentials(
            token_path: str,
            credentials_dir: str,
            credentials_file: str,
            scopes: List[str],
            save_token: bool,
            host: str,
            port: int,
            bind_addr: str
    ) -> Credentials:
        credentials = None

        if os.path.exists(token_path):
            with open(token_path, 'rb') as token_file:
                credentials = pickle.load(token_file)

        if not credentials or not credentials.valid:
            if credentials and credentials.expired and credentials.refresh_token:
                credentials.refresh(Request())
            else:
                credentials_path = os.path.join(credentials_dir, credentials_file)
                flow = InstalledAppFlow.from_client_secrets_file(credentials_path, scopes)
                credentials = flow.run_local_server(host=host, port=port, bind_addr=bind_addr)

            if save_token:
                with open(token_path, 'wb') as token_file:
                    pickle.dump(credentials, token_file)

        return credentials

    @staticmethod
    def _get_default_credentials_path() -> str:
        """Checks if `.credentials` folder in home directory exists and contains `credentials.json` or
        `client_secret*.json` file.

        :raises ValueError: if `.credentials` folder does not exist, none of `credentials.json` or `client_secret*.json`
        files do not exist, or there are multiple `client_secret*.json` files.
        :return: expanded path to `credentials.json` or `client_secret*.json` file
        """
        home_dir = os.path.expanduser('~')
        credential_dir = os.path.join(home_dir, '.credentials')
        if not os.path.exists(credential_dir):
            raise FileNotFoundError(f'Default credentials directory "{credential_dir}" does not exist.')
        credential_path = os.path.join(credential_dir, 'credentials.json')
        if os.path.exists(credential_path):
            return credential_path
        else:
            credentials_files = glob.glob(credential_dir + '/client_secret*.json')
            if len(credentials_files) > 1:
                raise ValueError(f"Multiple credential files found in {credential_dir}.\n"
                                 f"Try specifying the credentials file, e.x.:\n"
                                 f"GoogleCalendar(credentials_path='{credentials_files[0]}')")
            elif not credentials_files:
                raise FileNotFoundError(f'Credentials file (credentials.json or client_secret*.json)'
                                        f'not found in the default path: "{credential_dir}".')
            else:
                return credentials_files[0]