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

How to use delete_url_config method in localstack

Best Python code snippet using localstack_python

lambda_api.py

Source:lambda_api.py

...1335    response = prev_url_config.copy()1336    response.pop("CustomId")1337    return response1338@app.route("%s/functions/<function>/url" % API_PATH_ROOT_2, methods=["DELETE"])1339def delete_url_config(function):1340    arn = func_arn(function)1341    qualifier = request.args.get("Qualifier")1342    q_arn = func_qualifier(function, qualifier)1343    arn = q_arn or arn1344    lambda_backend = LambdaRegion.get()1345    if arn not in lambda_backend.url_configs:1346        response = error_response("Function does not exist", 404, "ResourceNotFoundException")1347        return response1348    lambda_backend.url_configs.pop(arn)1349    return {}1350def add_permission_policy_statement(1351    resource_name, resource_arn, resource_arn_qualified, qualifier=None1352):1353    region = LambdaRegion.get()...

models.py

Source:models.py

...797        return alias798    def create_url_config(self, config):799        self.url_config = FunctionUrlConfig(function=self, config=config)800        return self.url_config801    def delete_url_config(self):802        self.url_config = None803    def get_url_config(self):804        if not self.url_config:805            raise FunctionUrlConfigNotFound()806        return self.url_config807    def update_url_config(self, config):808        self.url_config.update(config)809        return self.url_config810class FunctionUrlConfig:811    def __init__(self, function: LambdaFunction, config):812        self.function = function813        self.config = config814        self.url = f"https://{uuid.uuid4().hex}.lambda-url.{function.region}.on.aws"815        self.created = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S")816        self.last_modified = self.created817    def to_dict(self):818        return {819            "FunctionUrl": self.url,820            "FunctionArn": self.function.function_arn,821            "AuthType": self.config.get("AuthType"),822            "Cors": self.config.get("Cors"),823            "CreationTime": self.created,824            "LastModifiedTime": self.last_modified,825        }826    def update(self, new_config):827        if new_config.get("Cors"):828            self.config["Cors"] = new_config["Cors"]829        if new_config.get("AuthType"):830            self.config["AuthType"] = new_config["AuthType"]831        self.last_modified = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S")832class EventSourceMapping(CloudFormationModel):833    def __init__(self, spec):834        # required835        self.function_name = spec["FunctionName"]836        self.event_source_arn = spec["EventSourceArn"]837        # optional838        self.batch_size = spec.get("BatchSize")839        self.starting_position = spec.get("StartingPosition", "TRIM_HORIZON")840        self.enabled = spec.get("Enabled", True)841        self.starting_position_timestamp = spec.get("StartingPositionTimestamp", None)842        self.function_arn = spec["FunctionArn"]843        self.uuid = str(uuid.uuid4())844        self.last_modified = time.mktime(datetime.datetime.utcnow().timetuple())845    def _get_service_source_from_arn(self, event_source_arn):846        return event_source_arn.split(":")[2].lower()847    def _validate_event_source(self, event_source_arn):848        valid_services = ("dynamodb", "kinesis", "sqs")849        service = self._get_service_source_from_arn(event_source_arn)850        return True if service in valid_services else False851    @property852    def event_source_arn(self):853        return self._event_source_arn854    @event_source_arn.setter855    def event_source_arn(self, event_source_arn):856        if not self._validate_event_source(event_source_arn):857            raise ValueError(858                "InvalidParameterValueException", "Unsupported event source type"859            )860        self._event_source_arn = event_source_arn861    @property862    def batch_size(self):863        return self._batch_size864    @batch_size.setter865    def batch_size(self, batch_size):866        batch_size_service_map = {867            "kinesis": (100, 10000),868            "dynamodb": (100, 1000),869            "sqs": (10, 10),870        }871        source_type = self._get_service_source_from_arn(self.event_source_arn)872        batch_size_for_source = batch_size_service_map[source_type]873        if batch_size is None:874            self._batch_size = batch_size_for_source[0]875        elif batch_size > batch_size_for_source[1]:876            error_message = "BatchSize {} exceeds the max of {}".format(877                batch_size, batch_size_for_source[1]878            )879            raise ValueError("InvalidParameterValueException", error_message)880        else:881            self._batch_size = int(batch_size)882    def get_configuration(self):883        return {884            "UUID": self.uuid,885            "BatchSize": self.batch_size,886            "EventSourceArn": self.event_source_arn,887            "FunctionArn": self.function_arn,888            "LastModified": self.last_modified,889            "LastProcessingResult": "",890            "State": "Enabled" if self.enabled else "Disabled",891            "StateTransitionReason": "User initiated",892        }893    def delete(self, account_id, region_name):894        lambda_backend = lambda_backends[account_id][region_name]895        lambda_backend.delete_event_source_mapping(self.uuid)896    @staticmethod897    def cloudformation_name_type():898        return None899    @staticmethod900    def cloudformation_type():901        # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html902        return "AWS::Lambda::EventSourceMapping"903    @classmethod904    def create_from_cloudformation_json(905        cls, resource_name, cloudformation_json, account_id, region_name, **kwargs906    ):907        properties = cloudformation_json["Properties"]908        lambda_backend = lambda_backends[account_id][region_name]909        return lambda_backend.create_event_source_mapping(properties)910    @classmethod911    def update_from_cloudformation_json(912        cls,913        original_resource,914        new_resource_name,915        cloudformation_json,916        account_id,917        region_name,918    ):919        properties = cloudformation_json["Properties"]920        event_source_uuid = original_resource.uuid921        lambda_backend = lambda_backends[account_id][region_name]922        return lambda_backend.update_event_source_mapping(event_source_uuid, properties)923    @classmethod924    def delete_from_cloudformation_json(925        cls, resource_name, cloudformation_json, account_id, region_name926    ):927        properties = cloudformation_json["Properties"]928        lambda_backend = lambda_backends[account_id][region_name]929        esms = lambda_backend.list_event_source_mappings(930            event_source_arn=properties["EventSourceArn"],931            function_name=properties["FunctionName"],932        )933        for esm in esms:934            if esm.uuid == resource_name:935                esm.delete(account_id, region_name)936    @property937    def physical_resource_id(self):938        return self.uuid939class LambdaVersion(CloudFormationModel):940    def __init__(self, spec):941        self.version = spec["Version"]942    def __repr__(self):943        return str(self.logical_resource_id)944    @staticmethod945    def cloudformation_name_type():946        return None947    @staticmethod948    def cloudformation_type():949        # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-version.html950        return "AWS::Lambda::Version"951    @classmethod952    def create_from_cloudformation_json(953        cls, resource_name, cloudformation_json, account_id, region_name, **kwargs954    ):955        properties = cloudformation_json["Properties"]956        function_name = properties["FunctionName"]957        func = lambda_backends[account_id][region_name].publish_function(function_name)958        spec = {"Version": func.version}959        return LambdaVersion(spec)960class LambdaStorage(object):961    def __init__(self, region_name, account_id):962        # Format 'func_name' {'versions': []}963        self._functions = {}964        self._aliases = dict()965        self._arns = weakref.WeakValueDictionary()966        self.region_name = region_name967        self.account_id = account_id968    def _get_latest(self, name):969        return self._functions[name]["latest"]970    def _get_version(self, name, version):971        index = version - 1972        try:973            return self._functions[name]["versions"][index]974        except IndexError:975            return None976    def delete_alias(self, name, function_name):977        fn = self.get_function_by_name_or_arn(function_name)978        return fn.delete_alias(name)979    def get_alias(self, name, function_name):980        fn = self.get_function_by_name_or_arn(function_name)981        return fn.get_alias(name)982    def put_alias(983        self, name, function_name, function_version, description, routing_config984    ):985        fn = self.get_function_by_name_or_arn(function_name)986        return fn.put_alias(name, description, function_version, routing_config)987    def update_alias(988        self, name, function_name, function_version, description, routing_config989    ):990        fn = self.get_function_by_name_or_arn(function_name)991        return fn.update_alias(name, description, function_version, routing_config)992    def get_function_by_name(self, name, qualifier=None):993        if name not in self._functions:994            return None995        if qualifier is None:996            return self._get_latest(name)997        try:998            return self._get_version(name, int(qualifier))999        except ValueError:1000            return self._functions[name]["latest"]1001    def list_versions_by_function(self, name):1002        if name not in self._functions:1003            return None1004        latest = copy.copy(self._functions[name]["latest"])1005        latest.function_arn += ":$LATEST"1006        return [latest] + self._functions[name]["versions"]1007    def get_arn(self, arn):1008        # Function ARN may contain an alias1009        # arn:aws:lambda:region:account_id:function:<fn_name>:<alias_name>1010        if ":" in arn.split(":function:")[-1]:1011            # arn = arn:aws:lambda:region:account_id:function:<fn_name>1012            arn = ":".join(arn.split(":")[0:-1])1013        return self._arns.get(arn, None)1014    def get_function_by_name_or_arn(1015        self, name_or_arn, qualifier=None1016    ) -> LambdaFunction:1017        fn = self.get_function_by_name(name_or_arn, qualifier) or self.get_arn(1018            name_or_arn1019        )1020        if fn is None:1021            if name_or_arn.startswith("arn:aws"):1022                arn = name_or_arn1023            else:1024                arn = make_function_arn(self.region_name, self.account_id, name_or_arn)1025            if qualifier:1026                arn = f"{arn}:{qualifier}"1027            raise UnknownFunctionException(arn)1028        return fn1029    def put_function(self, fn):1030        """1031        :param fn: Function1032        :type fn: LambdaFunction1033        """1034        valid_role = re.match(InvalidRoleFormat.pattern, fn.role)1035        if valid_role:1036            account = valid_role.group(2)1037            if account != self.account_id:1038                raise CrossAccountNotAllowed()1039            try:1040                iam_backend = iam_backends[self.account_id]["global"]1041                iam_backend.get_role_by_arn(fn.role)1042            except IAMNotFoundException:1043                raise InvalidParameterValueException(1044                    "The role defined for the function cannot be assumed by Lambda."1045                )1046        else:1047            raise InvalidRoleFormat(fn.role)1048        if fn.function_name in self._functions:1049            self._functions[fn.function_name]["latest"] = fn1050        else:1051            self._functions[fn.function_name] = {"latest": fn, "versions": []}1052        # instantiate a new policy for this version of the lambda1053        fn.policy = Policy(fn)1054        self._arns[fn.function_arn] = fn1055    def publish_function(self, name_or_arn, description=""):1056        function = self.get_function_by_name_or_arn(name_or_arn)1057        name = function.function_name1058        if name not in self._functions:1059            return None1060        if not self._functions[name]["latest"]:1061            return None1062        new_version = len(self._functions[name]["versions"]) + 11063        fn = copy.copy(self._functions[name]["latest"])1064        fn.set_version(new_version)1065        if description:1066            fn.description = description1067        self._functions[name]["versions"].append(fn)1068        self._arns[fn.function_arn] = fn1069        return fn1070    def del_function(self, name_or_arn, qualifier=None):1071        function = self.get_function_by_name_or_arn(name_or_arn, qualifier)1072        name = function.function_name1073        if not qualifier:1074            # Something is still reffing this so delete all arns1075            latest = self._functions[name]["latest"].function_arn1076            del self._arns[latest]1077            for fn in self._functions[name]["versions"]:1078                del self._arns[fn.function_arn]1079            del self._functions[name]1080        elif qualifier == "$LATEST":1081            self._functions[name]["latest"] = None1082            # If theres no functions left1083            if (1084                not self._functions[name]["versions"]1085                and not self._functions[name]["latest"]1086            ):1087                del self._functions[name]1088        else:1089            fn = self.get_function_by_name(name, qualifier)1090            if fn:1091                self._functions[name]["versions"].remove(fn)1092                # If theres no functions left1093                if (1094                    not self._functions[name]["versions"]1095                    and not self._functions[name]["latest"]1096                ):1097                    del self._functions[name]1098    def all(self):1099        result = []1100        for function_group in self._functions.values():1101            latest = copy.deepcopy(function_group["latest"])1102            latest.function_arn = "{}:$LATEST".format(latest.function_arn)1103            result.append(latest)1104            result.extend(function_group["versions"])1105        return result1106    def latest(self):1107        """1108        Return the list of functions with version @LATEST1109        :return:1110        """1111        result = []1112        for function_group in self._functions.values():1113            if function_group["latest"] is not None:1114                result.append(function_group["latest"])1115        return result1116class LayerStorage(object):1117    def __init__(self):1118        self._layers = {}1119        self._arns = weakref.WeakValueDictionary()1120    def put_layer_version(self, layer_version):1121        """1122        :param layer_version: LayerVersion1123        """1124        if layer_version.name not in self._layers:1125            self._layers[layer_version.name] = Layer(layer_version)1126        self._layers[layer_version.name].attach_version(layer_version)1127    def list_layers(self):1128        return [1129            layer.to_dict() for layer in self._layers.values() if layer.layer_versions1130        ]1131    def delete_layer_version(self, layer_name, layer_version):1132        self._layers[layer_name].delete_version(layer_version)1133    def get_layer_version(self, layer_name, layer_version):1134        if layer_name not in self._layers:1135            raise UnknownLayerException()1136        for lv in self._layers[layer_name].layer_versions.values():1137            if lv.version == int(layer_version):1138                return lv1139        raise UnknownLayerException()1140    def get_layer_versions(self, layer_name):1141        if layer_name in self._layers:1142            return list(iter(self._layers[layer_name].layer_versions.values()))1143        return []1144    def get_layer_version_by_arn(self, layer_version_arn):1145        split_arn = split_layer_arn(layer_version_arn)1146        if split_arn.layer_name in self._layers:1147            return self._layers[split_arn.layer_name].layer_versions.get(1148                split_arn.version, None1149            )1150        return None1151class LambdaBackend(BaseBackend):1152    """1153    Implementation of the AWS Lambda endpoint.1154    Invoking functions is supported - they will run inside a Docker container, emulating the real AWS behaviour as closely as possible.1155    It is possible to connect from AWS Lambdas to other services, as long as you are running Moto in ServerMode.1156    The Lambda has access to environment variables `MOTO_HOST` and `MOTO_PORT`, which can be used to build the url that MotoServer runs on:1157    .. sourcecode:: python1158        def lambda_handler(event, context):1159            host = os.environ.get("MOTO_HOST")1160            port = os.environ.get("MOTO_PORT")1161            url = host + ":" + port1162            ec2 = boto3.client('ec2', region_name='us-west-2', endpoint_url=url)1163            # Or even simpler:1164            full_url = os.environ.get("MOTO_HTTP_ENDPOINT")1165            ec2 = boto3.client("ec2", region_name="eu-west-1", endpoint_url=full_url)1166            ec2.do_whatever_inside_the_existing_moto_server()1167    Moto will run on port 5000 by default. This can be overwritten by setting an environment variable when starting Moto:1168    .. sourcecode:: bash1169        # This env var will be propagated to the Docker container running the Lambda functions1170        MOTO_PORT=5000 moto_server1171    The Docker container uses the default network mode, `bridge`.1172    The following environment variables are available for fine-grained control over the Docker connection options:1173    .. sourcecode:: bash1174        # Provide the name of a custom network to connect to1175        MOTO_DOCKER_NETWORK_NAME=mycustomnetwork moto_server1176        # Override the network mode1177        # For example, network_mode=host would use the network of the host machine1178        # Note that this option will be ignored if MOTO_DOCKER_NETWORK_NAME is also set1179        MOTO_DOCKER_NETWORK_MODE=host moto_server1180    The Docker images used by Moto are taken from the `lambci/lambda`-repo by default. Use the following environment variable to configure a different repo:1181    .. sourcecode:: bash1182        MOTO_DOCKER_LAMBDA_IMAGE=mLupin/docker-lambda1183    .. note:: When using the decorators, a Docker container cannot reach Moto, as it does not run as a server. Any boto3-invocations used within your Lambda will try to connect to AWS.1184    """1185    def __init__(self, region_name, account_id):1186        super().__init__(region_name, account_id)1187        self._lambdas = LambdaStorage(region_name=region_name, account_id=account_id)1188        self._event_source_mappings = {}1189        self._layers = LayerStorage()1190    @staticmethod1191    def default_vpc_endpoint_service(service_region, zones):1192        """Default VPC endpoint service."""1193        return BaseBackend.default_vpc_endpoint_service_factory(1194            service_region, zones, "lambda"1195        )1196    def create_alias(1197        self, name, function_name, function_version, description, routing_config1198    ):1199        return self._lambdas.put_alias(1200            name, function_name, function_version, description, routing_config1201        )1202    def delete_alias(self, name, function_name):1203        return self._lambdas.delete_alias(name, function_name)1204    def get_alias(self, name, function_name):1205        return self._lambdas.get_alias(name, function_name)1206    def update_alias(1207        self, name, function_name, function_version, description, routing_config1208    ):1209        """1210        The RevisionId parameter is not yet implemented1211        """1212        return self._lambdas.update_alias(1213            name, function_name, function_version, description, routing_config1214        )1215    def create_function(self, spec):1216        function_name = spec.get("FunctionName", None)1217        if function_name is None:1218            raise RESTError("InvalidParameterValueException", "Missing FunctionName")1219        fn = LambdaFunction(1220            account_id=self.account_id,1221            spec=spec,1222            region=self.region_name,1223            version="$LATEST",1224        )1225        self._lambdas.put_function(fn)1226        if spec.get("Publish"):1227            ver = self.publish_function(function_name)1228            fn = copy.deepcopy(1229                fn1230            )  # We don't want to change the actual version - just the return value1231            fn.version = ver.version1232        return fn1233    def create_function_url_config(self, name_or_arn, config):1234        """1235        The Qualifier-parameter is not yet implemented.1236        Function URLs are not yet mocked, so invoking them will fail1237        """1238        function = self._lambdas.get_function_by_name_or_arn(name_or_arn)1239        return function.create_url_config(config)1240    def delete_function_url_config(self, name_or_arn):1241        """1242        The Qualifier-parameter is not yet implemented1243        """1244        function = self._lambdas.get_function_by_name_or_arn(name_or_arn)1245        function.delete_url_config()1246    def get_function_url_config(self, name_or_arn):1247        """1248        The Qualifier-parameter is not yet implemented1249        """1250        function = self._lambdas.get_function_by_name_or_arn(name_or_arn)1251        if not function:1252            raise UnknownFunctionException(arn=name_or_arn)1253        return function.get_url_config()1254    def update_function_url_config(self, name_or_arn, config):1255        """1256        The Qualifier-parameter is not yet implemented1257        """1258        function = self._lambdas.get_function_by_name_or_arn(name_or_arn)1259        return function.update_url_config(config)...

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.