def process()

in processors/secret.py [0:0]

• 30 lines of code
• 9 McCabe index (conditional complexity)

    def process(self, output_var='secret'):
        if 'secret' not in self.config:
            raise NotConfiguredException(
                'No Secret Manager secret config specified.')
        secret_name = self.config['secret']
        if 'version' in self.config:
            secret_version = self.config['version']

        credentials, credentials_project_id = google.auth.default()
        project = self.config[
            'project'] if 'project' in self.config else credentials_project_id
        if not project:
            project = credentials.quota_project_id

        secret_full_name = "projects/%s/secrets/%s/versions/%s" % (
            project, secret_name, secret_version)

        client = secretmanager.SecretManagerServiceClient(
            client_info=get_grpc_client_info())
        response = client.access_secret_version(name=secret_full_name)
        secret_contents = response.payload.data.decode('UTF-8')
        if 'mode' in self.config:
            if 'base64' in self.config['mode']:
                secret_contents = base64.b64decode(secret_contents).decode(
                    'UTF-8')
            if 'json' in self.config['mode']:
                secret_contents = json.loads(secret_contents)
            if 'yaml' in self.config['mode']:
                secret_contents = yaml.load(secret_contents,
                                            Loader=yaml.SafeLoader)
        return {
            output_var: secret_contents,
        }