def _construct_api_domain()

in samtranslator/model/api/http_api_generator.py [0:0]

• 70 lines of code
• 19 McCabe index (conditional complexity)

    def _construct_api_domain(self, http_api):
        """
        Constructs and returns the ApiGateway Domain and BasepathMapping
        """
        if self.domain is None:
            return None, None, None

        if self.domain.get("DomainName") is None or self.domain.get("CertificateArn") is None:
            raise InvalidResourceException(
                self.logical_id, "Custom Domains only works if both DomainName and CertificateArn" " are provided."
            )

        self.domain["ApiDomainName"] = "{}{}".format(
            "ApiGatewayDomainNameV2", logical_id_generator.LogicalIdGenerator("", self.domain.get("DomainName")).gen()
        )

        domain = ApiGatewayV2DomainName(
            self.domain.get("ApiDomainName"), attributes=self.passthrough_resource_attributes
        )
        domain_config = dict()
        domain.DomainName = self.domain.get("DomainName")
        domain.Tags = self.tags
        endpoint = self.domain.get("EndpointConfiguration")

        if endpoint is None:
            endpoint = "REGIONAL"
            # to make sure that default is always REGIONAL
            self.domain["EndpointConfiguration"] = "REGIONAL"
        elif endpoint not in ["REGIONAL"]:
            raise InvalidResourceException(
                self.logical_id,
                "EndpointConfiguration for Custom Domains must be one of {}.".format(["REGIONAL"]),
            )
        domain_config["EndpointType"] = endpoint

        if self.domain.get("OwnershipVerificationCertificateArn", None):
            domain_config["OwnershipVerificationCertificateArn"] = self.domain.get(
                "OwnershipVerificationCertificateArn"
            )

        domain_config["CertificateArn"] = self.domain.get("CertificateArn")
        if self.domain.get("SecurityPolicy", None):
            domain_config["SecurityPolicy"] = self.domain.get("SecurityPolicy")

        domain.DomainNameConfigurations = [domain_config]

        mutual_tls_auth = self.domain.get("MutualTlsAuthentication", None)
        if mutual_tls_auth:
            if isinstance(mutual_tls_auth, dict):
                if not set(mutual_tls_auth.keys()).issubset({"TruststoreUri", "TruststoreVersion"}):
                    invalid_keys = []
                    for key in mutual_tls_auth.keys():
                        if key not in {"TruststoreUri", "TruststoreVersion"}:
                            invalid_keys.append(key)
                    invalid_keys.sort()
                    raise InvalidResourceException(
                        ",".join(invalid_keys),
                        "Available MutualTlsAuthentication fields are {}.".format(
                            ["TruststoreUri", "TruststoreVersion"]
                        ),
                    )
                domain.MutualTlsAuthentication = {}
                if mutual_tls_auth.get("TruststoreUri", None):
                    domain.MutualTlsAuthentication["TruststoreUri"] = mutual_tls_auth["TruststoreUri"]
                if mutual_tls_auth.get("TruststoreVersion", None):
                    domain.MutualTlsAuthentication["TruststoreVersion"] = mutual_tls_auth["TruststoreVersion"]
            else:
                raise InvalidResourceException(
                    mutual_tls_auth,
                    "MutualTlsAuthentication must be a map with at least one of the following fields {}.".format(
                        ["TruststoreUri", "TruststoreVersion"]
                    ),
                )

        # Create BasepathMappings
        if self.domain.get("BasePath") and isinstance(self.domain.get("BasePath"), str):
            basepaths = [self.domain.get("BasePath")]
        elif self.domain.get("BasePath") and isinstance(self.domain.get("BasePath"), list):
            basepaths = self.domain.get("BasePath")
        else:
            basepaths = None
        basepath_resource_list = self._construct_basepath_mappings(basepaths, http_api)

        # Create the Route53 RecordSetGroup resource
        record_set_group = self._construct_route53_recordsetgroup()

        return domain, basepath_resource_list, record_set_group