Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use modify_context_region method in localstack

Best Python code snippet using localstack_python

provider.py

Source:provider.py

...262        return None263    original_region = list(replicas.keys())[0]264    return original_region if target_region in replicas[original_region] else None265@contextmanager266def modify_context_region(context: RequestContext, region: str):267    """268    Context manager that modifies the region of a `RequestContext`. At the exit, the context is restored to its269    original state.270    :param context: the context to modify271    :param region: the modified region272    :return: a modified `RequestContext`273    """274    original_region = context.region275    original_authorization = context.request.headers.get("Authorization")276    context.region = region277    context.request.headers["Authorization"] = re.sub(278        r"Credential=([^/]+/[^/]+)/(.*?)/",279        rf"Credential=\1/{region}/",280        original_authorization or "",281        flags=re.IGNORECASE,282    )283    yield context284    # revert the original context285    context.region = original_region286    context.request.headers["Authorization"] = original_authorization287class DynamoDBProvider(DynamodbApi, ServiceLifecycleHook):288    def __init__(self):289        self.request_forwarder = get_request_forwarder_http(self.get_forward_url)290    def on_after_init(self):291        # add response processor specific to ddblocal292        handlers.modify_service_response.append(self.service, self._modify_ddblocal_arns)293        # routes for the shell ui294        ROUTER.add(295            path="/shell",296            endpoint=self.handle_shell_ui_redirect,297            methods=["GET"],298        )299        ROUTER.add(300            path="/shell/<regex('.*'):req_path>",301            endpoint=self.handle_shell_ui_request,302        )303    def _modify_ddblocal_arns(self, chain, context: RequestContext, response: Response):304        """A service response handler that modifies the dynamodb backend response."""305        if response_content := response.get_data(as_text=True):306            # fix the table and latest stream ARNs (DynamoDBLocal hardcodes "ddblocal" as the region)307            content_replaced = re.sub(308                r'("TableArn"|"LatestStreamArn"|"StreamArn")\s*:\s*"arn:([a-z-]+):dynamodb:ddblocal:([^"]+)"',309                rf'\1: "arn:\2:dynamodb:{aws_stack.get_region()}:\3"',310                response_content,311            )312            if content_replaced != response_content:313                response.data = content_replaced314                context.service_response = (315                    None  # make sure the service response is parsed again later316                )317        # update x-amz-crc32 header required by some clients318        response.headers["x-amz-crc32"] = crc32(response.data) & 0xFFFFFFFF319    def forward_request(320        self, context: RequestContext, service_request: ServiceRequest = None321    ) -> ServiceResponse:322        # check rate limiting for this request and raise an error, if provisioned throughput is exceeded323        self.check_provisioned_throughput(context.operation.name)324        # note: modifying headers in-place here before forwarding the request325        self.prepare_request_headers(context.request.headers)326        return self.request_forwarder(context, service_request)327    def get_forward_url(self) -> str:328        """Return the URL of the backend DynamoDBLocal server to forward requests to"""329        return f"http://{LOCALHOST}:{server.get_server().port}"330    def on_before_start(self):331        start_dynamodb()332        wait_for_dynamodb()333    def handle_shell_ui_redirect(self, request: werkzeug.Request) -> Response:334        headers = {"Refresh": f"0; url={config.service_url('dynamodb')}/shell/index.html"}335        return Response("", headers=headers)336    def handle_shell_ui_request(self, request: werkzeug.Request, req_path: str) -> Response:337        # TODO: "DynamoDB Local Web Shell was deprecated with version 1.16.X and is not available any338        #  longer from 1.17.X to latest. There are no immediate plans for a new Web Shell to be introduced."339        #  -> keeping this for now, to allow configuring custom installs; should consider removing it in the future340        # https://repost.aws/questions/QUHyIzoEDqQ3iOKlUEp1LPWQ#ANdBm9Nz9TRf6VqR3jZtcA1g341        req_path = f"/{req_path}" if not req_path.startswith("/") else req_path342        url = f"{self.get_forward_url()}/shell{req_path}"343        result = requests.request(344            method=request.method, url=url, headers=request.headers, data=request.data345        )346        return Response(result.content, headers=dict(result.headers), status=result.status_code)347    @handler("CreateTable", expand=False)348    def create_table(349        self,350        context: RequestContext,351        create_table_input: CreateTableInput,352    ) -> CreateTableOutput:353        # Check if table exists, to avoid error log output from DynamoDBLocal354        table_name = create_table_input["TableName"]355        if self.table_exists(table_name):356            raise ResourceInUseException("Cannot create preexisting table")357        billing_mode = create_table_input.get("BillingMode")358        provisioned_throughput = create_table_input.get("ProvisionedThroughput")359        if billing_mode == BillingMode.PAY_PER_REQUEST and provisioned_throughput is not None:360            raise ValidationException(361                "One or more parameter values were invalid: Neither ReadCapacityUnits nor WriteCapacityUnits can be "362                "specified when BillingMode is PAY_PER_REQUEST"363            )364        # forward request to backend365        result = self.forward_request(context)366        table_description = result["TableDescription"]367        backend = get_store(context)368        backend.table_definitions[table_name] = table_definitions = dict(create_table_input)369        if "TableId" not in table_definitions:370            table_definitions["TableId"] = long_uid()371        if "SSESpecification" in table_definitions:372            sse_specification = table_definitions.pop("SSESpecification")373            table_definitions["SSEDescription"] = SSEUtils.get_sse_description(sse_specification)374        if table_definitions:375            table_content = result.get("Table", {})376            table_content.update(table_definitions)377            table_description.update(table_content)378        if "StreamSpecification" in table_definitions:379            create_dynamodb_stream(table_definitions, table_description.get("LatestStreamLabel"))380        if "TableClass" in table_definitions:381            table_class = table_description.pop("TableClass", None) or table_definitions.pop(382                "TableClass"383            )384            table_description["TableClassSummary"] = {"TableClass": table_class}385        tags = table_definitions.pop("Tags", [])386        if tags:387            table_arn = table_description["TableArn"]388            table_arn = self.fix_table_arn(table_arn)389            get_store(context).TABLE_TAGS[table_arn] = {tag["Key"]: tag["Value"] for tag in tags}390        # remove invalid attributes from result391        table_description.pop("Tags", None)392        table_description.pop("BillingMode", None)393        return result394    def delete_table(self, context: RequestContext, table_name: TableName) -> DeleteTableOutput:395        # Check if table exists, to avoid error log output from DynamoDBLocal396        if not self.table_exists(table_name):397            raise ResourceNotFoundException("Cannot do operations on a non-existent table")398        # forward request to backend399        result = self.forward_request(context)400        table_arn = result.get("TableDescription", {}).get("TableArn")401        table_arn = self.fix_table_arn(table_arn)402        self.delete_all_event_source_mappings(table_arn)403        dynamodbstreams_api.delete_streams(table_arn)404        get_store(context).TABLE_TAGS.pop(table_arn, None)405        return result406    def _forward_request(self, context: RequestContext, region: str | None) -> ServiceResponse:407        if region:408            with modify_context_region(context, region):409                return self.forward_request(context)410        return self.forward_request(context)411    def describe_table(self, context: RequestContext, table_name: TableName) -> DescribeTableOutput:412        global_table_region: str | None = find_global_table_region(413            table_name=table_name, target_region=context.region414        )415        # Check if table exists, to avoid error log output from DynamoDBLocal416        if not self.table_exists(table_name) and not global_table_region:417            raise ResourceNotFoundException("Cannot do operations on a non-existent table")418        result = self._forward_request(context=context, region=global_table_region)419        # update response with additional props420        table_props = get_store(context).table_properties.get(table_name)421        if table_props:422            result.get("Table", {}).update(table_props)...

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.