Retrieve a session by ID from the pool. This method first ensures the session pool is at its maximum capacity. It then tries to retrieve a specific session by ID. If the session is not found or not usable, `None` is returned. Args: session_id: The ID of the sess
(self, session_id: str)
| 198 | |
| 199 | @ensure_context |
| 200 | async def get_session_by_id(self, session_id: str) -> Session | None: |
| 201 | """Retrieve a session by ID from the pool. |
| 202 | |
| 203 | This method first ensures the session pool is at its maximum capacity. It then tries to retrieve a specific |
| 204 | session by ID. If the session is not found or not usable, `None` is returned. |
| 205 | |
| 206 | Args: |
| 207 | session_id: The ID of the session to retrieve. |
| 208 | |
| 209 | Returns: |
| 210 | The session object if found and usable, otherwise `None`. |
| 211 | """ |
| 212 | await self._fill_sessions_to_max() |
| 213 | session = self._state.current_value.sessions.get(session_id) |
| 214 | |
| 215 | if not session: |
| 216 | logger.warning(f'Session with ID {session_id} not found.') |
| 217 | return None |
| 218 | |
| 219 | if not session.is_usable: |
| 220 | logger.warning(f'Session with ID {session_id} is not usable.') |
| 221 | return None |
| 222 | |
| 223 | return session |
| 224 | |
| 225 | async def reset_store(self) -> None: |
| 226 | """Reset the KVS where the pool state is persisted.""" |