22709357475

Committed 05 Mar 2026 08:35AM UTC coverage: 59.732% (-27.2%) from 86.974%

Build # 22709357475

Build Type

Pull #13880

github

Committed by

web-flow

Commit Message

Merge 28fcab93c into 710618057

Pull Request Pull Request #13880: Firehose: Replace TaggingService

Coverage Stats

12 of 12 new or added lines in 2 files covered. (100.0%)

20464 existing lines in 510 files now uncovered.

45290 of 75822 relevant lines covered (59.73%)

0.6 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

76.32

/localstack-core/localstack/utils/aws/client.py

import requests
from botocore.auth import BaseSigner
from botocore.awsrequest import AWSRequest


class SigningHttpClient:
    """
    A wrapper around ``requests`` that uses botocore to sign HTTP requests using a ``botocore.auth.BaseSigner``.

    For example, using a Sig4QueryAuth signer, invocations to ``client.get(
    "http://localhost:4566/000000000000/test-queue")`` it will transparently change the URL to something like:

    http://localhost:4566/000000000000/test-queue
        ?X-Amz-Algorithm=AWS4-HMAC-SHA256
        &X-Amz-Credential=__test_call__%2F20220513%2Fus-east-1%2Fsqs%2Faws4_request
        &X-Amz-Date=20220513T192006Z
        &X-Amz-Expires=3600
        &X-Amz-SignedHeaders=host
        &X-Amz-Signature=ae39eb839d0501d731d5dccffd1e6e86fab53749f956caabbb8211b6593f5f9d

    You can also create a client with an endpoint_url set, where you can then make requests with the hostname part.
    For example, to create a raw GetQueueUrl request, run:

       client = SigningHttpClient(signer, endpoint_url="https://sqs.us-east-1.amazonaws.com")
       client.post("/", params={"Action", "GetQueueUrl", "QueueName": "my-queue"})
    """

    def __init__(
        self, signer: BaseSigner, session: requests.Session = None, endpoint_url: str = None
    ):
        self.signer = signer
        self.session = session or requests.Session()
        self.endpoint_url = endpoint_url

    def request(self, method, url, **kwargs) -> requests.Response:
        if url.startswith("/"):
            if not self.endpoint_url:
                raise ValueError("no hostname provided in url and no endpoint_url set")

            url = self.endpoint_url.rstrip("/") + url

        request = AWSRequest(
            method=method,
            url=url,
            data=kwargs.pop("data", None),
            params=kwargs.pop("params", None),
            headers=kwargs.pop("headers", None),
        )
        request = self.sign(request)

        url = request.url
        method = request.method

        kwargs["data"] = request.data
        kwargs["params"] = request.params
        kwargs["headers"] = request.headers

        return self.session.request(method, url, **kwargs)

    def get(self, url, **kwargs):
        return self.request("GET", url, **kwargs)

    def head(self, url, **kwargs):
        return self.request("HEAD", url, **kwargs)

    def post(self, url, **kwargs):
        return self.request("POST", url, **kwargs)

    def put(self, url, **kwargs):
        return self.request("PUT", url, **kwargs)

    def patch(self, url, **kwargs):
        return self.request("PATCH", url, **kwargs)

    def delete(self, url, **kwargs):
        return self.request("DELETE", url, **kwargs)

    def options(self, url, **kwargs):
        return self.request("OPTIONS", url, **kwargs)

    def sign(self, request: AWSRequest) -> AWSRequest:
        self.signer.add_auth(request)
        return request

1	import requests	1✔
2	from botocore.auth import BaseSigner	1✔
3	from botocore.awsrequest import AWSRequest	1✔
4
5
6	class SigningHttpClient:	1✔
7	"""
8	A wrapper around ``requests`` that uses botocore to sign HTTP requests using a ``botocore.auth.BaseSigner``.
9
10	For example, using a Sig4QueryAuth signer, invocations to ``client.get(
11	"http://localhost:4566/000000000000/test-queue")`` it will transparently change the URL to something like:
12
13	http://localhost:4566/000000000000/test-queue
14	?X-Amz-Algorithm=AWS4-HMAC-SHA256
15	&X-Amz-Credential=__test_call__%2F20220513%2Fus-east-1%2Fsqs%2Faws4_request
16	&X-Amz-Date=20220513T192006Z
17	&X-Amz-Expires=3600
18	&X-Amz-SignedHeaders=host
19	&X-Amz-Signature=ae39eb839d0501d731d5dccffd1e6e86fab53749f956caabbb8211b6593f5f9d
20
21	You can also create a client with an endpoint_url set, where you can then make requests with the hostname part.
22	For example, to create a raw GetQueueUrl request, run:
23
24	client = SigningHttpClient(signer, endpoint_url="https://sqs.us-east-1.amazonaws.com")
25	client.post("/", params={"Action", "GetQueueUrl", "QueueName": "my-queue"})
26	"""
27
28	def __init__(	1✔
29	self, signer: BaseSigner, session: requests.Session = None, endpoint_url: str = None
30	):
31	self.signer = signer	1✔
32	self.session = session or requests.Session()	1✔
33	self.endpoint_url = endpoint_url	1✔
34
35	def request(self, method, url, **kwargs) -> requests.Response:	1✔
36	if url.startswith("/"):	1✔
UNCOV 37	if not self.endpoint_url:	×
38	raise ValueError("no hostname provided in url and no endpoint_url set")	×
39
UNCOV 40	url = self.endpoint_url.rstrip("/") + url	×
41
42	request = AWSRequest(	1✔
43	method=method,
44	url=url,
45	data=kwargs.pop("data", None),
46	params=kwargs.pop("params", None),
47	headers=kwargs.pop("headers", None),
48	)
49	request = self.sign(request)	1✔
50
51	url = request.url	1✔
52	method = request.method	1✔
53
54	kwargs["data"] = request.data	1✔
55	kwargs["params"] = request.params	1✔
56	kwargs["headers"] = request.headers	1✔
57
58	return self.session.request(method, url, **kwargs)	1✔
59
60	def get(self, url, **kwargs):	1✔
UNCOV 61	return self.request("GET", url, **kwargs)	×
62
63	def head(self, url, **kwargs):	1✔
UNCOV 64	return self.request("HEAD", url, **kwargs)	×
65
66	def post(self, url, **kwargs):	1✔
67	return self.request("POST", url, **kwargs)	1✔
68
69	def put(self, url, **kwargs):	1✔
UNCOV 70	return self.request("PUT", url, **kwargs)	×
71
72	def patch(self, url, **kwargs):	1✔
73	return self.request("PATCH", url, **kwargs)	×
74
75	def delete(self, url, **kwargs):	1✔
UNCOV 76	return self.request("DELETE", url, **kwargs)	×
77
78	def options(self, url, **kwargs):	1✔
79	return self.request("OPTIONS", url, **kwargs)	×
80
81	def sign(self, request: AWSRequest) -> AWSRequest:	1✔
82	self.signer.add_auth(request)	1✔
83	return request	1✔

localstack / localstack / 22709357475

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous