20565403496

Committed 29 Dec 2025 05:11AM UTC coverage: 84.103% (-2.8%) from 86.921%

Build # 20565403496

Build Type

Pull #13567

github

Committed by

web-flow

Commit Message

Merge 4816837a5 into 2417384aa

Pull Request Pull Request #13567: Update ASF APIs

Run Details

67166 of 79862 relevant lines covered (84.1%)

0.84 hits per line

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

45.21

/localstack-core/localstack/services/apigateway/next_gen/execute_api/integrations/http.py

import logging
from http import HTTPMethod
from typing import TypedDict

import requests
from werkzeug.datastructures import Headers

from localstack.aws.api.apigateway import Integration

from ..context import EndpointResponse, IntegrationRequest, RestApiInvocationContext
from ..gateway_response import ApiConfigurationError, IntegrationFailureError, InternalServerError
from ..header_utils import build_multi_value_headers
from .core import RestApiIntegration

LOG = logging.getLogger(__name__)

NO_BODY_METHODS = {HTTPMethod.OPTIONS, HTTPMethod.GET, HTTPMethod.HEAD}


class SimpleHttpRequest(TypedDict, total=False):
    method: HTTPMethod | str
    url: str
    params: dict[str, str | list[str]] | None
    data: bytes
    headers: dict[str, str] | None
    cookies: dict[str, str] | None
    timeout: int | None
    allow_redirects: bool | None
    stream: bool | None
    verify: bool | None
    # TODO: check if there was a situation where we'd pass certs?
    cert: str | tuple[str, str] | None


class BaseRestApiHttpIntegration(RestApiIntegration):
    @staticmethod
    def _get_integration_timeout(integration: Integration) -> float:
        return int(integration.get("timeoutInMillis", 29000)) / 1000


class RestApiHttpIntegration(BaseRestApiHttpIntegration):
    """
    This is a REST API integration responsible to send a request to another HTTP API.
    https://docs.aws.amazon.com/apigateway/latest/developerguide/setup-http-integrations.html#api-gateway-set-up-http-proxy-integration-on-proxy-resource
    """

    name = "HTTP"

    def invoke(self, context: RestApiInvocationContext) -> EndpointResponse:
        integration_req: IntegrationRequest = context.integration_request
        method = integration_req["http_method"]
        uri = integration_req["uri"]

        request_parameters: SimpleHttpRequest = {
            "method": method,
            "url": uri,
            "params": integration_req["query_string_parameters"],
            "headers": integration_req["headers"],
        }

        if method not in NO_BODY_METHODS:
            request_parameters["data"] = integration_req["body"]

        # TODO: configurable timeout (29 by default) (check type and default value in provider)
        # integration: Integration = context.resource_method["methodIntegration"]
        # request_parameters["timeout"] = self._get_integration_timeout(integration)
        # TODO: check for redirects
        # request_parameters["allow_redirects"] = False
        try:
            request_response = requests.request(**request_parameters)

        except (requests.exceptions.InvalidURL, requests.exceptions.InvalidSchema) as e:
            LOG.warning("Execution failed due to configuration error: Invalid endpoint address")
            LOG.debug("The URI specified for the HTTP/HTTP_PROXY integration is invalid: %s", uri)
            raise InternalServerError("Internal server error") from e

        except (requests.exceptions.Timeout, requests.exceptions.SSLError) as e:
            # TODO make the exception catching more fine grained
            # this can be reproduced in AWS if you try to hit an HTTP endpoint which is HTTPS only like lambda URL
            LOG.warning("Execution failed due to a network error communicating with endpoint")
            raise IntegrationFailureError("Network error communicating with endpoint") from e

        except requests.exceptions.ConnectionError as e:
            raise ApiConfigurationError("Internal server error") from e

        return EndpointResponse(
            body=request_response.content,
            status_code=request_response.status_code,
            headers=Headers(dict(request_response.headers)),
        )


