def _create_collection()

in datastore/providers/milvus_datastore.py [0:0]

• 35 lines of code
• 11 McCabe index (conditional complexity)

    def _create_collection(self, collection_name, create_new: bool) -> None:
        """Create a collection based on environment and passed in variables.

        Args:
            create_new (bool): Whether to overwrite if collection already exists.
        """
        try:
            self._schema_ver = "V1"
            # If the collection exists and create_new is True, drop the existing collection
            if utility.has_collection(collection_name, using=self.alias) and create_new:
                utility.drop_collection(collection_name, using=self.alias)

            # Check if the collection doesnt exist
            if utility.has_collection(collection_name, using=self.alias) is False:
                # If it doesnt exist use the field params from init to create a new schem
                schema = [field[1] for field in SCHEMA_V2]
                schema = CollectionSchema(schema)
                # Use the schema to create a new collection
                self.col = Collection(
                    collection_name,
                    schema=schema,
                    using=self.alias,
                    consistency_level=self._consistency_level,
                )
                self._schema_ver = "V2"
                logger.info(
                    "Create Milvus collection '{}' with schema {} and consistency level {}".format(
                        collection_name, self._schema_ver, self._consistency_level
                    )
                )
            else:
                # If the collection exists, point to it
                self.col = Collection(collection_name, using=self.alias)  # type: ignore
                # Which sechma is used
                for field in self.col.schema.fields:
                    if field.name == "id" and field.is_primary:
                        self._schema_ver = "V2"
                        break
                logger.info(
                    "Milvus collection '{}' already exists with schema {}".format(
                        collection_name, self._schema_ver
                    )
                )
        except Exception as e:
            logger.error(
                "Failed to create collection '{}', error: {}".format(collection_name, e)
            )