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

How to use integration_uri method in localstack

Best Python code snippet using localstack_python

models.py

Source:models.py

1"""ApiGatewayV2Backend class with methods for supported APIs."""2import random3import string4import yaml5from moto.core import BaseBackend, BaseModel6from moto.core.utils import BackendDict, unix_time7from moto.utilities.tagging_service import TaggingService8from .exceptions import (9    ApiNotFound,10    AuthorizerNotFound,11    BadRequestException,12    ModelNotFound,13    RouteResponseNotFound,14    IntegrationNotFound,15    IntegrationResponseNotFound,16    RouteNotFound,17    VpcLinkNotFound,18)19class Authorizer(BaseModel):20    def __init__(21        self,22        auth_creds_arn,23        auth_payload_format_version,24        auth_result_ttl,25        authorizer_type,26        authorizer_uri,27        enable_simple_response,28        identity_source,29        identity_validation_expr,30        jwt_config,31        name,32    ):33        self.id = "".join(random.choice(string.ascii_lowercase) for _ in range(8))34        self.auth_creds_arn = auth_creds_arn35        self.auth_payload_format_version = auth_payload_format_version36        self.auth_result_ttl = auth_result_ttl37        self.authorizer_type = authorizer_type38        self.authorizer_uri = authorizer_uri39        self.enable_simple_response = enable_simple_response40        self.identity_source = identity_source41        self.identity_validation_expr = identity_validation_expr42        self.jwt_config = jwt_config43        self.name = name44    def update(45        self,46        auth_creds_arn,47        auth_payload_format_version,48        auth_result_ttl,49        authorizer_type,50        authorizer_uri,51        enable_simple_response,52        identity_source,53        identity_validation_expr,54        jwt_config,55        name,56    ):57        if auth_creds_arn is not None:58            self.auth_creds_arn = auth_creds_arn59        if auth_payload_format_version is not None:60            self.auth_payload_format_version = auth_payload_format_version61        if auth_result_ttl is not None:62            self.auth_result_ttl = auth_result_ttl63        if authorizer_type is not None:64            self.authorizer_type is authorizer_type65        if authorizer_uri is not None:66            self.authorizer_uri = authorizer_uri67        if enable_simple_response is not None:68            self.enable_simple_response = enable_simple_response69        if identity_source is not None:70            self.identity_source = identity_source71        if identity_validation_expr is not None:72            self.identity_validation_expr = identity_validation_expr73        if jwt_config is not None:74            self.jwt_config = jwt_config75        if name is not None:76            self.name = name77    def to_json(self):78        return {79            "authorizerId": self.id,80            "authorizerCredentialsArn": self.auth_creds_arn,81            "authorizerPayloadFormatVersion": self.auth_payload_format_version,82            "authorizerResultTtlInSeconds": self.auth_result_ttl,83            "authorizerType": self.authorizer_type,84            "authorizerUri": self.authorizer_uri,85            "enableSimpleResponses": self.enable_simple_response,86            "identitySource": self.identity_source,87            "identityValidationExpression": self.identity_validation_expr,88            "jwtConfiguration": self.jwt_config,89            "name": self.name,90        }91class Integration(BaseModel):92    def __init__(93        self,94        connection_id,95        connection_type,96        content_handling_strategy,97        credentials_arn,98        description,99        integration_method,100        integration_type,101        integration_uri,102        passthrough_behavior,103        payload_format_version,104        integration_subtype,105        request_parameters,106        request_templates,107        response_parameters,108        template_selection_expression,109        timeout_in_millis,110        tls_config,111    ):112        self.id = "".join(random.choice(string.ascii_lowercase) for _ in range(8))113        self.connection_id = connection_id114        self.connection_type = connection_type115        self.content_handling_strategy = content_handling_strategy116        self.credentials_arn = credentials_arn117        self.description = description118        self.integration_method = integration_method119        self.integration_response_selection_expression = None120        self.integration_type = integration_type121        self.integration_subtype = integration_subtype122        self.integration_uri = integration_uri123        self.passthrough_behavior = passthrough_behavior124        self.payload_format_version = payload_format_version125        self.request_parameters = request_parameters126        self.request_templates = request_templates127        self.response_parameters = response_parameters128        self.template_selection_expression = template_selection_expression129        self.timeout_in_millis = timeout_in_millis130        self.tls_config = tls_config131        if self.integration_type in ["MOCK", "HTTP"]:132            self.integration_response_selection_expression = (133                "${integration.response.statuscode}"134            )135        elif self.integration_type in ["AWS"]:136            self.integration_response_selection_expression = (137                "${integration.response.body.errorMessage}"138            )139        if (140            self.integration_type in ["AWS", "MOCK", "HTTP"]141            and self.passthrough_behavior is None142        ):143            self.passthrough_behavior = "WHEN_NO_MATCH"144        if self.integration_uri is not None and self.integration_method is None:145            self.integration_method = "POST"146        if self.integration_type in ["AWS", "MOCK"]:147            self.timeout_in_millis = self.timeout_in_millis or 29000148        else:149            self.timeout_in_millis = self.timeout_in_millis or 30000150        self.responses = dict()151    def create_response(152        self,153        content_handling_strategy,154        integration_response_key,155        response_parameters,156        response_templates,157        template_selection_expression,158    ):159        response = IntegrationResponse(160            content_handling_strategy=content_handling_strategy,161            integration_response_key=integration_response_key,162            response_parameters=response_parameters,163            response_templates=response_templates,164            template_selection_expression=template_selection_expression,165        )166        self.responses[response.id] = response167        return response168    def delete_response(self, integration_response_id):169        self.responses.pop(integration_response_id)170    def get_response(self, integration_response_id):171        if integration_response_id not in self.responses:172            raise IntegrationResponseNotFound(integration_response_id)173        return self.responses[integration_response_id]174    def get_responses(self):175        return self.responses.values()176    def update_response(177        self,178        integration_response_id,179        content_handling_strategy,180        integration_response_key,181        response_parameters,182        response_templates,183        template_selection_expression,184    ):185        int_response = self.responses[integration_response_id]186        int_response.update(187            content_handling_strategy=content_handling_strategy,188            integration_response_key=integration_response_key,189            response_parameters=response_parameters,190            response_templates=response_templates,191            template_selection_expression=template_selection_expression,192        )193        return int_response194    def update(195        self,196        connection_id,197        connection_type,198        content_handling_strategy,199        credentials_arn,200        description,201        integration_method,202        integration_type,203        integration_uri,204        passthrough_behavior,205        payload_format_version,206        integration_subtype,207        request_parameters,208        request_templates,209        response_parameters,210        template_selection_expression,211        timeout_in_millis,212        tls_config,213    ):214        if connection_id is not None:215            self.connection_id = connection_id216        if connection_type is not None:217            self.connection_type = connection_type218        if content_handling_strategy is not None:219            self.content_handling_strategy = content_handling_strategy220        if credentials_arn is not None:221            self.credentials_arn = credentials_arn222        if description is not None:223            self.description = description224        if integration_method is not None:225            self.integration_method = integration_method226        if integration_type is not None:227            self.integration_type = integration_type228        if integration_uri is not None:229            self.integration_uri = integration_uri230        if passthrough_behavior is not None:231            self.passthrough_behavior = passthrough_behavior232        if payload_format_version is not None:233            self.payload_format_version = payload_format_version234        if integration_subtype is not None:235            self.integration_subtype = integration_subtype236        if request_parameters is not None:237            # Skip parameters with an empty value238            req_params = {239                key: value for (key, value) in request_parameters.items() if value240            }241            self.request_parameters = req_params242        if request_templates is not None:243            self.request_templates = request_templates244        if response_parameters is not None:245            self.response_parameters = response_parameters246        if template_selection_expression is not None:247            self.template_selection_expression = template_selection_expression248        if timeout_in_millis is not None:249            self.timeout_in_millis = timeout_in_millis250        if tls_config is not None:251            self.tls_config = tls_config252    def to_json(self):253        return {254            "connectionId": self.connection_id,255            "connectionType": self.connection_type,256            "contentHandlingStrategy": self.content_handling_strategy,257            "credentialsArn": self.credentials_arn,258            "description": self.description,259            "integrationId": self.id,260            "integrationMethod": self.integration_method,261            "integrationResponseSelectionExpression": self.integration_response_selection_expression,262            "integrationType": self.integration_type,263            "integrationSubtype": self.integration_subtype,264            "integrationUri": self.integration_uri,265            "passthroughBehavior": self.passthrough_behavior,266            "payloadFormatVersion": self.payload_format_version,267            "requestParameters": self.request_parameters,268            "requestTemplates": self.request_templates,269            "responseParameters": self.response_parameters,270            "templateSelectionExpression": self.template_selection_expression,271            "timeoutInMillis": self.timeout_in_millis,272            "tlsConfig": self.tls_config,273        }274class IntegrationResponse(BaseModel):275    def __init__(276        self,277        content_handling_strategy,278        integration_response_key,279        response_parameters,280        response_templates,281        template_selection_expression,282    ):283        self.id = "".join(random.choice(string.ascii_lowercase) for _ in range(8))284        self.content_handling_strategy = content_handling_strategy285        self.integration_response_key = integration_response_key286        self.response_parameters = response_parameters287        self.response_templates = response_templates288        self.template_selection_expression = template_selection_expression289    def update(290        self,291        content_handling_strategy,292        integration_response_key,293        response_parameters,294        response_templates,295        template_selection_expression,296    ):297        if content_handling_strategy is not None:298            self.content_handling_strategy = content_handling_strategy299        if integration_response_key is not None:300            self.integration_response_key = integration_response_key301        if response_parameters is not None:302            self.response_parameters = response_parameters303        if response_templates is not None:304            self.response_templates = response_templates305        if template_selection_expression is not None:306            self.template_selection_expression = template_selection_expression307    def to_json(self):308        return {309            "integrationResponseId": self.id,310            "integrationResponseKey": self.integration_response_key,311            "contentHandlingStrategy": self.content_handling_strategy,312            "responseParameters": self.response_parameters,313            "responseTemplates": self.response_templates,314            "templateSelectionExpression": self.template_selection_expression,315        }316class Model(BaseModel):317    def __init__(self, content_type, description, name, schema):318        self.id = "".join(random.choice(string.ascii_lowercase) for _ in range(8))319        self.content_type = content_type320        self.description = description321        self.name = name322        self.schema = schema323    def update(self, content_type, description, name, schema):324        if content_type is not None:325            self.content_type = content_type326        if description is not None:327            self.description = description328        if name is not None:329            self.name = name330        if schema is not None:331            self.schema = schema332    def to_json(self):333        return {334            "modelId": self.id,335            "contentType": self.content_type,336            "description": self.description,337            "name": self.name,338            "schema": self.schema,339        }340class RouteResponse(BaseModel):341    def __init__(self, route_response_key, model_selection_expression, response_models):342        self.id = "".join(random.choice(string.ascii_lowercase) for _ in range(8))343        self.route_response_key = route_response_key344        self.model_selection_expression = model_selection_expression345        self.response_models = response_models346    def to_json(self):347        return {348            "modelSelectionExpression": self.model_selection_expression,349            "responseModels": self.response_models,350            "routeResponseId": self.id,351            "routeResponseKey": self.route_response_key,352        }353class Route(BaseModel):354    def __init__(355        self,356        api_key_required,357        authorization_scopes,358        authorization_type,359        authorizer_id,360        model_selection_expression,361        operation_name,362        request_models,363        request_parameters,364        route_key,365        route_response_selection_expression,366        target,367    ):368        self.route_id = "".join(random.choice(string.ascii_lowercase) for _ in range(8))369        self.api_key_required = api_key_required370        self.authorization_scopes = authorization_scopes371        self.authorization_type = authorization_type372        self.authorizer_id = authorizer_id373        self.model_selection_expression = model_selection_expression374        self.operation_name = operation_name375        self.request_models = request_models376        self.request_parameters = request_parameters or {}377        self.route_key = route_key378        self.route_response_selection_expression = route_response_selection_expression379        self.target = target380        self.route_responses = dict()381    def create_route_response(382        self, route_response_key, model_selection_expression, response_models383    ):384        route_response = RouteResponse(385            route_response_key,386            model_selection_expression=model_selection_expression,387            response_models=response_models,388        )389        self.route_responses[route_response.id] = route_response390        return route_response391    def get_route_response(self, route_response_id):392        if route_response_id not in self.route_responses:393            raise RouteResponseNotFound(route_response_id)394        return self.route_responses[route_response_id]395    def delete_route_response(self, route_response_id):396        self.route_responses.pop(route_response_id, None)397    def delete_route_request_parameter(self, request_param):398        del self.request_parameters[request_param]399    def update(400        self,401        api_key_required,402        authorization_scopes,403        authorization_type,404        authorizer_id,405        model_selection_expression,406        operation_name,407        request_models,408        request_parameters,409        route_key,410        route_response_selection_expression,411        target,412    ):413        if api_key_required is not None:414            self.api_key_required = api_key_required415        if authorization_scopes:416            self.authorization_scopes = authorization_scopes417        if authorization_type:418            self.authorization_type = authorization_type419        if authorizer_id is not None:420            self.authorizer_id = authorizer_id421        if model_selection_expression:422            self.model_selection_expression = model_selection_expression423        if operation_name is not None:424            self.operation_name = operation_name425        if request_models:426            self.request_models = request_models427        if request_parameters:428            self.request_parameters = request_parameters429        if route_key:430            self.route_key = route_key431        if route_response_selection_expression is not None:432            self.route_response_selection_expression = (433                route_response_selection_expression434            )435        if target:436            self.target = target437    def to_json(self):438        return {439            "apiKeyRequired": self.api_key_required,440            "authorizationScopes": self.authorization_scopes,441            "authorizationType": self.authorization_type,442            "authorizerId": self.authorizer_id,443            "modelSelectionExpression": self.model_selection_expression,444            "operationName": self.operation_name,445            "requestModels": self.request_models,446            "requestParameters": self.request_parameters,447            "routeId": self.route_id,448            "routeKey": self.route_key,449            "routeResponseSelectionExpression": self.route_response_selection_expression,450            "target": self.target,451        }452class Api(BaseModel):453    def __init__(454        self,455        region,456        name,457        api_key_selection_expression,458        cors_configuration,459        description,460        disable_execute_api_endpoint,461        disable_schema_validation,462        protocol_type,463        route_selection_expression,464        tags,465        version,466        backend,467    ):468        self.api_id = "".join(random.choice(string.ascii_lowercase) for _ in range(8))469        self.api_endpoint = f"https://{self.api_id}.execute-api.{region}.amazonaws.com"470        self.backend = backend471        self.name = name472        self.api_key_selection_expression = (473            api_key_selection_expression or "$request.header.x-api-key"474        )475        self.created_date = unix_time()476        self.cors_configuration = cors_configuration477        self.description = description478        self.disable_execute_api_endpoint = disable_execute_api_endpoint or False479        self.disable_schema_validation = disable_schema_validation480        self.protocol_type = protocol_type481        self.route_selection_expression = (482            route_selection_expression or "$request.method $request.path"483        )484        self.version = version485        self.authorizers = dict()486        self.integrations = dict()487        self.models = dict()488        self.routes = dict()489        self.arn = f"arn:aws:apigateway:{region}::/apis/{self.api_id}"490        self.backend.tag_resource(self.arn, tags)491    def clear(self):492        self.authorizers = dict()493        self.integrations = dict()494        self.models = dict()495        self.routes = dict()496    def delete_cors_configuration(self):497        self.cors_configuration = None498    def create_authorizer(499        self,500        auth_creds_arn,501        auth_payload_format_version,502        auth_result_ttl,503        authorizer_type,504        authorizer_uri,505        enable_simple_response,506        identity_source,507        identity_validation_expr,508        jwt_config,509        name,510    ):511        authorizer = Authorizer(512            auth_creds_arn=auth_creds_arn,513            auth_payload_format_version=auth_payload_format_version,514            auth_result_ttl=auth_result_ttl,515            authorizer_type=authorizer_type,516            authorizer_uri=authorizer_uri,517            enable_simple_response=enable_simple_response,518            identity_source=identity_source,519            identity_validation_expr=identity_validation_expr,520            jwt_config=jwt_config,521            name=name,522        )523        self.authorizers[authorizer.id] = authorizer524        return authorizer525    def delete_authorizer(self, authorizer_id):526        self.authorizers.pop(authorizer_id, None)527    def get_authorizer(self, authorizer_id):528        if authorizer_id not in self.authorizers:529            raise AuthorizerNotFound(authorizer_id)530        return self.authorizers[authorizer_id]531    def update_authorizer(532        self,533        authorizer_id,534        auth_creds_arn,535        auth_payload_format_version,536        auth_result_ttl,537        authorizer_type,538        authorizer_uri,539        enable_simple_response,540        identity_source,541        identity_validation_expr,542        jwt_config,543        name,544    ):545        authorizer = self.authorizers[authorizer_id]546        authorizer.update(547            auth_creds_arn=auth_creds_arn,548            auth_payload_format_version=auth_payload_format_version,549            auth_result_ttl=auth_result_ttl,550            authorizer_type=authorizer_type,551            authorizer_uri=authorizer_uri,552            enable_simple_response=enable_simple_response,553            identity_source=identity_source,554            identity_validation_expr=identity_validation_expr,555            jwt_config=jwt_config,556            name=name,557        )558        return authorizer559    def create_model(self, content_type, description, name, schema):560        model = Model(content_type, description, name, schema)561        self.models[model.id] = model562        return model563    def delete_model(self, model_id):564        self.models.pop(model_id, None)565    def get_model(self, model_id):566        if model_id not in self.models:567            raise ModelNotFound(model_id)568        return self.models[model_id]569    def update_model(self, model_id, content_type, description, name, schema):570        model = self.models[model_id]571        model.update(content_type, description, name, schema)572        return model573    def import_api(self, body, fail_on_warnings):574        self.clear()575        body = yaml.safe_load(body)576        for path, path_details in body.get("paths", {}).items():577            for method, method_details in path_details.items():578                route_key = f"{method.upper()} {path}"579                for int_type, type_details in method_details.items():580                    if int_type == "responses":581                        for status_code, response_details in type_details.items():582                            content = response_details.get("content", {})583                            for content_type in content.values():584                                for ref in content_type.get("schema", {}).values():585                                    if ref not in self.models and fail_on_warnings:586                                        attr = f"paths.'{path}'({method}).{int_type}.{status_code}.content.schema.{ref}"587                                        raise BadRequestException(588                                            f"Warnings found during import:\n\tParse issue: attribute {attr} is missing"589                                        )590                    if int_type == "x-amazon-apigateway-integration":591                        integration = self.create_integration(592                            connection_type="INTERNET",593                            description="AutoCreate from OpenAPI Import",594                            integration_type=type_details.get("type"),595                            integration_method=type_details.get("httpMethod"),596                            payload_format_version=type_details.get(597                                "payloadFormatVersion"598                            ),599                            integration_uri=type_details.get("uri"),600                        )601                        self.create_route(602                            api_key_required=False,603                            authorization_scopes=[],604                            route_key=route_key,605                            target=f"integrations/{integration.id}",606                        )607        if "title" in body.get("info", {}):608            self.name = body["info"]["title"]609        if "version" in body.get("info", {}):610            self.version = str(body["info"]["version"])611        if "x-amazon-apigateway-cors" in body:612            self.cors_configuration = body["x-amazon-apigateway-cors"]613    def update(614        self,615        api_key_selection_expression,616        cors_configuration,617        description,618        disable_schema_validation,619        disable_execute_api_endpoint,620        name,621        route_selection_expression,622        version,623    ):624        if api_key_selection_expression is not None:625            self.api_key_selection_expression = api_key_selection_expression626        if cors_configuration is not None:627            self.cors_configuration = cors_configuration628        if description is not None:629            self.description = description630        if disable_execute_api_endpoint is not None:631            self.disable_execute_api_endpoint = disable_execute_api_endpoint632        if disable_schema_validation is not None:633            self.disable_schema_validation = disable_schema_validation634        if name is not None:635            self.name = name636        if route_selection_expression is not None:637            self.route_selection_expression = route_selection_expression638        if version is not None:639            self.version = version640    def create_integration(641        self,642        connection_type,643        description,644        integration_method,645        integration_type,646        integration_uri,647        connection_id=None,648        content_handling_strategy=None,649        credentials_arn=None,650        passthrough_behavior=None,651        payload_format_version=None,652        integration_subtype=None,653        request_parameters=None,654        request_templates=None,655        response_parameters=None,656        template_selection_expression=None,657        timeout_in_millis=None,658        tls_config=None,659    ):660        integration = Integration(661            connection_id=connection_id,662            connection_type=connection_type,663            content_handling_strategy=content_handling_strategy,664            credentials_arn=credentials_arn,665            description=description,666            integration_method=integration_method,667            integration_type=integration_type,668            integration_uri=integration_uri,669            passthrough_behavior=passthrough_behavior,670            payload_format_version=payload_format_version,671            integration_subtype=integration_subtype,672            request_parameters=request_parameters,673            request_templates=request_templates,674            response_parameters=response_parameters,675            template_selection_expression=template_selection_expression,676            timeout_in_millis=timeout_in_millis,677            tls_config=tls_config,678        )679        self.integrations[integration.id] = integration680        return integration681    def delete_integration(self, integration_id):682        self.integrations.pop(integration_id, None)683    def get_integration(self, integration_id):684        if integration_id not in self.integrations:685            raise IntegrationNotFound(integration_id)686        return self.integrations[integration_id]687    def get_integrations(self):688        return self.integrations.values()689    def update_integration(690        self,691        integration_id,692        connection_id,693        connection_type,694        content_handling_strategy,695        credentials_arn,696        description,697        integration_method,698        integration_type,699        integration_uri,700        passthrough_behavior,701        payload_format_version,702        integration_subtype,703        request_parameters,704        request_templates,705        response_parameters,706        template_selection_expression,707        timeout_in_millis,708        tls_config,709    ):710        integration = self.integrations[integration_id]711        integration.update(712            connection_id=connection_id,713            connection_type=connection_type,714            content_handling_strategy=content_handling_strategy,715            credentials_arn=credentials_arn,716            description=description,717            integration_method=integration_method,718            integration_type=integration_type,719            integration_uri=integration_uri,720            passthrough_behavior=passthrough_behavior,721            payload_format_version=payload_format_version,722            integration_subtype=integration_subtype,723            request_parameters=request_parameters,724            request_templates=request_templates,725            response_parameters=response_parameters,726            template_selection_expression=template_selection_expression,727            timeout_in_millis=timeout_in_millis,728            tls_config=tls_config,729        )730        return integration731    def create_integration_response(732        self,733        integration_id,734        content_handling_strategy,735        integration_response_key,736        response_parameters,737        response_templates,738        template_selection_expression,739    ):740        integration = self.get_integration(integration_id)741        return integration.create_response(742            content_handling_strategy=content_handling_strategy,743            integration_response_key=integration_response_key,744            response_parameters=response_parameters,745            response_templates=response_templates,746            template_selection_expression=template_selection_expression,747        )748    def delete_integration_response(self, integration_id, integration_response_id):749        integration = self.get_integration(integration_id)750        integration.delete_response(integration_response_id)751    def get_integration_response(self, integration_id, integration_response_id):752        integration = self.get_integration(integration_id)753        return integration.get_response(integration_response_id)754    def get_integration_responses(self, integration_id):755        integration = self.get_integration(integration_id)756        return integration.get_responses()757    def update_integration_response(758        self,759        integration_id,760        integration_response_id,761        content_handling_strategy,762        integration_response_key,763        response_parameters,764        response_templates,765        template_selection_expression,766    ):767        integration = self.get_integration(integration_id)768        return integration.update_response(769            integration_response_id=integration_response_id,770            content_handling_strategy=content_handling_strategy,771            integration_response_key=integration_response_key,772            response_parameters=response_parameters,773            response_templates=response_templates,774            template_selection_expression=template_selection_expression,775        )776    def create_route(777        self,778        api_key_required,779        authorization_scopes,780        route_key,781        target,782        authorization_type=None,783        authorizer_id=None,784        model_selection_expression=None,785        operation_name=None,786        request_models=None,787        request_parameters=None,788        route_response_selection_expression=None,789    ):790        route = Route(791            api_key_required=api_key_required,792            authorization_scopes=authorization_scopes,793            authorization_type=authorization_type,794            authorizer_id=authorizer_id,795            model_selection_expression=model_selection_expression,796            operation_name=operation_name,797            request_models=request_models,798            request_parameters=request_parameters,799            route_key=route_key,800            route_response_selection_expression=route_response_selection_expression,801            target=target,802        )803        self.routes[route.route_id] = route804        return route805    def delete_route(self, route_id):806        self.routes.pop(route_id, None)807    def delete_route_request_parameter(self, route_id, request_param):808        route = self.get_route(route_id)809        route.delete_route_request_parameter(request_param)810    def get_route(self, route_id):811        if route_id not in self.routes:812            raise RouteNotFound(route_id)813        return self.routes[route_id]814    def get_routes(self):815        return self.routes.values()816    def update_route(817        self,818        route_id,819        api_key_required,820        authorization_scopes,821        authorization_type,822        authorizer_id,823        model_selection_expression,824        operation_name,825        request_models,826        request_parameters,827        route_key,828        route_response_selection_expression,829        target,830    ):831        route = self.get_route(route_id)832        route.update(833            api_key_required=api_key_required,834            authorization_scopes=authorization_scopes,835            authorization_type=authorization_type,836            authorizer_id=authorizer_id,837            model_selection_expression=model_selection_expression,838            operation_name=operation_name,839            request_models=request_models,840            request_parameters=request_parameters,841            route_key=route_key,842            route_response_selection_expression=route_response_selection_expression,843            target=target,844        )845        return route846    def create_route_response(847        self, route_id, route_response_key, model_selection_expression, response_models848    ):849        route = self.get_route(route_id)850        return route.create_route_response(851            route_response_key,852            model_selection_expression=model_selection_expression,853            response_models=response_models,854        )855    def delete_route_response(self, route_id, route_response_id):856        route = self.get_route(route_id)857        route.delete_route_response(route_response_id)858    def get_route_response(self, route_id, route_response_id):859        route = self.get_route(route_id)860        return route.get_route_response(route_response_id)861    def to_json(self):862        return {863            "apiId": self.api_id,864            "apiEndpoint": self.api_endpoint,865            "apiKeySelectionExpression": self.api_key_selection_expression,866            "createdDate": self.created_date,867            "corsConfiguration": self.cors_configuration,868            "description": self.description,869            "disableExecuteApiEndpoint": self.disable_execute_api_endpoint,870            "disableSchemaValidation": self.disable_schema_validation,871            "name": self.name,872            "protocolType": self.protocol_type,873            "routeSelectionExpression": self.route_selection_expression,874            "tags": self.backend.get_tags(self.arn),875            "version": self.version,876        }877class VpcLink(BaseModel):878    def __init__(self, name, sg_ids, subnet_ids, tags, backend):879        self.created = unix_time()880        self.id = "".join(random.choice(string.ascii_lowercase) for _ in range(8))881        self.name = name882        self.sg_ids = sg_ids883        self.subnet_ids = subnet_ids884        self.arn = f"arn:aws:apigateway:{backend.region_name}::/vpclinks/{self.id}"885        self.backend = backend886        self.backend.tag_resource(self.arn, tags)887    def update(self, name):888        self.name = name889    def to_json(self):890        return {891            "createdDate": self.created,892            "name": self.name,893            "securityGroupIds": self.sg_ids,894            "subnetIds": self.subnet_ids,895            "tags": self.backend.get_tags(self.arn),896            "vpcLinkId": self.id,897            "vpcLinkStatus": "AVAILABLE",898            "vpcLinkVersion": "V2",899        }900class ApiGatewayV2Backend(BaseBackend):901    """Implementation of ApiGatewayV2 APIs."""902    def __init__(self, region_name=None):903        self.region_name = region_name904        self.apis = dict()905        self.vpc_links = dict()906        self.tagger = TaggingService()907    def reset(self):908        """Re-initialize all attributes for this instance."""909        region_name = self.region_name910        self.__dict__ = {}911        self.__init__(region_name)912    def create_api(913        self,914        api_key_selection_expression,915        cors_configuration,916        credentials_arn,917        description,918        disable_schema_validation,919        disable_execute_api_endpoint,920        name,921        protocol_type,922        route_key,923        route_selection_expression,924        tags,925        target,926        version,927    ):928        """929        The following parameters are not yet implemented:930        CredentialsArn, RouteKey, Tags, Target931        """932        api = Api(933            region=self.region_name,934            cors_configuration=cors_configuration,935            description=description,936            name=name,937            api_key_selection_expression=api_key_selection_expression,938            disable_execute_api_endpoint=disable_execute_api_endpoint,939            disable_schema_validation=disable_schema_validation,940            protocol_type=protocol_type,941            route_selection_expression=route_selection_expression,942            tags=tags,943            version=version,944            backend=self,945        )946        self.apis[api.api_id] = api947        return api948    def delete_api(self, api_id):949        self.apis.pop(api_id, None)950    def get_api(self, api_id):951        if api_id not in self.apis:952            raise ApiNotFound(api_id)953        return self.apis[api_id]954    def get_apis(self):955        """956        Pagination is not yet implemented957        """958        return self.apis.values()959    def update_api(960        self,961        api_id,962        api_key_selection_expression,963        cors_configuration,964        description,965        disable_schema_validation,966        disable_execute_api_endpoint,967        name,968        route_selection_expression,969        version,970    ):971        """972        The following parameters have not yet been implemented: CredentialsArn, RouteKey, Target973        """974        api = self.get_api(api_id)975        api.update(976            api_key_selection_expression=api_key_selection_expression,977            cors_configuration=cors_configuration,978            description=description,979            disable_schema_validation=disable_schema_validation,980            disable_execute_api_endpoint=disable_execute_api_endpoint,981            name=name,982            route_selection_expression=route_selection_expression,983            version=version,984        )985        return api986    def reimport_api(self, api_id, body, fail_on_warnings):987        """988        Only YAML is supported at the moment. Full OpenAPI-support is not guaranteed. Only limited validation is implemented989        """990        api = self.get_api(api_id)991        api.import_api(body, fail_on_warnings)992        return api993    def delete_cors_configuration(self, api_id):994        api = self.get_api(api_id)995        api.delete_cors_configuration()996    def create_authorizer(997        self,998        api_id,999        auth_creds_arn,1000        auth_payload_format_version,1001        auth_result_ttl,1002        authorizer_uri,1003        authorizer_type,1004        enable_simple_response,1005        identity_source,1006        identity_validation_expr,1007        jwt_config,1008        name,1009    ):1010        api = self.get_api(api_id)1011        authorizer = api.create_authorizer(1012            auth_creds_arn=auth_creds_arn,1013            auth_payload_format_version=auth_payload_format_version,1014            auth_result_ttl=auth_result_ttl,1015            authorizer_type=authorizer_type,1016            authorizer_uri=authorizer_uri,1017            enable_simple_response=enable_simple_response,1018            identity_source=identity_source,1019            identity_validation_expr=identity_validation_expr,1020            jwt_config=jwt_config,1021            name=name,1022        )1023        return authorizer1024    def delete_authorizer(self, api_id, authorizer_id):1025        api = self.get_api(api_id)1026        api.delete_authorizer(authorizer_id=authorizer_id)1027    def get_authorizer(self, api_id, authorizer_id):1028        api = self.get_api(api_id)1029        authorizer = api.get_authorizer(authorizer_id=authorizer_id)1030        return authorizer1031    def update_authorizer(1032        self,1033        api_id,1034        authorizer_id,1035        auth_creds_arn,1036        auth_payload_format_version,1037        auth_result_ttl,1038        authorizer_uri,1039        authorizer_type,1040        enable_simple_response,1041        identity_source,1042        identity_validation_expr,1043        jwt_config,1044        name,1045    ):1046        api = self.get_api(api_id)1047        authorizer = api.update_authorizer(1048            authorizer_id=authorizer_id,1049            auth_creds_arn=auth_creds_arn,1050            auth_payload_format_version=auth_payload_format_version,1051            auth_result_ttl=auth_result_ttl,1052            authorizer_type=authorizer_type,1053            authorizer_uri=authorizer_uri,1054            enable_simple_response=enable_simple_response,1055            identity_source=identity_source,1056            identity_validation_expr=identity_validation_expr,1057            jwt_config=jwt_config,1058            name=name,1059        )1060        return authorizer1061    def create_model(self, api_id, content_type, description, name, schema):1062        api = self.get_api(api_id)1063        model = api.create_model(1064            content_type=content_type, description=description, name=name, schema=schema1065        )1066        return model1067    def delete_model(self, api_id, model_id):1068        api = self.get_api(api_id)1069        api.delete_model(model_id=model_id)1070    def get_model(self, api_id, model_id):1071        api = self.get_api(api_id)1072        return api.get_model(model_id)1073    def update_model(self, api_id, model_id, content_type, description, name, schema):1074        api = self.get_api(api_id)1075        return api.update_model(model_id, content_type, description, name, schema)1076    def get_tags(self, resource_id):1077        return self.tagger.get_tag_dict_for_resource(resource_id)1078    def tag_resource(self, resource_arn, tags):1079        tags = TaggingService.convert_dict_to_tags_input(tags or {})1080        self.tagger.tag_resource(resource_arn, tags)1081    def untag_resource(self, resource_arn, tag_keys):1082        self.tagger.untag_resource_using_names(resource_arn, tag_keys)1083    def create_route(1084        self,1085        api_id,1086        api_key_required,1087        authorization_scopes,1088        authorization_type,1089        authorizer_id,1090        model_selection_expression,1091        operation_name,1092        request_models,1093        request_parameters,1094        route_key,1095        route_response_selection_expression,1096        target,1097    ):1098        api = self.get_api(api_id)1099        route = api.create_route(1100            api_key_required=api_key_required,1101            authorization_scopes=authorization_scopes,1102            authorization_type=authorization_type,1103            authorizer_id=authorizer_id,1104            model_selection_expression=model_selection_expression,1105            operation_name=operation_name,1106            request_models=request_models,1107            request_parameters=request_parameters,1108            route_key=route_key,1109            route_response_selection_expression=route_response_selection_expression,1110            target=target,1111        )1112        return route1113    def delete_route(self, api_id, route_id):1114        api = self.get_api(api_id)1115        api.delete_route(route_id)1116    def delete_route_request_parameter(self, api_id, route_id, request_param):1117        api = self.get_api(api_id)1118        api.delete_route_request_parameter(route_id, request_param)1119    def get_route(self, api_id, route_id):1120        api = self.get_api(api_id)1121        return api.get_route(route_id)1122    def get_routes(self, api_id):1123        """1124        Pagination is not yet implemented1125        """1126        api = self.get_api(api_id)1127        return api.get_routes()1128    def update_route(1129        self,1130        api_id,1131        api_key_required,1132        authorization_scopes,1133        authorization_type,1134        authorizer_id,1135        model_selection_expression,1136        operation_name,1137        request_models,1138        request_parameters,1139        route_id,1140        route_key,1141        route_response_selection_expression,1142        target,1143    ):1144        api = self.get_api(api_id)1145        route = api.update_route(1146            route_id=route_id,1147            api_key_required=api_key_required,1148            authorization_scopes=authorization_scopes,1149            authorization_type=authorization_type,1150            authorizer_id=authorizer_id,1151            model_selection_expression=model_selection_expression,1152            operation_name=operation_name,1153            request_models=request_models,1154            request_parameters=request_parameters,1155            route_key=route_key,1156            route_response_selection_expression=route_response_selection_expression,1157            target=target,1158        )1159        return route1160    def create_route_response(1161        self,1162        api_id,1163        route_id,1164        route_response_key,1165        model_selection_expression,1166        response_models,1167    ):1168        """1169        The following parameters are not yet implemented: ResponseModels, ResponseParameters1170        """1171        api = self.get_api(api_id)1172        return api.create_route_response(1173            route_id,1174            route_response_key,1175            model_selection_expression=model_selection_expression,1176            response_models=response_models,1177        )1178    def delete_route_response(self, api_id, route_id, route_response_id):1179        api = self.get_api(api_id)1180        api.delete_route_response(route_id, route_response_id)1181    def get_route_response(self, api_id, route_id, route_response_id):1182        api = self.get_api(api_id)1183        return api.get_route_response(route_id, route_response_id)1184    def create_integration(1185        self,1186        api_id,1187        connection_id,1188        connection_type,1189        content_handling_strategy,1190        credentials_arn,1191        description,1192        integration_method,1193        integration_subtype,1194        integration_type,1195        integration_uri,1196        passthrough_behavior,1197        payload_format_version,1198        request_parameters,1199        request_templates,1200        response_parameters,1201        template_selection_expression,1202        timeout_in_millis,1203        tls_config,1204    ):1205        api = self.get_api(api_id)1206        integration = api.create_integration(1207            connection_id=connection_id,1208            connection_type=connection_type,1209            content_handling_strategy=content_handling_strategy,1210            credentials_arn=credentials_arn,1211            description=description,1212            integration_method=integration_method,1213            integration_type=integration_type,1214            integration_uri=integration_uri,1215            passthrough_behavior=passthrough_behavior,1216            payload_format_version=payload_format_version,1217            integration_subtype=integration_subtype,1218            request_parameters=request_parameters,1219            request_templates=request_templates,1220            response_parameters=response_parameters,1221            template_selection_expression=template_selection_expression,1222            timeout_in_millis=timeout_in_millis,1223            tls_config=tls_config,1224        )1225        return integration1226    def get_integration(self, api_id, integration_id):1227        api = self.get_api(api_id)1228        integration = api.get_integration(integration_id)1229        return integration1230    def get_integrations(self, api_id):1231        """1232        Pagination is not yet implemented1233        """1234        api = self.get_api(api_id)1235        return api.get_integrations()1236    def delete_integration(self, api_id, integration_id):1237        api = self.get_api(api_id)1238        api.delete_integration(integration_id)1239    def update_integration(1240        self,1241        api_id,1242        connection_id,1243        connection_type,1244        content_handling_strategy,1245        credentials_arn,1246        description,1247        integration_id,1248        integration_method,1249        integration_subtype,1250        integration_type,1251        integration_uri,1252        passthrough_behavior,1253        payload_format_version,1254        request_parameters,1255        request_templates,1256        response_parameters,1257        template_selection_expression,1258        timeout_in_millis,1259        tls_config,1260    ):1261        api = self.get_api(api_id)1262        integration = api.update_integration(1263            integration_id=integration_id,1264            connection_id=connection_id,1265            connection_type=connection_type,1266            content_handling_strategy=content_handling_strategy,1267            credentials_arn=credentials_arn,1268            description=description,1269            integration_method=integration_method,1270            integration_type=integration_type,1271            integration_uri=integration_uri,1272            passthrough_behavior=passthrough_behavior,1273            payload_format_version=payload_format_version,1274            integration_subtype=integration_subtype,1275            request_parameters=request_parameters,1276            request_templates=request_templates,1277            response_parameters=response_parameters,1278            template_selection_expression=template_selection_expression,1279            timeout_in_millis=timeout_in_millis,1280            tls_config=tls_config,1281        )1282        return integration1283    def create_integration_response(1284        self,1285        api_id,1286        integration_id,1287        content_handling_strategy,1288        integration_response_key,1289        response_parameters,1290        response_templates,1291        template_selection_expression,1292    ):1293        api = self.get_api(api_id)1294        integration_response = api.create_integration_response(1295            integration_id=integration_id,1296            content_handling_strategy=content_handling_strategy,1297            integration_response_key=integration_response_key,1298            response_parameters=response_parameters,1299            response_templates=response_templates,1300            template_selection_expression=template_selection_expression,1301        )1302        return integration_response1303    def delete_integration_response(1304        self, api_id, integration_id, integration_response_id1305    ):1306        api = self.get_api(api_id)1307        api.delete_integration_response(1308            integration_id, integration_response_id=integration_response_id1309        )1310    def get_integration_response(self, api_id, integration_id, integration_response_id):1311        api = self.get_api(api_id)1312        return api.get_integration_response(1313            integration_id, integration_response_id=integration_response_id1314        )1315    def get_integration_responses(self, api_id, integration_id):1316        api = self.get_api(api_id)1317        return api.get_integration_responses(integration_id)1318    def update_integration_response(1319        self,1320        api_id,1321        integration_id,1322        integration_response_id,1323        content_handling_strategy,1324        integration_response_key,1325        response_parameters,1326        response_templates,1327        template_selection_expression,1328    ):1329        api = self.get_api(api_id)1330        integration_response = api.update_integration_response(1331            integration_id=integration_id,1332            integration_response_id=integration_response_id,1333            content_handling_strategy=content_handling_strategy,1334            integration_response_key=integration_response_key,1335            response_parameters=response_parameters,1336            response_templates=response_templates,1337            template_selection_expression=template_selection_expression,1338        )1339        return integration_response1340    def create_vpc_link(self, name, sg_ids, subnet_ids, tags):1341        vpc_link = VpcLink(1342            name, sg_ids=sg_ids, subnet_ids=subnet_ids, tags=tags, backend=self1343        )1344        self.vpc_links[vpc_link.id] = vpc_link1345        return vpc_link1346    def get_vpc_link(self, vpc_link_id):1347        if vpc_link_id not in self.vpc_links:1348            raise VpcLinkNotFound(vpc_link_id)1349        return self.vpc_links[vpc_link_id]1350    def delete_vpc_link(self, vpc_link_id):1351        self.vpc_links.pop(vpc_link_id, None)1352    def get_vpc_links(self):1353        return self.vpc_links.values()1354    def update_vpc_link(self, vpc_link_id, name):1355        vpc_link = self.get_vpc_link(vpc_link_id)1356        vpc_link.update(name)1357        return vpc_link...