class RestApiHttpProxyIntegration(BaseRestApiHttpIntegration):
    """
    This is a simplified REST API integration responsible to send a request to another HTTP API by proxying it almost
    directly.
    https://docs.aws.amazon.com/apigateway/latest/developerguide/setup-http-integrations.html#api-gateway-set-up-http-proxy-integration-on-proxy-resource
    """

    name = "HTTP_PROXY"

    def invoke(self, context: RestApiInvocationContext) -> EndpointResponse:
        integration_req: IntegrationRequest = context.integration_request
        method = integration_req["http_method"]
        uri = integration_req["uri"]

        multi_value_headers = build_multi_value_headers(integration_req["headers"])
        request_headers = {key: ",".join(value) for key, value in multi_value_headers.items()}

        request_parameters: SimpleHttpRequest = {
            "method": method,
            "url": uri,
            "params": integration_req["query_string_parameters"],
            "headers": request_headers,
        }

        # TODO: validate this for HTTP_PROXY
        if method not in NO_BODY_METHODS:
            request_parameters["data"] = integration_req["body"]

        # TODO: configurable timeout (29 by default) (check type and default value in provider)
        # integration: Integration = context.resource_method["methodIntegration"]
        # request_parameters["timeout"] = self._get_integration_timeout(integration)
        try:
            request_response = requests.request(**request_parameters)

        except (requests.exceptions.InvalidURL, requests.exceptions.InvalidSchema) as e:
            LOG.warning("Execution failed due to configuration error: Invalid endpoint address")
            LOG.debug("The URI specified for the HTTP/HTTP_PROXY integration is invalid: %s", uri)
            raise InternalServerError("Internal server error") from e

        except (requests.exceptions.Timeout, requests.exceptions.SSLError):
            # TODO make the exception catching more fine grained
            # this can be reproduced in AWS if you try to hit an HTTP endpoint which is HTTPS only like lambda URL
            LOG.warning("Execution failed due to a network error communicating with endpoint")
            raise IntegrationFailureError("Network error communicating with endpoint")

        except requests.exceptions.ConnectionError:
            raise ApiConfigurationError("Internal server error")

        response_headers = Headers(dict(request_response.headers))

        return EndpointResponse(
            body=request_response.content,
            status_code=request_response.status_code,
            headers=response_headers,
        )

