Method execute_async

gunicorn/dirty/client.py:223–277

Execute an action on a dirty app (async/non-blocking). Args: app_path: Import path of the dirty app action: Action to call on the app *args: Positional arguments **kwargs: Keyword arguments Returns: Result from th

(self, app_path, action, *args, **kwargs)

            ) from e

    async def execute_async(self, app_path, action, *args, **kwargs):
        """
        Execute an action on a dirty app (async/non-blocking).

        Args:
            app_path: Import path of the dirty app
            action: Action to call on the app
            *args: Positional arguments
            **kwargs: Keyword arguments

        Returns:
            Result from the dirty app action

        Raises:
            DirtyConnectionError: If connection fails
            DirtyTimeoutError: If operation times out
            DirtyError: If execution fails
        """
        # Ensure connected
        if self._writer is None:
            await self.connect_async()

        # 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
            await DirtyProtocol.write_message_async(self._writer, request)

            # Receive response with timeout
            response = await asyncio.wait_for(
                DirtyProtocol.read_message_async(self._reader),
                timeout=self.timeout
            )

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

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