Python oauth2client.client 模块，SignedJwtAssertionCredentials() 实例源码

def authorize_service_account(json_credentials, scope, sub=None):
    """
    authorize to the provide scope with a service credentials json file.

    :param json_credentials: dictonary representing a service account, in most cases created from json.load
    :param scope: scope(s) to authorize the application for
    :param sub: User ID to authorize the application for (if needed)
    :return Credentials to be used for http object
    """
    try:
        from oauth2client.service_account import ServiceAccountCredentials
    except ImportError:
        ServiceAccountCredentials = None

    if ServiceAccountCredentials is not None and hasattr(ServiceAccountCredentials, "from_json_keyfile_dict"):
        # running oauth2client version 2.0
        creds = ServiceAccountCredentials.from_json_keyfile_dict(json_credentials, scope)
        if sub is not None:
            creds = creds.create_delegated(sub)
    else:
        try:
            from oauth2client.client import SignedJwtAssertionCredentials
        except ImportError:
            raise EnvironmentError("Service account can not be used because PyCrypto is not available. Please install PyCrypto.")

        if not isinstance(scope, (list, tuple)):
            scope = [scope]

        creds = SignedJwtAssertionCredentials(
            service_account_name=json_credentials['client_email'],
            private_key=json_credentials['private_key'],
            scope=scope,
            sub=sub)

    return creds

def get_service(api_name, api_version, scope):
  f = open(config.secret_file_location, 'rb')
  key = f.read()
  f.close()
  credentials = SignedJwtAssertionCredentials(config.service_account_email, key, scope=scope)
  http = credentials.authorize(httplib2.Http())
  service = build(api_name, api_version, http=http)
  return service

#RUN ANALYTICS QUERY

def create_sheets_client(scope):
  f = open(config.secret_file_location, 'rb')
  key = f.read()
  f.close()
  credentials = SignedJwtAssertionCredentials(config.service_account_email, key, scope)
  sheets_client = gspread.authorize(credentials)
  return sheets_client

def get_service(api_name, api_version, scope):
  f = open(config.secret_file_location, 'rb')
  key = f.read()
  f.close()
  credentials = SignedJwtAssertionCredentials(config.service_account_email, key, scope=scope)
  http = credentials.authorize(httplib2.Http())
  service = build(api_name, api_version, http=http)
  return service

def build_arrays():
    json_key = json.load(open('chopped bot-610249be55b6.json'))
    scope = ['https://spreadsheets.google.com/feeds']
    credentials = SignedJwtAssertionCredentials(json_key['client_email'], json_key['private_key'].encode(), scope)
    gc = gspread.authorize(credentials)
    sh = gc.open("choppedbot")
    weird_sheet = sh.worksheet("weird")
    normal_sheet = sh.worksheet("normal")
    weird_array = weird_sheet.col_values(1)
    normal_array = normal_sheet.col_values(1)

    return normal_array, weird_array

def list_instances(client_cert_file, client_email, project, zone):
    with open(client_cert_file) as f:
        private_key = f.read()
    credentials = SignedJwtAssertionCredentials(client_email, private_key,
        'https://www.googleapis.com/auth/compute.readonly')
    http = Http()
    credentials.authorize(http)
    try:
        compute = build('compute', 'v1', http=http)
        resp_json = compute.instances().list(project=project, zone=zone).execute()
    except HttpError:
        raise GoogleClientError('Failed to make "list instances" Google cloud API request')
    return resp_json

def __init__(self, cfg_name='google_drive'):
        import gspread
        from oauth2client.client import SignedJwtAssertionCredentials
        credentials_json = json.loads(luigi.configuration.get_config().get(cfg_name, 'credentials_json'))
        client_email = credentials_json["client_email"]
        private_key = credentials_json["private_key"]
        scope = ['https://spreadsheets.google.com/feeds']
        credentials = SignedJwtAssertionCredentials(client_email, private_key.encode(), scope)
        self.client = gspread.authorize(credentials)

def run(self):
        self.filename = self.table + '.csv'
        credentials = SignedJwtAssertionCredentials(JSON_KEY['client_email'],
                                                    JSON_KEY['private_key'].encode(),
                                                    SCOPE)
        gc = gspread.authorize(credentials)
        sheet = gc.open(self.gsheet)
        locsheet = sheet.worksheet(self.sheet_name)
        data = locsheet.get_all_values()
        # list of lists, first value of each list is column header
        header = locsheet.get_all_values()[0]
        data = [l for l in data if l != header]
        data = pd.DataFrame(data, columns=header).to_csv("data/" + self.filename,
                                                         index=False,
                                                         header=True)
        global records_local
        records_local = len(data)
        self.output().done()

def get_sheets_client(config):
    scope = ['https://spreadsheets.google.com/feeds']
    credentials = SignedJwtAssertionCredentials(config['client_email'], config['private_key'].encode(), scope)
    return gspread.authorize(credentials)

def get_service_account_credentials(self):
        # Bug fix for https://github.com/pydata/pandas/issues/12572
        # We need to know that a supported version of oauth2client is installed
        # Test that either of the following is installed:
        # - SignedJwtAssertionCredentials from oauth2client.client
        # - ServiceAccountCredentials from oauth2client.service_account
        # SignedJwtAssertionCredentials is available in oauthclient < 2.0.0
        # ServiceAccountCredentials is available in oauthclient >= 2.0.0
        oauth2client_v1 = True
        oauth2client_v2 = True

        try:
            from oauth2client.client import SignedJwtAssertionCredentials
        except ImportError:
            oauth2client_v1 = False

        try:
            from oauth2client.service_account import ServiceAccountCredentials
        except ImportError:
            oauth2client_v2 = False

        if not oauth2client_v1 and not oauth2client_v2:
            raise ImportError("Missing oauth2client required for BigQuery "
                              "service account support")

        from os.path import isfile

        try:
            if isfile(self.private_key):
                with open(self.private_key) as f:
                    json_key = json.loads(f.read())
            else:
                # ugly hack: 'private_key' field has new lines inside,
                # they break json parser, but we need to preserve them
                json_key = json.loads(self.private_key.replace('\n', '   '))
                json_key['private_key'] = json_key['private_key'].replace(
                    '   ', '\n')

            if compat.PY3:
                json_key['private_key'] = bytes(
                    json_key['private_key'], 'UTF-8')

            if oauth2client_v1:
                return SignedJwtAssertionCredentials(
                    json_key['client_email'],
                    json_key['private_key'],
                    self.scope,
                )
            else:
                return ServiceAccountCredentials.from_json_keyfile_dict(
                    json_key,
                    self.scope)
        except (KeyError, ValueError, TypeError, AttributeError):
            raise InvalidPrivateKeyFormat(
                "Private key is missing or invalid. It should be service "
                "account private key JSON (file path or string contents) "
                "with at least two keys: 'client_email' and 'private_key'. "
                "Can be obtained from: https://console.developers.google."
                "com/permissions/serviceaccounts")