def _create_connection()

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

• 38 lines of code
• 8 McCabe index (conditional complexity)

    def _create_connection(self):
        try:
            self.alias = ""
            # Check if the connection already exists
            for x in connections.list_connections():
                addr = connections.get_connection_addr(x[0])
                if (
                    x[1]
                    and ("address" in addr)
                    and (addr["address"] == "{}:{}".format(MILVUS_HOST, MILVUS_PORT))
                ):
                    self.alias = x[0]
                    logger.info(
                        "Reuse connection to Milvus server '{}:{}' with alias '{:s}'".format(
                            MILVUS_HOST, MILVUS_PORT, self.alias
                        )
                    )
                    break

            # Connect to the Milvus instance using the passed in Environment variables
            if len(self.alias) == 0:
                self.alias = uuid4().hex
                connections.connect(
                    alias=self.alias,
                    host=MILVUS_HOST,
                    port=MILVUS_PORT,
                    user=MILVUS_USER,  # type: ignore
                    password=MILVUS_PASSWORD,  # type: ignore
                    secure=MILVUS_USE_SECURITY,
                )
                logger.info(
                    "Create connection to Milvus server '{}:{}' with alias '{:s}'".format(
                        MILVUS_HOST, MILVUS_PORT, self.alias
                    )
                )
        except Exception as e:
            logger.error(
                "Failed to create connection to Milvus server '{}:{}', error: {}".format(
                    MILVUS_HOST, MILVUS_PORT, e
                )
            )