def __get_or_create_container()

in libraries/botbuilder-azure/botbuilder/azure/cosmosdb_partitioned_storage.py [0:0]

• 34 lines of code
• 7 McCabe index (conditional complexity)

    def __get_or_create_container(self):
        with self.__lock:
            container_def = {
                "id": self.config.container_id,
                "partitionKey": {
                    "paths": ["/id"],
                    "kind": documents.PartitionKind.Hash,
                },
            }
            try:
                if not self.container:
                    self.container = self.client.CreateContainer(
                        "dbs/" + self.database["id"],
                        container_def,
                        {"offerThroughput": self.config.container_throughput},
                    )
            except cosmos_errors.HTTPFailure as err:
                if err.status_code == http_constants.StatusCodes.CONFLICT:
                    self.container = self.client.ReadContainer(
                        "dbs/" + self.database["id"] + "/colls/" + container_def["id"]
                    )
                    if "partitionKey" not in self.container:
                        self.compatability_mode_partition_key = True
                    else:
                        paths = self.container["partitionKey"]["paths"]
                        if "/partitionKey" in paths:
                            self.compatability_mode_partition_key = True
                        elif "/id" not in paths:
                            raise Exception(
                                f"Custom Partition Key Paths are not supported. {self.config.container_id} "
                                "has a custom Partition Key Path of {paths[0]}."
                            )

                else:
                    raise err