async def vacuum_all_tables(
        self,
        analyze: bool = False,
        full: bool = False,
    ):
        """Vacuum all tables in the database"""

        vacuum_query = "VACUUM"
        if analyze:
            vacuum_query += " ANALYZE"
        if full:
            logger.warning(
                "VACUUM FULL not implemented yet. Running plain VACUUM instead."
            )
        try:
            await self.connection_manager.execute_query(vacuum_query)
        except Exception as e:
            logger.error(f"Error vacuuming all tables: {str(e)}")
            raise e
Vacuum all tables in the database
py/core/providers/database/maintenance.py