Method _execute_locked

gunicorn/dirty/client.py:101–136

Execute while holding the lock.

(self, app_path, action, args, kwargs)

            return self._execute_locked(app_path, action, args, kwargs)

    def _execute_locked(self, app_path, action, args, kwargs):
        """Execute while holding the lock."""
        # Ensure connected
        if self._sock is None:
            self.connect()

        # Build request
        request_id = str(uuid.uuid4())
        request = make_request(
            request_id=request_id,
            app_path=app_path,
            action=action,
            args=args,
            kwargs=kwargs
        )

        try:
            # Send request
            DirtyProtocol.write_message(self._sock, request)

            # Receive response
            response = DirtyProtocol.read_message(self._sock)

            # Handle response
            return self._handle_response(response)
        except socket.timeout:
            self._close_socket()
            raise DirtyTimeoutError(
                "Timeout waiting for dirty app response",
                timeout=self.timeout
            )
        except Exception as e:
            self._close_socket()
            if isinstance(e, DirtyError):
                raise
            raise DirtyConnectionError(f"Communication error: {e}") from e

    def stream(self, app_path, action, *args, **kwargs):
        """