def get_users_with_active_streaming_sessions()

in lambda/launcher/workstation_launcher.py [0:0]

• 11 lines of code
• 6 McCabe index (conditional complexity)

    def get_users_with_active_streaming_sessions(self, configs : List[AutoLaunchConfig]) -> List[str]:
        users_with_active_streaming_sessions : List[str] = list()
        paginator = self.nimble_client.get_paginator('list_streaming_sessions')

        studio_ids = WorkstationLauncher.get_studio_ids_from_launch_configs(configs)

        for studio_id in studio_ids:
            page_iterator = paginator.paginate(studioId=studio_id)
            for page in page_iterator:
                for session in page['sessions']:
                    if (session['state'] == 'CREATE_IN_PROGRESS' or session['state'] == 'READY'):
                        users_with_active_streaming_sessions.append(session['ownedBy'])

        return users_with_active_streaming_sessions