def _purge_closed_sockets()

in google/cloud/sql/connector/monitored_cache.py [0:0]

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

    def _purge_closed_sockets(self) -> None:
        """Remove closed sockets from monitored cache.

        If a socket is closed by the database driver we should remove it from
        list of sockets.
        """
        open_sockets = []
        for socket in self.sockets:
            # Check fileno for if socket is closed. Will return
            # -1 on failure, which will be used to signal socket closed.
            if socket.fileno() != -1:
                open_sockets.append(socket)
        self.sockets = open_sockets