How can I connect to a new leader when my database leader switch happens?

[trikker]

Hi, all, my question and background are as below:

Background:

My backend database is MySQ group replication with group_replication_single_primary_mode = ON, which means only single node(we call it leader or primary node) is allowed to write data at a specific time, but any node can become the new leader as time goes by and the old leader may be in the crashed state for some time.

Question:

Because the leader node may change, I need to create the async engine after I detect the leader switch. However, I don't want the frontend user(here it is FastAPI restful client) to receive the "Lost connection to MySQL server during query" exception. So when leader switch happens I want to:

• create a new async engine, this is easy
• get a new connection from the new created async engine and use the new connection to do db operations

However, even I add the parameter pool_pre_ping=True to function create_async_engine, I fail to get a connection from the new created async engine in the first request after the leader switch, it seems the connection will be recreated automatically and try to connect the old leader but the old leader is already down. The error is sqlalchemy.exc.OperationalError: (pymysql.err.OperationalError) (2003, "Can't connect to MySQL server on '192.168.0.1'"), the detailed error stack is at the very end after the code.

Hope can get some help, thanks very much!

Below is my code:

• database.py

import time
from typing import AsyncGenerator
from sqlalchemy import text, event, ExceptionContext
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncEngine
from sqlalchemy.orm import declarative_base

from app.utils.config import app_config

from .log import logger
from .mysql_mgr import MySQLMGR

SQL_TEST_CONNECTION = "SELECT * FROM alembic_version"


