Method _handle_error

pymongo/asynchronous/topology.py:861–929

(self, address: _Address, err_ctx: _ErrorContext)

        return _is_stale_error_topology_version(cur_tv, error_tv)

    async def _handle_error(self, address: _Address, err_ctx: _ErrorContext) -> None:
        if self._is_stale_error(address, err_ctx):
            return

        server = self._servers[address]
        error = err_ctx.error
        service_id = err_ctx.service_id

        # Ignore a handshake error if the server is behind a load balancer but
        # the service ID is unknown. This indicates that the error happened
        # when dialing the connection or during the MongoDB  handshake, so we
        # don't know the service ID to use for clearing the pool.
        if self._settings.load_balanced and not service_id and not err_ctx.completed_handshake:
            return

        if isinstance(error, NetworkTimeout) and err_ctx.completed_handshake:
            # The socket has been closed. Don't reset the server.
            # Server Discovery And Monitoring Spec: "When an application
            # operation fails because of any network error besides a socket
            # timeout...."
            return
        elif isinstance(error, WriteError):
            # Ignore writeErrors.
            return
        elif isinstance(error, (NotPrimaryError, OperationFailure)):
            # As per the SDAM spec if:
            #   - the server sees a "not primary" error, and
            #   - the server is not shutting down, and
            #   - the server version is >= 4.2, then
            # we keep the existing connection pool, but mark the server type
            # as Unknown and request an immediate check of the server.
            # Otherwise, we clear the connection pool, mark the server as
            # Unknown and request an immediate check of the server.
            if hasattr(error, "code"):
                err_code = error.code
            else:
                # Default error code if one does not exist.
                default = 10107 if isinstance(error, NotPrimaryError) else None
                err_code = error.details.get("code", default)  # type: ignore[union-attr]
            if err_code in helpers_shared._NOT_PRIMARY_CODES:
                is_shutting_down = err_code in helpers_shared._SHUTDOWN_CODES
                # Mark server Unknown, clear the pool, and request check.
                if not self._settings.load_balanced:
                    await self._process_change(ServerDescription(address, error=error))
                if is_shutting_down or (err_ctx.max_wire_version <= 7):
                    # Clear the pool.
                    await server.reset(service_id)
                server.request_check()
            elif not err_ctx.completed_handshake:
                # Unknown command error during the connection handshake.
                if not self._settings.load_balanced:
                    await self._process_change(ServerDescription(address, error=error))
                # Clear the pool.
                await server.reset(service_id)
        elif isinstance(error, ConnectionFailure):
            if isinstance(error, WaitQueueTimeoutError) or (
                error.has_error_label("SystemOverloadedError")
            ):