Send endSessions command(s) with the given session ids.
(self, session_ids: list[_ServerSession])
| 1697 | return self._server_property("server_type") == SERVER_TYPE.Mongos |
| 1698 | |
| 1699 | def _end_sessions(self, session_ids: list[_ServerSession]) -> None: |
| 1700 | """Send endSessions command(s) with the given session ids.""" |
| 1701 | try: |
| 1702 | # Use Connection.command directly to avoid implicitly creating |
| 1703 | # another session. |
| 1704 | with self._conn_for_reads( |
| 1705 | ReadPreference.PRIMARY_PREFERRED, None, operation=_Op.END_SESSIONS |
| 1706 | ) as ( |
| 1707 | conn, |
| 1708 | read_pref, |
| 1709 | ): |
| 1710 | if not conn.supports_sessions: |
| 1711 | return |
| 1712 | |
| 1713 | for i in range(0, len(session_ids), common._MAX_END_SESSIONS): |
| 1714 | spec = {"endSessions": session_ids[i : i + common._MAX_END_SESSIONS]} |
| 1715 | conn.command("admin", spec, read_preference=read_pref, client=self) |
| 1716 | except PyMongoError: |
| 1717 | # Drivers MUST ignore any errors returned by the endSessions |
| 1718 | # command. |
| 1719 | pass |
| 1720 | |
| 1721 | def close(self) -> None: |
| 1722 | """Cleanup client resources and disconnect from MongoDB. |
no test coverage detected