Method connection

lib/sqlalchemy/orm/session.py:2054–2100

r"""Return a :class:`_engine.Connection` object corresponding to this :class:`.Session` object's transactional state. Either the :class:`_engine.Connection` corresponding to the current transaction is returned, or if no transaction is in progress, a new one is begun

(
        self,
        bind_arguments: Optional[_BindArguments] = None,
        execution_options: Optional[CoreExecuteOptionsParameter] = None,
    )

        trans.prepare()

    def connection(
        self,
        bind_arguments: Optional[_BindArguments] = None,
        execution_options: Optional[CoreExecuteOptionsParameter] = None,
    ) -> Connection:
        r"""Return a :class:`_engine.Connection` object corresponding to this
        :class:`.Session` object's transactional state.

        Either the :class:`_engine.Connection` corresponding to the current
        transaction is returned, or if no transaction is in progress, a new
        one is begun and the :class:`_engine.Connection`
        returned (note that no
        transactional state is established with the DBAPI until the first
        SQL statement is emitted).

        Ambiguity in multi-bind or unbound :class:`.Session` objects can be
        resolved through any of the optional keyword arguments.   This
        ultimately makes usage of the :meth:`.get_bind` method for resolution.

        :param bind_arguments: dictionary of bind arguments.  May include
         "mapper", "bind", "clause", other custom arguments that are passed
         to :meth:`.Session.get_bind`.

        :param execution_options: a dictionary of execution options that will
         be passed to :meth:`_engine.Connection.execution_options`, **when the
         connection is first procured only**.   If the connection is already
         present within the :class:`.Session`, a warning is emitted and
         the arguments are ignored.

         .. seealso::

            :ref:`session_transaction_isolation`

        """

        if bind_arguments:
            bind = bind_arguments.pop("bind", None)

            if bind is None:
                bind = self.get_bind(**bind_arguments)
        else:
            bind = self.get_bind()

        return self._connection_for_bind(
            bind,
            execution_options=execution_options,
        )

    def _connection_for_bind(
        self,