Class ControlClient

gunicorn/ctl/client.py:24–111

Client for connecting to gunicorn control socket. Can be used as a context manager: with ControlClient('/path/to/gunicorn.ctl') as client: result = client.send_command('show workers')

class ControlClient:
    """
    Client for connecting to gunicorn control socket.

    Can be used as a context manager:

        with ControlClient('/path/to/gunicorn.ctl') as client:
            result = client.send_command('show workers')
    """

    def __init__(self, socket_path: str, timeout: float = 30.0):
        """
        Initialize control client.

        Args:
            socket_path: Path to the Unix socket
            timeout: Socket timeout in seconds (default 30)
        """
        self.socket_path = socket_path
        self.timeout = timeout
        self._sock = None
        self._request_id = 0

    def connect(self):
        """
        Connect to control socket.

        Raises:
            ControlClientError: If connection fails
        """
        if self._sock:
            return

        try:
            self._sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
            self._sock.settimeout(self.timeout)
            self._sock.connect(self.socket_path)
        except socket.error as e:
            self._sock = None
            raise ControlClientError(f"Failed to connect to {self.socket_path}: {e}")

    def close(self):
        """Close connection."""
        if self._sock:
            try:
                self._sock.close()
            except Exception:
                pass
            self._sock = None

    def send_command(self, command: str, args: list = None) -> dict:
        """
        Send command and wait for response.

        Args:
            command: Command string (e.g., "show workers")
            args: Optional additional arguments