API request using IAM authentication

iam_api_request.py

"""Use IAM credentials as authentication"""

import boto3
from botocore.awsrequest import AWSRequest
from botocore.auth import SigV4Auth
from requests.auth import AuthBase
from requests import PreparedRequest


class ApiGateway(AuthBase):  # pylint: disable=too-few-public-methods
    """API Gateway IAM authorizer

    Usage:
    ======
    response = requests.post(
        url=url,
        headers=headers,
        json=data,
        auth=ApiGateway(botocore_session=session)  # Use IAM authentication
    )
    """

    def __init__(self, boto3_session: boto3.Session | None = None) -> None:

        boto3_session = boto3_session or boto3.Session()
        self.credentials = boto3_session.get_credentials()
        self.region_name = boto3_session.region_name

    def __call__(self, r: PreparedRequest) -> PreparedRequest:

        def get_signed_request():

            signed_request = AWSRequest(
                method=r.method.upper(),
                headers=r.headers,
                url=r.url,
                data=r.body,
            )
            SigV4Auth(
                credentials=self.credentials,
                service_name='execute-api',
                region_name=self.region_name,
            ).add_auth(
                request=signed_request,
            )

            return signed_request

        if 'connection' in r.headers:  # pylint: disable=magic-value-comparison
            connection = r.headers['connection']
            del r.headers['connection']
            signed_request = get_signed_request()
            r.headers['connection'] = connection
        else:
            signed_request = get_signed_request()

        r.headers.update(signed_request.headers)

        return r

Then you can do:

import requests

boto3_session = boto3.Session()
requests_session = requests.Session(auth=ApiGateway(boto3_session=boto3_session))
response = requests_session.get('http://www.domain.tld/endpoint')