Method get

tortoise/connection.py:155–193

Return the connection object for the given alias, creating it if needed. If the connection's event loop has changed (e.g., in a test with a new event loop), the connection is transparently replaced with a fresh one and a :class:`TortoiseLoopSwitchWarning<tortoise.wa

(self, conn_alias: str)

        return connection

    def get(self, conn_alias: str) -> BaseDBAsyncClient:
        """
        Return the connection object for the given alias, creating it if needed.

        If the connection's event loop has changed (e.g., in a test with a new event loop),
        the connection is transparently replaced with a fresh one and a
        :class:`TortoiseLoopSwitchWarning<tortoise.warnings.TortoiseLoopSwitchWarning>`
        is emitted.

        Used for accessing the low-level connection object
        (:class:`BaseDBAsyncClient<tortoise.backends.base.client.BaseDBAsyncClient>`) for the
        given alias.

        :param conn_alias: The alias for which the connection has to be fetched

        :raises ConfigurationError: If the connection alias does not exist.
        """
        storage = self._get_storage()
        try:
            conn = storage[conn_alias]
            if not conn._check_loop():
                from tortoise.warnings import TortoiseLoopSwitchWarning

                warnings.warn(
                    f"Tortoise connection '{conn_alias}' was created on a different "
                    f"event loop and will be reconnected. If this is expected (e.g., "
                    f"in tests), use tortoise_test_context() or suppress with: "
                    f"warnings.filterwarnings('ignore', "
                    f"category=TortoiseLoopSwitchWarning)",
                    TortoiseLoopSwitchWarning,
                    stacklevel=2,
                )
                conn = self._create_connection(conn_alias)
                storage[conn_alias] = conn
            return conn
        except KeyError:
            connection: BaseDBAsyncClient = self._create_connection(conn_alias)
            storage[conn_alias] = connection
            return connection

    def set(self, conn_alias: str, conn_obj: BaseDBAsyncClient) -> ConnectionToken:
        """