[docs]    def create_hnsw_extension(self) -> None:
        try:
            with Session(self._conn) as session:
                statement = sqlalchemy.text("CREATE EXTENSION IF NOT EXISTS embedding")
                session.execute(statement)
                session.commit()
        except Exception as e:
            self.logger.exception(e)
[docs]    def create_tables_if_not_exists(self) -> None:
        with self._conn.begin():
            Base.metadata.create_all(self._conn)
[docs]    def drop_tables(self) -> None:
        with self._conn.begin():
            Base.metadata.drop_all(self._conn)
[docs]    def create_collection(self) -> None:
        if self.pre_delete_collection:
            self.delete_collection()
        with Session(self._conn) as session:
            CollectionStore.get_or_create(
                session, self.collection_name, cmetadata=self.collection_metadata
            )
[docs]    def create_hnsw_index(
        self,
        max_elements: int = 10000,
        dims: int = ADA_TOKEN_COUNT,
        m: int = 8,
        ef_construction: int = 16,
        ef_search: int = 16,
    ) -> None:
        create_index_query = sqlalchemy.text(
            "CREATE INDEX IF NOT EXISTS langchain_pg_embedding_idx "
            "ON langchain_pg_embedding USING hnsw (embedding) "
            "WITH ("
            "maxelements = {}, "
            "dims = {}, "
            "m = {}, "
            "efconstruction = {}, "
            "efsearch = {}"
            ");".format(max_elements, dims, m, ef_construction, ef_search)
        )
        # Execute the queries
        try:
            with Session(self._conn) as session: