Join the 1-day Testing & QA Summit featuring 15+ Expert Speakers.Register for FREE! Join TestMu Conference

How to use get_proxy_request_for_thread method in localstack

Best Python code snippet using localstack_python

request_context.py

Source:request_context.py

...9from localstack.utils.run import FuncThread10LOG = logging.getLogger(__name__)11THREAD_LOCAL = threading.local()12MARKER_APIGW_REQUEST_REGION = "__apigw_request_region__"13def get_proxy_request_for_thread():14    try:15        return THREAD_LOCAL.request_context16    except Exception:17        return None18def get_flask_request_for_thread():19    try:20        return Request(21            url=request.path,22            data=request.data,23            headers=CaseInsensitiveDict(request.headers),24            method=request.method,25        )26    except Exception as e:27        # swallow error: "Working outside of request context."28        if "Working outside" in str(e):29            return None30        raise31def extract_region_from_auth_header(headers):32    # TODO: use method from aws_stack directly (leaving import here for now, to avoid circular dependency)33    from localstack.utils.aws import aws_stack34    auth = headers.get("Authorization") or ""35    region = re.sub(r".*Credential=[^/]+/[^/]+/([^/]+)/.*", r"\1", auth)36    if region == auth:37        return None38    region = region or aws_stack.get_local_region()39    return region40def get_request_context():41    candidates = [get_proxy_request_for_thread(), get_flask_request_for_thread()]42    for req in candidates:43        if req is not None:44            return req45class RequestContextManager(object):46    """Context manager which sets the given request context (i.e., region) for the scope of the block."""47    def __init__(self, request_context):48        self.request_context = request_context49    def __enter__(self):50        THREAD_LOCAL.request_context = self.request_context51    def __exit__(self, type, value, traceback):52        THREAD_LOCAL.request_context = None53def get_region_from_request_context():54    """look up region from request context"""55    if config.USE_SINGLE_REGION:56        return57    request_context = get_request_context()58    if not request_context:59        return60    region = extract_region_from_auth_header(request_context.headers)61    # Fix region lookup for certain requests, e.g., API gateway invocations62    #  that do not contain region details in the Authorization header.63    region = request_context.headers.get(MARKER_APIGW_REQUEST_REGION) or region64    return region65def configure_region_for_current_request(region_name: str, service_name: str):66    """Manually configure (potentially overwrite) the region in the current request context. This may be67    used by API endpoints that are invoked directly by the user (without specifying AWS Authorization68    headers), to still enable transparent region lookup via aws_stack.get_region() ..."""69    # TODO: leaving import here for now, to avoid circular dependency70    from localstack.utils.aws import aws_stack71    request_context = get_request_context()72    if not request_context:73        LOG.info(74            "Unable to set region '%s' in undefined request context: %s",75            region_name,76            request_context,77        )78        return79    headers = request_context.headers80    auth_header = headers.get("Authorization")81    auth_header = auth_header or aws_stack.mock_aws_request_headers(service_name)82    auth_header = auth_header.replace("/%s/" % aws_stack.get_region(), "/%s/" % region_name)83    try:84        headers["Authorization"] = auth_header85    except Exception as e:86        if "immutable" not in str(e):87            raise88        _context_to_update = get_proxy_request_for_thread() or request89        _context_to_update.headers = CaseInsensitiveDict({**headers, "Authorization": auth_header})90def patch_request_handling():91    if config.USE_SINGLE_REGION:92        return93    # TODO: move into generic_proxy.py, instead of patching here (leaving import here for now, to avoid circular dependency)94    from localstack.services import generic_proxy95    def modify_and_forward(method=None, path=None, data_bytes=None, headers=None, *args, **kwargs):96        """Patch proxy forward method and store request in thread local."""97        request_context = get_proxy_request_for_thread()98        context_manager = empty_context_manager()99        if not request_context:100            request_context = Request(url=path, data=data_bytes, headers=headers, method=method)101            context_manager = RequestContextManager(request_context)102        with context_manager:103            result = modify_and_forward_orig(104                method, path, data_bytes=data_bytes, headers=headers, *args, **kwargs105            )106        return result107    modify_and_forward_orig = generic_proxy.modify_and_forward108    generic_proxy.modify_and_forward = modify_and_forward109    # make sure that we inherit THREAD_LOCAL request contexts to spawned sub-threads110    def thread_init(self, *args, **kwargs):111        self._req_context = get_request_context()...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.