class AsyncDBUtil:
    """
    sqlalchemy async db util
    """
    def __init__(
            self,
            engine: str,
            driver: str,
            dialect: str,
            user: str,
            encrypted_password: str,
            host: str,
            port: int = 3306,
            database: str = None,
            pool_size: int = 10,
            max_overflow: int = 10,
            pool_pre_ping: bool = True
    ):
        """
        init function, will init the parameters

        Args:
            engine: db engine, mysql, mysql_mgr and so on
            driver: db driver
            dialect: db dialect
            user: db user
            password: db password
            host: db host
            port: db port
            database: db database
            pool_size: db connection pool size
            max_overflow: db connection pool max overflow
            pool_pre_ping: whether to check the connection before use
                - if set to True, will check the connection before use, if it is not alive, will reconnect it
                - if set to False, will not check the connection before use
        """
        assert engine.lower() in ("mysql", "mysql_mgr", "oceanbase", "adb")
        self.__engine = engine
        self.__driver = driver
        self.__dialect = dialect
        self.__user = user
        self.__password = password
        self.__host = host
        self.__host_list = host.replace(' ', '').split(',')
        self.__port = port
        self.__database = database
        self.__pool_size = pool_size
        self.__max_overflow = max_overflow
        self.__pool_pre_ping = pool_pre_ping
        # current leader host in mysql_mgr
        self.__mysql_mgr_leader_host_and_port: tuple[str, int] = (None, None)
        self.__db_sqlalchemy_database_url: str | None = None
        self.__async_engine: AsyncEngine | None = None
        self.__async_session_factory: async_sessionmaker | None = None
        if engine.lower() == "mysql_mgr":
            assert self.__host_list, "mysql_mgr host list cannot be empty"
            self.__mysql_mgr = MySQLMGR(
                server_list=self.__host_list,
                user=self.__user,
                password=self.__password
            )
        else:
            assert len(self.__host_list) == 1, "only one host is allowed if engine is not mysql_mgr"

        # init db finally
        self.init()

    def init(self) -> None:
        """
        init async engine, async session and register mysql_mgr handle_error
        """
        host = None
        port = None
        if self.__engine.lower() == "mysql_mgr":
            while host is None:
                self.__mysql_mgr_leader_host_and_port = (host, port) = self.__mysql_mgr.get_leader_host_and_port()
        else:
            host, port = self.__host, self.__port
        logger.info("engine is %s, db host, port to be connected is %s:%d", self.__engine, host, port)

        self.__db_sqlalchemy_database_url = "{driver}+{dialect}://{user}:{password}@{host}:{port}/{database}".format(
            driver=self.__driver,
            dialect=self.__dialect,
            user=self.__user,
            password=self.__password,
            host=host,
            port=port,
            database=self.__database
        )

        # reinit method will call init() and recreate async engine, the old async engine should be garbage collected
        self.__async_engine = create_async_engine(
            self.__db_sqlalchemy_database_url,
            pool_size=self.__pool_size,
            max_overflow=self.__max_overflow,
            pool_pre_ping=self.__pool_pre_ping,
            echo=False
        )
        self.__async_session_factory = async_sessionmaker(
            autocommit=False,
            autoflush=False,
            bind=self.__async_engine
        )
        # only for mysql_mgr
        self.__register_handle_error()
        logger.info("init db completed")

    def reinit(self) -> None:
        """
        reinit async engine, used when mysql_mgr leader changed
        """
        # @TODO: old connections in async_session are not really closed here,
        #        but it is ok because they will be never used.
        self.init()

    def get_mysql_mgr_leader_host_and_port(self) -> tuple[str, int] | tuple[None, None]:
        """
        get mysql_mgr leader host and port

        Returns:
            mysql_mgr leader host and port
        """
        return self.__mysql_mgr_leader_host_and_port

    @property
    def db_url(self) -> str | None:
        """
        get db url, like mysql+pymysql://root:123456@127.0.0.1:3306/test

        Returns:
            db url
        """
        return self.__db_sqlalchemy_database_url

    @property
    def async_engine(self) -> AsyncEngine:
        """
        get async engine

        Returns:
            async engine of db
        """
        return self.__async_engine

    @property
    def async_session(self) -> async_sessionmaker:
        """
        get async session

        Returns:
            async session of db
        """
        return self.__async_session_factory

    async def get_db(self) -> AsyncGenerator:
        """
        get async session

        Yields:
            async session of db
        """
        async with self.async_session() as session:
            yield session

    async def test_connection(self) -> None:
        """
        async test connection
        """
        async with self.async_session() as session:
            await session.execute(text(SQL_TEST_CONNECTION))

    def sync_test_connection(self) -> None:
        """
        sync test connection
        """
        with self.__async_engine.sync_engine.connect() as session:
            session.execute(text(SQL_TEST_CONNECTION))

    async def close_connection(self) -> None:
        """
        close connection
        """
        await self.async_engine.dispose()

    def close_sync_connection(self) -> None:
        """
        close sync connection
        """
        self.__async_engine.sync_engine.dispose()

    def __receive_handle_error(self, exception_context: ExceptionContext) -> None:
        """
        listen for the 'handle_error' event, used for mysql_mgr

        Args:
            exception_context: exception context of the error
        """
        # ERROR 2003 (HY000): Can't connect to MySQL server on 'xxx.xxx.xxx.xxx'
        # ERROR 2013 (HY000): Lost connection to MySQL server during query
        # ERROR 3208 (HY000): The consensus follower is not allowed to do current operation.
        error_no = exception_context.original_exception.args[0]
        logger.warning("mysql_mgr handle_error, error no is %d", error_no)
        if error_no in (2003, 2013, 3208):
            if self.__engine.lower() == "mysql_mgr":
                if self.get_mysql_mgr_leader_host_and_port() == self.__mysql_mgr.get_leader_host_and_port():
                    logger.warning("mysql_mgr current host is leader, no need to reinit db connection object.")
                else:
                    logger.warning("start to reinit db connection object.")
                    time.sleep(1)
                    self.reinit()
                    logger.warning("reinit db connection object completed.")
                    self.sync_test_connection()
                    logger.info("sync test connection successfully.")

    def __register_handle_error(self) -> None:
        """
        register mysql_mgr handle_error
        """
        if self.__engine.lower() == "mysql_mgr":
            assert not event.contains(
                self.async_engine.sync_engine,
                "handle_error",
                self.__receive_handle_error
            )
            event.listen(
                self.async_engine.sync_engine,
                "handle_error",
                self.__receive_handle_error
            )
            assert event.contains(
                self.async_engine.sync_engine,
                "handle_error",
                self.__receive_handle_error
            )


# initialize the metadb
metadb = AsyncDBUtil(**app_config["meta_db_config"])

Base = declarative_base()

• app_config["meta_db_config"] contents from the yaml file:

  meta_db_config:
    driver: mysql
    engine: mysql_mgr
    dialect: aiomysql
    host: 192.168.1.100:3306,192.168.1.101:3306,192.168.1.102:3306
    user: u_test
    password: xxxxxx
    database: testdb
    pool_size: 10
    max_overflow: 10

• How to use the async engine in my code?

from app.utils.database import metadb

