Send request with SigV4 in python using boto3

sigv4_using_http_client.py

from boto3.session import Session
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest
from botocore.credentials import Credentials
from http.client import HTTPConnection, HTTPSConnection
import json
import os
from urllib.parse import urlparse

def sigv4_request(
  url,
  method='GET',
  body=None,
  params=None,
  headers=None,
  service='execute-api',
  region=os.environ['AWS_REGION'],
  credentials=Session().get_credentials().get_frozen_credentials()
):
  """Sends an HTTP request signed with SigV4

  Args:
    url: The request URL (e.g. 'https://www.example.com').
    method: The request method (e.g. 'GET', 'POST', 'PUT', 'DELETE'). Defaults to 'GET'.
    body: The request body (e.g. json.dumps({ 'foo': 'bar' })). Defaults to None.
    params: The request query params (e.g. { 'foo': 'bar' }). Defaults to None.
    headers: The request headers (e.g. { 'content-type': 'application/json' }). Defaults to None.
    service: The AWS service name. Defaults to 'execute-api'.
    region: The AWS region id. Defaults to the env var 'AWS_REGION'.
    credentials: The AWS credentials. Defaults to the current boto3 session's credentials.
  Returns:
     The HTTP response
  """

  # sign request
  req = AWSRequest(
    method=method,
    url=url,
    data=body,
    params=params,
    headers=headers
  )
  SigV4Auth(credentials, service, region).add_auth(req)
  req = req.prepare()
    
  # parse URL
  u = urlparse(req.url)
  path_and_query = u.path if u.query is None else u.path + '?' + u.query

  # send request
  cnn = HTTPSConnection(u.hostname) if u.scheme == 'https' else HTTPConnection(u.hostname)
  cnn.request(
    req.method,
    path_and_query,
    headers=req.headers,
    body=req.body
  )
  return cnn.getresponse()

def __init__():
  # send request
  response = sigv4_request(
    'https://www.example.com',
    method='POST',
    headers={
      'content-type': 'application/json',
    },
    body=json.dumps({
      'foo': 'bar'
    }),
    region='us-west-2',
    credentials=Credentials(
      'ACCESS_KEY_ID',
      'SECRET_ACCESS_KEY',
      'SESSION_TOKEN'
    )
  )
  
  # do something with response
  print(response.status)
  print(json.loads(response.read().decode()))

sigv4_using_requests.py

from boto3.session import Session
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest
from botocore.credentials import Credentials
import json
import os
from requests import request

def sigv4_request(
  url,
  method='GET',
  body=None,
  params=None,
  headers=None,
  service='execute-api',
  region=os.environ['AWS_REGION'],
  credentials=Session().get_credentials().get_frozen_credentials()
):
  """Sends an HTTP request signed with SigV4

  Args:
    url: The request URL (e.g. 'https://www.example.com').
    method: The request method (e.g. 'GET', 'POST', 'PUT', 'DELETE'). Defaults to 'GET'.
    body: The request body (e.g. json.dumps({ 'foo': 'bar' })). Defaults to None.
    params: The request query params (e.g. { 'foo': 'bar' }). Defaults to None.
    headers: The request headers (e.g. { 'content-type': 'application/json' }). Defaults to None.
    service: The AWS service name. Defaults to 'execute-api'.
    region: The AWS region id. Defaults to the env var 'AWS_REGION'.
    credentials: The AWS credentials. Defaults to the current boto3 session's credentials.
  Returns:
     The HTTP response
  """

  # sign request
  req = AWSRequest(
    method=method,
    url=url,
    data=body,
    params=params,
    headers=headers
  )
  SigV4Auth(credentials, service, region).add_auth(req)
  req = req.prepare()

  # send request
  return request(
    method=req.method,
    url=req.url,
    headers=req.headers,
    data=req.body
  )

def __init__():
  # send request
  response = sigv4_request(
    'https://www.example.com',
    method='POST',
    headers={
      'content-type': 'application/json',
    },
    body=json.dumps({
      'foo': 'bar'
    }),
    # In Lambda functions, you can omit 'region' and 'credentials' to use the Lambda's region and credentials
  )
  
  # do something with response
  print(response.status_code)
  print(response.json())

i will test it.