def handle_error(e: sa.engine.ExceptionContext):
            if isinstance(e.original_exception, asynch.errors.UnexpectedPacketFromServerError):
                raise sqlalchemy.exc.InvalidatePoolError(
                    "`asynch.errors.UnexpectedPacketFromServerError`, invalidate pool",
                )
        sqlalchemy.event.listen(self.engine.sync_engine, "handle_error", handle_error)

        try:
            return self.do_ping(dbapi_connection)
        except self.loaded_dbapi.Error as err:
            ...

    class Error(Exception):
        pass

import asyncio
import unittest.mock

import asynch.errors
import clickhouse_sqlalchemy.drivers.asynch.connector
import sqlalchemy as sa
import sqlalchemy.event
import sqlalchemy.exc
import sqlalchemy.ext.asyncio as saea
import logging

DSN = "clickhouse+asynch://localhost:9000"

# NOTE This is a hack to fix this issue:
# clickhouse_sqlalchemy.drivers.asynch.connector.AsyncAdapt_asynch_dbapi.Error = (
#     asynch.errors.ClickHouseException
# )  # type: ignore


def failed_when_select_one(self, operation, params=None, context=None):
    if operation == "SELECT 1":
        raise asynch.errors.UnexpectedPacketFromServerError("Unexpected packet")
    else:
        return self.await_(self._execute_async(operation, params, context))


@unittest.mock.patch.object(
    clickhouse_sqlalchemy.drivers.asynch.connector.AsyncAdapt_asynch_cursor,
    "execute",
    failed_when_select_one,
)
async def main():
    engine = saea.create_async_engine(DSN, echo=True,pool_pre_ping=True)

    @sqlalchemy.event.listens_for(engine.sync_engine, "handle_error")
    def handle_error(e: sa.engine.ExceptionContext):
        if isinstance(e.original_exception, asynch.errors.UnexpectedPacketFromServerError):
            raise sqlalchemy.exc.InvalidatePoolError(
                "`asynch.errors.UnexpectedPacketFromServerError`, invalidate pool",
            )

    for _ in range(3):
        async with engine.connect() as conn:
            await conn.execute(sa.text("SELECT 123"))


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO)
    logging.getLogger("sqlalchemy.pool").setLevel(logging.DEBUG)

    asyncio.run(main())