async def my_function(...):
    async with metadb.async_session() as session:
        await session.scalars(select(XXXModel).where(...)

• error stack of the first request after leader switch happens:

ERROR:    Exception in ASGI application
Traceback (most recent call last):
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/aiomysql/connection.py", line 527, in _connect
    self._reader, self._writer = await \
                                 ^^^^^^^
  File "/opt/homebrew/Cellar/python@3.12/3.12.4/Frameworks/Python.framework/Versions/3.12/lib/python3.12/asyncio/tasks.py", line 520, in wait_for
    return await fut
           ^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/aiomysql/connection.py", line 88, in _open_connection
    transport, _ = await loop.create_connection(
                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "uvloop/loop.pyx", line 2039, in create_connection
  File "uvloop/loop.pyx", line 2016, in uvloop.loop.Loop.create_connection
ConnectionRefusedError: [Errno 61] Connection refused
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 146, in __init__
    self._dbapi_connection = engine.raw_connection()
                             ^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 3300, in raw_connection
    return self.pool.connect()
           ^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/pool/base.py", line 449, in connect
    return _ConnectionFairy._checkout(self)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/pool/base.py", line 1344, in _checkout
    with util.safe_reraise():
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/util/langhelpers.py", line 146, in __exit__
    raise exc_value.with_traceback(exc_tb)
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/pool/base.py", line 1341, in _checkout
    fairy._connection_record.get_connection()
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/pool/base.py", line 837, in get_connection
    self.__connect()
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/pool/base.py", line 900, in __connect
    with util.safe_reraise():
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/util/langhelpers.py", line 146, in __exit__
    raise exc_value.with_traceback(exc_tb)
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/pool/base.py", line 896, in __connect
    self.dbapi_connection = connection = pool._invoke_creator(self)
                                         ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/engine/create.py", line 643, in connect
    return dialect.connect(*cargs, **cparams)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/engine/default.py", line 620, in connect
    return self.loaded_dbapi.connect(*cargs, **cparams)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/dialects/mysql/aiomysql.py", line 267, in connect
    await_only(creator_fn(*arg, **kw)),
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 131, in await_only
    return current.driver.switch(awaitable)  # type: ignore[no-any-return]
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 196, in greenlet_spawn
    value = await result
            ^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/aiomysql/connection.py", line 75, in _connect
    await conn._connect()
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/aiomysql/connection.py", line 563, in _connect
    raise OperationalError(
pymysql.err.OperationalError: (2003, "Can't connect to MySQL server on '11.163.221.184'")
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/uvicorn/protocols/http/httptools_impl.py", line 399, in run_asgi
    result = await app(  # type: ignore[func-returns-value]
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/uvicorn/middleware/proxy_headers.py", line 70, in __call__
    return await self.app(scope, receive, send)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/fastapi/applications.py", line 1054, in __call__
    await super().__call__(scope, receive, send)
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/starlette/applications.py", line 123, in __call__
    await self.middleware_stack(scope, receive, send)
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/starlette/middleware/errors.py", line 186, in __call__
    raise exc
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/starlette/middleware/errors.py", line 164, in __call__
    await self.app(scope, receive, _send)
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/starlette/middleware/exceptions.py", line 65, in __call__
    await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/starlette/_exception_handler.py", line 64, in wrapped_app
    raise exc
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app
    await app(scope, receive, sender)
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/starlette/routing.py", line 756, in __call__
    await self.middleware_stack(scope, receive, send)
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/starlette/routing.py", line 776, in app
    await route.handle(scope, receive, send)
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/starlette/routing.py", line 297, in handle
    await self.app(scope, receive, send)
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/starlette/routing.py", line 77, in app
    await wrap_app_handling_exceptions(app, request)(scope, receive, send)
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/starlette/_exception_handler.py", line 64, in wrapped_app
    raise exc
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app
    await app(scope, receive, sender)
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/starlette/routing.py", line 72, in app
    response = await func(request)
               ^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/fastapi/routing.py", line 278, in app
    raw_response = await run_endpoint_function(
                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/fastapi/routing.py", line 191, in run_endpoint_function
    return await dependant.call(**values)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/app/cmdb/cluster/api.py", line 52, in read_clusters
    cluster_list = await crud.get_cluster(
                   ^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/app/cmdb/cluster/crud.py", line 52, in get_cluster
    result = await session.scalars(filters)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/ext/asyncio/session.py", line 574, in scalars
    result = await self.execute(
             ^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/ext/asyncio/session.py", line 461, in execute
    result = await greenlet_spawn(
             ^^^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 201, in greenlet_spawn
    result = context.throw(*sys.exc_info())
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/orm/session.py", line 2351, in execute
    return self._execute_internal(
           ^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/orm/session.py", line 2226, in _execute_internal
    conn = self._connection_for_bind(bind)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/orm/session.py", line 2095, in _connection_for_bind
    return trans._connection_for_bind(engine, execution_options)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "<string>", line 2, in _connection_for_bind
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/orm/state_changes.py", line 139, in _go
    ret_value = fn(self, *arg, **kw)
                ^^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/orm/session.py", line 1189, in _connection_for_bind
    conn = bind.connect()
           ^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 3276, in connect
    return self._connection_cls(self)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 148, in __init__
    Connection._handle_dbapi_exception_noconnection(
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 2440, in _handle_dbapi_exception_noconnection
    raise sqlalchemy_exception.with_traceback(exc_info[2]) from e
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 146, in __init__
    self._dbapi_connection = engine.raw_connection()
                             ^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 3300, in raw_connection
    return self.pool.connect()
           ^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/pool/base.py", line 449, in connect
    return _ConnectionFairy._checkout(self)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/pool/base.py", line 1344, in _checkout
    with util.safe_reraise():
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/util/langhelpers.py", line 146, in __exit__
    raise exc_value.with_traceback(exc_tb)
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/pool/base.py", line 1341, in _checkout
    fairy._connection_record.get_connection()
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/pool/base.py", line 837, in get_connection
    self.__connect()
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/pool/base.py", line 900, in __connect
    with util.safe_reraise():
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/util/langhelpers.py", line 146, in __exit__
    raise exc_value.with_traceback(exc_tb)
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/pool/base.py", line 896, in __connect
    self.dbapi_connection = connection = pool._invoke_creator(self)
                                         ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/engine/create.py", line 643, in connect
    return dialect.connect(*cargs, **cparams)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/engine/default.py", line 620, in connect
    return self.loaded_dbapi.connect(*cargs, **cparams)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/dialects/mysql/aiomysql.py", line 267, in connect
    await_only(creator_fn(*arg, **kw)),
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 131, in await_only
    return current.driver.switch(awaitable)  # type: ignore[no-any-return]
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 196, in greenlet_spawn
    value = await result
            ^^^^^^^^^^^^
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/aiomysql/connection.py", line 75, in _connect
    await conn._connect()
  File "/Users/test/coding/myproj/.venv/lib/python3.12/site-packages/aiomysql/connection.py", line 563, in _connect
    raise OperationalError(
sqlalchemy.exc.OperationalError: (pymysql.err.OperationalError) (2003, "Can't connect to MySQL server on '192.168.0.1'")
(Background on this error at: https://sqlalche.me/e/20/e3q8)

[zzzeek]

Because the leader node may change, I need to create the async engine after I detect the leader switch.

I would stop right there. you do not want to make a new engine, this would be extremely complicated. All of that self.__mysql_mgr.get_leader_host_and_port(), I wouldn't go there at all, this is going to be complicated and unreliable - and if you are using fallback servers, the whole point is because you're trying to have a setup that's reliable!

When you have a server setup with leader/follower and fallback, you provide your applications a single ip address using one of two tools: haproxy or proxysql. These are extremely robust and mature tools that will perform the switchover task a million times better than any client side python code can. then in your create_async_engine turn on pool_pre_ping so that when the switchover happens, the pool will automatically flush its pool of connections when a new connect detects that the existing server has switched out.

that way, you write zero code and you are using industry standard approaches. anything else is wasting time.

[trikker]

Thanks. That is a feasible way, but we don't want to introduce a new component that this not 100% under our control, and we need to find a reliable way to solve the HA problem of the proxy server(we don't want to run a proxy server on every python app machine).

Here I find a new way and it works after my tests. We use a API auth middleware to auth the api access key and its signature before actually processing a request. The auth process need to read the access key and secret key from database for every request(out QPS is not high, we will consider to use Redis to reduce the pressure of backend database). If leader switch happens, getting the access key and secret key from db will fail, and in this process the new sqlalchemy async engine will be created synchronously. Then we retry the retrival of the access key and secret key from db, it will use the new db session in the new leader server of backend database.

In a word, we use the API auth internal db request as the shield of the actual request to avoid the db session failure. Lastly, to prevent the db access error if leader switch happens during the actual process of the request(that is after the API auth process), we will retry the request in the middleware if the request method is GET if it failed, we will not retry the request if the request method is not GET because we cannot tell whether the data has been written to db or not if the leader switch happens even if the request return error, the backend database may recover the data and commit the transaction with its WAL log(e.g. redo in innodb and binlog in MySQL).

[zzzeek]

im not really following all of that, but i would say we support full flexibility in how a particular engine connects to a database, in such a way that can change over time, using new access tokens, etc. see https://docs.sqlalchemy.org/en/20/core/engines.html#generating-dynamic-authentication-tokens for an example. you can do a lot with that.

however, changing the actual engine = create_async_engine to point to another engine means that you no longer can have your engine be global and no longer set up in things like sessionmakers, so switching at that level is not something I would want to do.

I've used haproxy for many years, it's extremely robust, and I dont really see what you would need to have "under control", SQLAlchemy itself is also not "100% under your control". but as said you can swap out connectivity routines for an existing engine, I'd go with that otherwise. would not try to switch the front facing engine itself.