Module pyfreedb.providers.google.auth

class GoogleAuthClient(abc.ABC):
    """An abstraction layer that represents way to authenticate with Google APIs."""

    @abc.abstractmethod
    def credentials(self) -> Union[credentials.Credentials, service_account.Credentials]:
        pass

class OAuth2GoogleAuthClient(GoogleAuthClient):
    def __init__(self, creds: Credentials) -> None:
        """Initialise auth client instance to perform authentication using OAuth2.

        Client is recommended to not instantiate this class directly, use `from_authorized_user_info` and
        `from_authorized_user_file` constructor instead.
        """
        if creds.expired and creds.refresh_token:
            creds.refresh(Request())

        self._creds = creds

    @classmethod
    def from_authorized_user_info(
        cls,
        authorized_user_info: Dict[str, str],
        scopes: Optional[List[str]] = None,
    ) -> "OAuth2GoogleAuthClient":
        """Initialise the auth client using the provided dict.

        Args:
            authorized_user_info: The user authentication info in dict form.
            scopes: List of permitted operation by the authentication info.

        Returns:
            OAuth2GoogleAuthClient: The auth client instance.
        """
        creds = Credentials.from_authorized_user_info(authorized_user_info, scopes=scopes)
        return cls(creds)

    @classmethod
    def from_authorized_user_file(
        cls,
        authorized_user_file: str,
        client_secret_filename: Optional[str] = None,
        scopes: Optional[List[str]] = None,
    ) -> "OAuth2GoogleAuthClient":
        """Initialise the auth client by reading the authentication info from files.

        If the file given in `authorized_user_file` is not found we will trigger the OAuth2 authentication flow (that
        requires interaction via browser) and will save the authentication info in the given `authorized_user_file.

        Args:
            authorized_user_file: The filename of the user authentication info.
            client_secret_filename: The service secret file (obtainable from the Google Credential dashboard).
            scopes: List of permitted operation by the authentication info.

        Returns:
            OAuth2GoogleAuthClient: The auth client instance.
        """
        if os.path.exists(authorized_user_file):
            creds = Credentials.from_authorized_user_file(authorized_user_file, scopes=scopes)
            return cls(creds)

        if not client_secret_filename:
            raise ValueError("client_secret_filename must be set if authorized_user_file is not exists")

        flow = InstalledAppFlow.from_client_secrets_file(client_secret_filename, scopes=scopes)
        creds = flow.run_local_server(port=0)
        with open(authorized_user_file, "w") as user_file:
            user_file.write(creds.to_json())

        return cls(creds)

    def credentials(self) -> Credentials:
        """Returns the authenticated Google credentials.

        Returns:
            google.oauth2.credentials.Credentials: The authenticated Google credentials.
        """
        return self._creds

class ServiceAccountGoogleAuthClient(GoogleAuthClient):
    def __init__(self, creds: service_account.Credentials) -> None:
        """Initialise auth client instance to perform authentication using Service Account.

        Client is recommended to not instantiate this class directly, use `from_service_account_info` and
        `from_service_account_file` constructor instead.
        """
        self._creds = creds

    @classmethod
    def from_service_account_info(
        cls,
        service_account_info: Dict[str, str],
        scopes: Optional[List[str]] = None,
    ) -> "ServiceAccountGoogleAuthClient":
        """Initialise the auth client using the provided service account dict.

        Args:
            service_account_info: The service account info in dict form.
            scopes: List of permitted operation by the authentication info.

        Returns:
            ServiceAccountGoogleAuthClient: The auth client instance.
        """
        creds = service_account.Credentials.from_service_account_info(service_account_info, scopes=scopes)
        return cls(creds)

    @classmethod
    def from_service_account_file(
        cls,
        filename: str,
        scopes: Optional[List[str]] = None,
    ) -> "ServiceAccountGoogleAuthClient":
        """Initialise the auth client by reading the service account info from a file.

        Args:
            filename: The path to file that contains the service account info.
            scopes: List of permitted operation by the authentication info.

        Returns:
            ServiceAccountGoogleAuthClient: The auth client instance.
        """
        creds = service_account.Credentials.from_service_account_file(filename, scopes=scopes)
        return cls(creds)

    def credentials(self) -> service_account.Credentials:
        """Returns the authenticated Google credentials.

        Returns:
            google.oauth2.service_account.Credentials: The authenticated Google credentials.
        """
        return self._creds