feat: Enable constructing GoogleServiceAccountClient with Credentials.from_service_account_info method

examples/ad_manager/authentication/create_ad_manager_client_with_service_account_info.py

@@ -0,0 +1,60 @@
+#!/usr/bin/env python
+#
+# Copyright 2024 Google Inc. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Initializes a AdManagerClient using a Service Account info dictionary."""
+
+import json
+
+from googleads import ad_manager
+from googleads import oauth2
+
+# OAuth2 credential information. In a real application, you'd probably be
+# pulling these values from a credential storage or environment variable.
+# For this example, we'll load them from a JSON file.
+KEY_FILE = 'INSERT_KEY_FILE_PATH'
+
+# Ad Manager API information.
+APPLICATION_NAME = 'INSERT_APPLICATION_NAME_HERE'
+
+
+def main(key_file, application_name):
+  # Load the service account info from a JSON file
+  with open(key_file, 'r') as json_file:
+    service_account_info = json.load(json_file)
+
+  # In a real application, you might get this info from:
+  # - Environment variables
+  # - A secret manager
+  # - A database
+  # - Any other secure storage
+
+  # Create the OAuth2 client using the service account info dictionary
+  oauth2_client = oauth2.GoogleServiceAccountClient.from_service_account_info(
+      service_account_info, oauth2.GetAPIScope('ad_manager'))
+
+  # Create the Ad Manager client
+  ad_manager_client = ad_manager.AdManagerClient(
+      oauth2_client, application_name)
+
+  # Make an API call to verify everything is working
+  networks = ad_manager_client.GetService('NetworkService').getAllNetworks()
+  for network in networks:
+    print('Network with network code "%s" and display name "%s" was found.'
+          % (network['networkCode'], network['displayName']))
+
+
+if __name__ == '__main__':
+  main(KEY_FILE, APPLICATION_NAME) 

googleads/oauth2.py

@@ -251,8 +251,8 @@ class GoogleServiceAccountClient(GoogleRefreshableOAuth2Client):
   """A simple client for using OAuth2 for Google APIs with a service account.
 
   This class is not capable of supporting any flows other than generating
-  credentials from a service account email and key file. This is incompatible
-  with App Engine.
+  credentials from a service account email and key file or service account info 
+  dictionary. Using a key file is incompatible with App Engine.
 
   Attributes:
     proxy_info: A ProxyInfo instance used for refresh requests.
@@ -288,6 +288,30 @@ def __init__(self, key_file, scope, sub=None, proxy_config=None):
                          googleads.common.ProxyConfig())
     self.Refresh()
 
+  @classmethod
+  def from_service_account_info(cls, service_account_info, scope, sub=None, proxy_config=None):
+    """Creates a GoogleServiceAccountClient from a service account info dict.
+
+    Args:
+      service_account_info: A dict containing the service account info in Google format.
+      scope: The scope of the API you're authorizing for.
+      [optional]
+      sub: A string containing the email address of a user account you want to
+           impersonate.
+      proxy_config: A googleads.common.ProxyConfig instance.
+
+    Returns:
+      A GoogleServiceAccountClient instance.
+    """
+    instance = cls.__new__(cls)
+    instance.creds = (
+        google.oauth2.service_account.Credentials.from_service_account_info(
+            service_account_info, scopes=[scope], subject=sub))
+    instance.proxy_config = (proxy_config if proxy_config else
+                       googleads.common.ProxyConfig())
+    instance.Refresh()
+    return instance
+
   def CreateHttpHeader(self):
     """Creates an OAuth2 HTTP header.
 

tests/oauth2_test.py

@@ -303,6 +303,43 @@ def testCreateDelegatedGoogleServiceAccountClient(self):
           self.key_file_path, scopes=[self.scope],
           subject=self.delegated_account)
 
+  def testCreateFromServiceAccountInfo(self):
+    with mock.patch('google.oauth2.service_account.Credentials') as mock_cred:
+      # Mock service account info dictionary
+      service_account_info = {'private_key': 'fake_key', 'client_email': 'test@example.com'}
+      mock_cred.from_service_account_info.return_value = self.mock_credentials_instance
+
+      # Create a GoogleServiceAccountClient using from_service_account_info
+      client = googleads.oauth2.GoogleServiceAccountClient.from_service_account_info(
+          service_account_info, self.scope, sub=self.delegated_account)
+
+      # Verify the credentials were instantiated correctly
+      mock_cred.from_service_account_info.assert_called_once_with(
+          service_account_info, scopes=[self.scope],
+          subject=self.delegated_account)
+
+      # Verify the client has the expected properties
+      self.assertEqual(client.creds, self.mock_credentials_instance)
+      self.assertEqual(client.proxy_config.proxies, {})
+
+  def testCreateFromServiceAccountInfoWithProxyConfig(self):
+    with mock.patch('google.oauth2.service_account.Credentials') as mock_cred:
+      # Mock service account info dictionary
+      service_account_info = {'private_key': 'fake_key', 'client_email': 'test@example.com'}
+      mock_cred.from_service_account_info.return_value = self.mock_credentials_instance
+
+      # Create proxy config
+      https_proxy = 'http://myproxy.com:443'
+      proxy_config = googleads.common.ProxyConfig(https_proxy=https_proxy)
+
+      # Create a GoogleServiceAccountClient using from_service_account_info with proxy config
+      client = googleads.oauth2.GoogleServiceAccountClient.from_service_account_info(
+          service_account_info, self.scope, sub=self.delegated_account, 
+          proxy_config=proxy_config)
+
+      # Verify the client has the expected proxy config
+      self.assertEqual(client.proxy_config.proxies, {'https': https_proxy})
+
   def testCreateHttpHeader_noRefresh(self):
     header = {'authorization': 'Bearer %s' % self.access_token_unrefreshed}
     self.mock_credentials_instance.expiry = (