1	import logging	1✔
2	from http import HTTPMethod	1✔
3	from typing import TypedDict	1✔
4
5	import requests	1✔
6	from werkzeug.datastructures import Headers	1✔
7
8	from localstack.aws.api.apigateway import Integration	1✔
9
10	from ..context import EndpointResponse, IntegrationRequest, RestApiInvocationContext	1✔
11	from ..gateway_response import ApiConfigurationError, IntegrationFailureError, InternalServerError	1✔
12	from ..header_utils import build_multi_value_headers	1✔
13	from .core import RestApiIntegration	1✔
14
15	LOG = logging.getLogger(__name__)	1✔
16
17	NO_BODY_METHODS = {HTTPMethod.OPTIONS, HTTPMethod.GET, HTTPMethod.HEAD}	1✔
18
19
20	class SimpleHttpRequest(TypedDict, total=False):	1✔
21	method: HTTPMethod \| str	1✔
22	url: str	1✔
23	params: dict[str, str \| list[str]] \| None	1✔
24	data: bytes	1✔
25	headers: dict[str, str] \| None	1✔
26	cookies: dict[str, str] \| None	1✔
27	timeout: int \| None	1✔
28	allow_redirects: bool \| None	1✔
29	stream: bool \| None	1✔
30	verify: bool \| None	1✔
31	# TODO: check if there was a situation where we'd pass certs?
32	cert: str \| tuple[str, str] \| None	1✔
33
34
35	class BaseRestApiHttpIntegration(RestApiIntegration):	1✔
36	@staticmethod	1✔
37	def _get_integration_timeout(integration: Integration) -> float:	1✔
38	return int(integration.get("timeoutInMillis", 29000)) / 1000	×
39
40
41	class RestApiHttpIntegration(BaseRestApiHttpIntegration):	1✔
42	"""
43	This is a REST API integration responsible to send a request to another HTTP API.
44	https://docs.aws.amazon.com/apigateway/latest/developerguide/setup-http-integrations.html#api-gateway-set-up-http-proxy-integration-on-proxy-resource
45	"""
46
47	name = "HTTP"	1✔
48
49	def invoke(self, context: RestApiInvocationContext) -> EndpointResponse:	1✔
50	integration_req: IntegrationRequest = context.integration_request	×
51	method = integration_req["http_method"]	×
52	uri = integration_req["uri"]	×
53
54	request_parameters: SimpleHttpRequest = {	×
55	"method": method,
56	"url": uri,
57	"params": integration_req["query_string_parameters"],
58	"headers": integration_req["headers"],
59	}
60
61	if method not in NO_BODY_METHODS:	×
62	request_parameters["data"] = integration_req["body"]	×
63
64	# TODO: configurable timeout (29 by default) (check type and default value in provider)
65	# integration: Integration = context.resource_method["methodIntegration"]
66	# request_parameters["timeout"] = self._get_integration_timeout(integration)
67	# TODO: check for redirects
68	# request_parameters["allow_redirects"] = False
69	try:	×
70	request_response = requests.request(**request_parameters)	×
71
72	except (requests.exceptions.InvalidURL, requests.exceptions.InvalidSchema) as e:	×
73	LOG.warning("Execution failed due to configuration error: Invalid endpoint address")	×
74	LOG.debug("The URI specified for the HTTP/HTTP_PROXY integration is invalid: %s", uri)	×
75	raise InternalServerError("Internal server error") from e	×
76
77	except (requests.exceptions.Timeout, requests.exceptions.SSLError) as e:	×
78	# TODO make the exception catching more fine grained
79	# this can be reproduced in AWS if you try to hit an HTTP endpoint which is HTTPS only like lambda URL
80	LOG.warning("Execution failed due to a network error communicating with endpoint")	×
81	raise IntegrationFailureError("Network error communicating with endpoint") from e	×
82
83	except requests.exceptions.ConnectionError as e:	×
84	raise ApiConfigurationError("Internal server error") from e	×
85
86	return EndpointResponse(	×
87	body=request_response.content,
88	status_code=request_response.status_code,
89	headers=Headers(dict(request_response.headers)),
90	)
91
92
93	class RestApiHttpProxyIntegration(BaseRestApiHttpIntegration):	1✔
94	"""
95	This is a simplified REST API integration responsible to send a request to another HTTP API by proxying it almost
96	directly.
97	https://docs.aws.amazon.com/apigateway/latest/developerguide/setup-http-integrations.html#api-gateway-set-up-http-proxy-integration-on-proxy-resource
98	"""
99
100	name = "HTTP_PROXY"	1✔
101
102	def invoke(self, context: RestApiInvocationContext) -> EndpointResponse:	1✔
103	integration_req: IntegrationRequest = context.integration_request	×
104	method = integration_req["http_method"]	×
105	uri = integration_req["uri"]	×
106
107	multi_value_headers = build_multi_value_headers(integration_req["headers"])	×
108	request_headers = {key: ",".join(value) for key, value in multi_value_headers.items()}	×
109
110	request_parameters: SimpleHttpRequest = {	×
111	"method": method,
112	"url": uri,
113	"params": integration_req["query_string_parameters"],
114	"headers": request_headers,
115	}
116
117	# TODO: validate this for HTTP_PROXY
118	if method not in NO_BODY_METHODS:	×
119	request_parameters["data"] = integration_req["body"]	×
120
121	# TODO: configurable timeout (29 by default) (check type and default value in provider)
122	# integration: Integration = context.resource_method["methodIntegration"]
123	# request_parameters["timeout"] = self._get_integration_timeout(integration)
124	try:	×
125	request_response = requests.request(**request_parameters)	×
126
127	except (requests.exceptions.InvalidURL, requests.exceptions.InvalidSchema) as e:	×
128	LOG.warning("Execution failed due to configuration error: Invalid endpoint address")	×
129	LOG.debug("The URI specified for the HTTP/HTTP_PROXY integration is invalid: %s", uri)	×
130	raise InternalServerError("Internal server error") from e	×
131
132	except (requests.exceptions.Timeout, requests.exceptions.SSLError):	×
133	# TODO make the exception catching more fine grained
134	# this can be reproduced in AWS if you try to hit an HTTP endpoint which is HTTPS only like lambda URL
135	LOG.warning("Execution failed due to a network error communicating with endpoint")	×
136	raise IntegrationFailureError("Network error communicating with endpoint")	×
137
138	except requests.exceptions.ConnectionError:	×
139	raise ApiConfigurationError("Internal server error")	×
140
141	response_headers = Headers(dict(request_response.headers))	×
142
143	return EndpointResponse(	×
144	body=request_response.content,
145	status_code=request_response.status_code,
146	headers=response_headers,
147	)

localstack / localstack / 20565403496

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