if self._connection:
            self._connection.close()
[docs]    def create_table_if_not_exists(self) -> None:
        """
        Helper function: create table if not exists
        """
        from psycopg2 import sql
        cursor = self._connection.cursor()
        cursor.execute(
            sql.SQL(
                "CREATE TABLE IF NOT EXISTS {} ( \
            id UUID, \
            embedding_id INTEGER, \
            text VARCHAR(60000), \
            metadata VARCHAR(1024), \
            embedding FLOAT)"
            ).format(sql.Identifier(self._table))
        )
        self._connection.commit()
        cursor.close()
[docs]    def drop(self, table: str) -> None:
        """
        Helper function: Drop data
        """
        from psycopg2 import sql
        cursor = self._connection.cursor()
        cursor.execute(sql.SQL("DROP TABLE IF EXISTS {}").format(sql.Identifier(table)))
        self._connection.commit()
        cursor.close()
[docs]    def check_database_utf8(self) -> bool:
        """
        Helper function: Test the database is UTF-8 encoded
        """
        cursor = self._connection.cursor()
        query = "SELECT pg_encoding_to_char(encoding) \
        FROM pg_database \
        WHERE datname = current_database();"
        cursor.execute(query)
        encoding = cursor.fetchone()[0]
        cursor.close()
        if encoding.lower() == "utf8" or encoding.lower() == "utf-8":
            return True
        else:
            raise Exception(
                f"Database \
           '{self.connection_string.split('/')[-1]}' encoding is not UTF-8"
            )
[docs]    def add_texts(
        self,
        texts: Iterable[str],