def delete_expired_sessions()

in src/backend/entrypoints/llm_backend/infrastructure/session.py [0:0]

• 18 lines of code
• 7 McCabe index (conditional complexity)

    def delete_expired_sessions(self, optimization_storage_url: str) -> None:
        deleted_count = 0
        for key in self._redis_client.scan_iter(f"{self.BASE_HASH_KEY}:*"):
            user_sessions = self._redis_client.hgetall(key)
            for serialized_session_id, serialized_session in user_sessions.items():
                session = self._deserialize_session(serialized_session)
                if session.is_expired():
                    studies_names = session.get("optimization_studies_names", [])
                    for study_name in studies_names:
                        try:
                            delete_optimization_study(optimization_storage_url, study_name)
                        except KeyError:
                            # Study doesn't exist, perhaps it was revoked or already deleted
                            pass

                    email = self._deserialize_bytes(key).split(":")[1]
                    session_id = self._deserialize_bytes(serialized_session_id)
                    self.delete(email, session_id)
                    deleted_count += 1
        logger.info("Deleted %s expired sessions", deleted_count)