(
self,
bind: _SessionBind,
execution_options: Optional[CoreExecuteOptionsParameter],
)
| 1166 | (SessionTransactionState.ACTIVE,), _StateChangeStates.NO_CHANGE |
| 1167 | ) |
| 1168 | def _connection_for_bind( |
| 1169 | self, |
| 1170 | bind: _SessionBind, |
| 1171 | execution_options: Optional[CoreExecuteOptionsParameter], |
| 1172 | ) -> Connection: |
| 1173 | if bind in self._connections: |
| 1174 | if execution_options: |
| 1175 | util.warn( |
| 1176 | "Connection is already established for the " |
| 1177 | "given bind; execution_options ignored" |
| 1178 | ) |
| 1179 | return self._connections[bind][0] |
| 1180 | |
| 1181 | self._state = SessionTransactionState.PROVISIONING_CONNECTION |
| 1182 | |
| 1183 | local_connect = False |
| 1184 | should_commit = True |
| 1185 | |
| 1186 | try: |
| 1187 | if self._parent: |
| 1188 | conn = self._parent._connection_for_bind( |
| 1189 | bind, execution_options |
| 1190 | ) |
| 1191 | if not self.nested: |
| 1192 | return conn |
| 1193 | else: |
| 1194 | if isinstance(bind, engine.Connection): |
| 1195 | conn = bind |
| 1196 | if conn.engine in self._connections: |
| 1197 | raise sa_exc.InvalidRequestError( |
| 1198 | "Session already has a Connection associated " |
| 1199 | "for the given Connection's Engine" |
| 1200 | ) |
| 1201 | else: |
| 1202 | conn = bind.connect() |
| 1203 | local_connect = True |
| 1204 | |
| 1205 | try: |
| 1206 | conn_exec_opts: Dict[str, Any] = {} |
| 1207 | if self.session.execution_options: |
| 1208 | conn_exec_opts.update(self.session.execution_options) |
| 1209 | if execution_options: |
| 1210 | conn_exec_opts.update(execution_options) |
| 1211 | if conn_exec_opts: |
| 1212 | conn = conn.execution_options(**conn_exec_opts) |
| 1213 | |
| 1214 | transaction: Transaction |
| 1215 | if self.session.twophase and self._parent is None: |
| 1216 | # TODO: shouldn't we only be here if not |
| 1217 | # conn.in_transaction() ? |
| 1218 | # if twophase is set and conn.in_transaction(), validate |
| 1219 | # that it is in fact twophase. |
| 1220 | transaction = conn.begin_twophase() |
| 1221 | elif self.nested: |
| 1222 | transaction = conn.begin_nested() |
| 1223 | elif conn.in_transaction(): |
| 1224 | |
| 1225 | if local_connect: |
no test coverage detected