Method _tmp_session

pymongo/synchronous/mongo_client.py:2292–2324

If provided session is None, lend a temporary session.

(
        self, session: Optional[client_session.ClientSession]
    )

    @contextlib.contextmanager
    def _tmp_session(
        self, session: Optional[client_session.ClientSession]
    ) -> Generator[Optional[client_session.ClientSession], None]:
        """If provided session is None, lend a temporary session."""
        if session is not None and not isinstance(session, client_session.ClientSession):
            raise ValueError(
                f"'session' argument must be a ClientSession or None, not {type(session)}"
            )

        # Check for a bound session. If one exists, treat it as an explicitly passed session.
        session = session or self._get_bound_session()
        if session:
            # Don't call end_session.
            yield session
            return

        s = self._ensure_session(session)
        if s:
            try:
                yield s
            except Exception as exc:
                if isinstance(exc, ConnectionFailure):
                    s._server_session.mark_dirty()

                # Always call end_session on error.
                s.end_session()
                raise
            finally:
                # Call end_session when we exit this scope.
                if not s._attached_to_cursor:
                    s.end_session()
        else:
            yield None

    def _process_response(self, reply: Mapping[str, Any], session: Optional[ClientSession]) -> None:
        self._topology.receive_cluster_time(reply.get("$clusterTime"))