Function _get_connection

tortoise/transactions.py:18–30

(connection_name: str | None)

def _get_connection(connection_name: str | None) -> BaseDBAsyncClient:
    conn_handler = get_connections()
    if connection_name:
        connection = conn_handler.get(connection_name)
    elif len(conn_handler.db_config) == 1:
        connection_name = next(iter(conn_handler.db_config.keys()))
        connection = conn_handler.get(connection_name)
    else:
        raise ParamsError(
            "You are running with multiple databases, so you should specify"
            f" connection_name: {list(conn_handler.db_config)}"
        )
    return connection


def in_transaction(connection_name: str | None = None) -> TransactionContext: