How to use _version_to_opensearch method in localstack

Best Python code snippet using localstack_python

provider.py

Source:provider.py

...66    VersionString,67)68from localstack.utils.analytics import event_publisher69from localstack.utils.aws import aws_stack70def _version_to_opensearch(71    version: Optional[ElasticsearchVersionString],72) -> Optional[VersionString]:73    if version is not None:74        if version.startswith("OpenSearch_"):75            return version76        else:77            return f"Elasticsearch_{version}"78def _version_from_opensearch(79    version: Optional[VersionString],80) -> Optional[ElasticsearchVersionString]:81    if version is not None:82        if version.startswith("Elasticsearch_"):83            return version.split("_")[1]84        else:85            return version86def _instancetype_to_opensearch(instance_type: Optional[str]) -> Optional[str]:87    if instance_type is not None:88        return instance_type.replace("elasticsearch", "search")89def _instancetype_from_opensearch(instance_type: Optional[str]) -> Optional[str]:90    if instance_type is not None:91        return instance_type.replace("search", "elasticsearch")92def _clusterconfig_from_opensearch(93    cluster_config: Optional[ClusterConfig],94) -> Optional[ElasticsearchClusterConfig]:95    if cluster_config is not None:96        # Just take the whole typed dict and typecast it to our target type97        result = cast(ElasticsearchClusterConfig, cluster_config)98        # Adjust the instance type names99        result["InstanceType"] = _instancetype_from_opensearch(cluster_config.get("InstanceType"))100        result["DedicatedMasterType"] = _instancetype_from_opensearch(101            cluster_config.get("DedicatedMasterType")102        )103        result["WarmType"] = _instancetype_from_opensearch(cluster_config.get("WarmType"))104        return result105def _domainstatus_from_opensearch(106    domain_status: Optional[DomainStatus],107) -> Optional[ElasticsearchDomainStatus]:108    if domain_status is not None:109        # Just take the whole typed dict and typecast it to our target type110        result = cast(ElasticsearchDomainStatus, domain_status)111        # Only specifically handle keys which are named differently or their values differ (version and clusterconfig)112        result["ElasticsearchVersion"] = _version_from_opensearch(113            domain_status.get("EngineVersion")114        )115        result["ElasticsearchClusterConfig"] = _clusterconfig_from_opensearch(116            domain_status.get("ClusterConfig")117        )118        result.pop("EngineVersion", None)119        result.pop("ClusterConfig", None)120        return result121def _clusterconfig_to_opensearch(122    elasticsearch_cluster_config: Optional[ElasticsearchClusterConfig],123) -> Optional[ClusterConfig]:124    if elasticsearch_cluster_config is not None:125        result = cast(ClusterConfig, elasticsearch_cluster_config)126        result["InstanceType"] = _instancetype_to_opensearch(result.get("InstanceType"))127        result["DedicatedMasterType"] = _instancetype_to_opensearch(128            result.get("DedicatedMasterType")129        )130        result["WarmType"] = _instancetype_to_opensearch(result.get("WarmType"))131        return result132def _domainconfig_from_opensearch(133    domain_config: Optional[DomainConfig],134) -> Optional[ElasticsearchDomainConfig]:135    if domain_config is not None:136        result = cast(ElasticsearchDomainConfig, domain_config)137        engine_version = domain_config.get("EngineVersion", {})138        result["ElasticsearchVersion"] = ElasticsearchVersionStatus(139            Options=_version_from_opensearch(engine_version.get("Options")),140            Status=cast(OptionStatus, engine_version.get("Status")),141        )142        cluster_config = domain_config.get("ClusterConfig", {})143        result["ElasticsearchClusterConfig"] = ElasticsearchClusterConfigStatus(144            Options=_clusterconfig_from_opensearch(cluster_config.get("Options")),145            Status=cluster_config.get("Status"),146        )147        result.pop("EngineVersion", None)148        result.pop("ClusterConfig", None)149        return result150def _compatible_version_list_from_opensearch(151    compatible_version_list: Optional[CompatibleVersionsList],152) -> Optional[CompatibleElasticsearchVersionsList]:153    if compatible_version_list is not None:154        return [155            CompatibleVersionsMap(156                SourceVersion=_version_from_opensearch(version_map["SourceVersion"]),157                TargetVersions=[158                    _version_from_opensearch(target_version)159                    for target_version in version_map["TargetVersions"]160                ],161            )162            for version_map in compatible_version_list163        ]164@contextmanager165def exception_mapper():166    """Maps an exception thrown by the OpenSearch client to an exception thrown by the ElasticSearch API."""167    try:168        yield169    except ClientError as err:170        exception_types = {171            "AccessDeniedException": AccessDeniedException,172            "BaseException": EsBaseException,173            "ConflictException": ConflictException,174            "DisabledOperationException": DisabledOperationException,175            "InternalException": InternalException,176            "InvalidPaginationTokenException": InvalidPaginationTokenException,177            "InvalidTypeException": InvalidTypeException,178            "LimitExceededException": LimitExceededException,179            "ResourceAlreadyExistsException": ResourceAlreadyExistsException,180            "ResourceNotFoundException": ResourceNotFoundException,181            "ValidationException": ValidationException,182        }183        mapped_exception_type = exception_types.get(err.response["Error"]["Code"], EsBaseException)184        raise mapped_exception_type(err.response["Error"]["Message"])185class EsProvider(EsApi):186    def create_elasticsearch_domain(187        self,188        context: RequestContext,189        domain_name: DomainName,190        elasticsearch_version: ElasticsearchVersionString = None,191        elasticsearch_cluster_config: ElasticsearchClusterConfig = None,192        ebs_options: EBSOptions = None,193        access_policies: PolicyDocument = None,194        snapshot_options: SnapshotOptions = None,195        vpc_options: VPCOptions = None,196        cognito_options: CognitoOptions = None,197        encryption_at_rest_options: EncryptionAtRestOptions = None,198        node_to_node_encryption_options: NodeToNodeEncryptionOptions = None,199        advanced_options: AdvancedOptions = None,200        log_publishing_options: LogPublishingOptions = None,201        domain_endpoint_options: DomainEndpointOptions = None,202        advanced_security_options: AdvancedSecurityOptionsInput = None,203        auto_tune_options: AutoTuneOptionsInput = None,204        tag_list: TagList = None,205    ) -> CreateElasticsearchDomainResponse:206        opensearch_client = aws_stack.connect_to_service("opensearch", region_name=context.region)207        # If no version is given, we set our default elasticsearch version208        engine_version = (209            _version_to_opensearch(elasticsearch_version)210            if elasticsearch_version211            else constants.ELASTICSEARCH_DEFAULT_VERSION212        )213        kwargs = {214            "DomainName": domain_name,215            "EngineVersion": engine_version,216            "ClusterConfig": _clusterconfig_to_opensearch(elasticsearch_cluster_config),217            "EBSOptions": ebs_options,218            "AccessPolicies": access_policies,219            "SnapshotOptions": snapshot_options,220            "VPCOptions": vpc_options,221            "CognitoOptions": cognito_options,222            "EncryptionAtRestOptions": encryption_at_rest_options,223            "NodeToNodeEncryptionOptions": node_to_node_encryption_options,...

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.