def call_function()

in processors/vertexgenai.py [0:0]

• 55 lines of code
• 17 McCabe index (conditional complexity)

    def call_function(self, name, params):
        method = 'GET'
        if 'method' in params:
            method = params['method'].upper()
        headers = {}
        if 'headers' in params:
            for header in params['headers']:
                headers[header['name'].lower()] = header['value']
        body = ''
        if 'body' in params:
            if isinstance(params['body'], dict):
                body = json.dumps(params['body'])
            else:
                body = params['body']
        loggable_headers = headers
        if 'authorization' in loggable_headers:
            del loggable_headers['authorization']
        if 'x-serverless-authorization' in loggable_headers:
            del loggable_headers['x-serverless-authorization']
        if 'api-key' in loggable_headers:
            del loggable_headers['api-key']
        if 'x-api-key' in loggable_headers:
            del loggable_headers['x-api-key']
        if 'proxy-authorization' in loggable_headers:
            del loggable_headers['proxy-authorization']

        self.logger.info('Calling function: %s' % (name),
                         extra={
                             'url': params['url'],
                             'method': method,
                             'body_length': len(body),
                             'headers': loggable_headers,
                             'id_token': True if 'idToken' in params else False
                         })

        id_token = None
        audience = None
        if 'idToken' in params and params['idToken']:
            if 'audience' in params:
                audience = params['audience']
            else:
                audience = params['url']
            auth_request = google.auth.transport.requests.Request()
            id_token = google.oauth2.id_token.fetch_id_token(
                auth_request, audience)
            headers['authorization'] = 'Bearer %s' % (id_token)

        req = urllib.request.Request(params['url'],
                                     headers=headers,
                                     method=method)
        response = urllib.request.urlopen(req, data=body.encode('utf-8'))
        if response.status < 200 or response.status >= 400:
            self.logger.error('Error calling function: %s' % (name),
                              extra={
                                  'status_code': response.status,
                                  'response': response.data.decode('utf-8')
                              })

        return (response.headers, json.loads(response.read().decode('utf-8')))