def create()

in aidial_adapter_bedrock/adapter_deployments.py [0:0]
39 lines of code
11 McCabe index (conditional complexity)

    def create(cls, *, compat_mapping: Dict[str, str]) -> "AdapterDeployments":

        chat_completions = {e.value for e in ChatCompletionDeployment}
        embeddings = {e.value for e in EmbeddingsDeployment}

        for deployment_id, supported_id in compat_mapping.items():
            if deployment_id in chat_completions or deployment_id in embeddings:
                log.warning(
                    f"{deployment_id!r} is one of the Bedrock deployments supported by the adapter already. "
                    f"Remove {deployment_id!r} from the compatibility mapping to avoid the warning."
                )

                if (
                    deployment_id in chat_completions
                    and supported_id in embeddings
                ):
                    raise ValueError(
                        f"The chat completion deployment {deployment_id!r} is mapped onto the embeddings deployment {supported_id!r}"
                    )

                if (
                    deployment_id in embeddings
                    and supported_id in chat_completions
                ):
                    raise ValueError(
                        f"The embeddings deployment {deployment_id!r} is mapped onto the chat completion deployment {supported_id!r}"
                    )

        compat_mapping, chat_completions = _create_deployments(
            compat_mapping,
            ChatCompletionDeployment,
            redirects=CHAT_COMPLETION_REDIRECTS,
        )
        compat_mapping, embeddings = _create_deployments(
            compat_mapping, EmbeddingsDeployment
        )

        if compat_mapping:
            raise ValueError(
                f"None of the values in the following compatibility mapping corresponds to a Bedrock deployment supported by the adapter: {json.dumps(compat_mapping)}. "
                f"Remap the deployments to the supported Bedrock deployments to fix the error."
            )

        ret = cls(chat_completions=chat_completions, embeddings=embeddings)

        log.debug(f"Adapter deployments: {ret.json()}")

        return ret