test_async_integration.py

Source:test_async_integration.py

...4from src.async_classes.async_data_enricher import DataEnricher5from src.async_classes.async_integration import Integration6from src.data_classes.data_classes_for_integration import IntegrationSettings7@pytest.fixture8def integration_uri():9    return 'uri'10@pytest.fixture11def db_conn():12    return 'db_connection'13@pytest.fixture14def integration_uri():15    return 'uri'16@pytest.fixture17def customer_id():18    return 'customer'19@pytest.fixture20def integration_settings(integration_uri):21    return IntegrationSettings(integration_uri=integration_uri)22@pytest.fixture23def enricher(db_conn, customer_id):24    return DataEnricher(db_conn, customer_id)25@pytest.fixture26def async_integration(integration_settings, enricher):27    return Integration(integration_settings, enricher)28@pytest.mark.asyncio...

IntegrationBaseUriRule.py

Source:IntegrationBaseUriRule.py

1from .rule_validator import RuleViolation2from .rules_helper import get_path_verbs, get_apigateway_integration3class IntegrationBaseUriRule:4    base_uri = ''5    def __init__(self, base_uri):6        self.rule_name = 'integration_base_uri'7        self.base_uri = base_uri8    def validate(self, spec):9        violations = []10        for path in spec['paths']:11            for path_verb in get_path_verbs(spec, path):12                if path_verb == 'options':13                    continue14                integration = get_apigateway_integration(spec, path, path_verb)15                integration_uri = integration['uri']16                if not integration_uri.startswith(self.base_uri):17                    violations.append(RuleViolation('integration_base_uri',18                                                    message='Base URI "{}" not present at the beginning of URI "{}"'19                                                    .format(self.base_uri, integration_uri),20                                                    path=path))...

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.