quicopt.client

client

quicopt.client — talk to the Quicopt service over HTTP.

Encode a model to wire bytes (quicopt.wire), POST it, read the result back. Stdlib-only (urllib): the transport adds no dependency, like the rest of the core. The request body is the versioned wire bytes; the response is the service's result JSON (status / objective / solution / a ready-to-print display / …). The first keyless call mints an API key, returned in the X-Quicopt-Api-Key response header and remembered on the :class:Client so later calls replay it as Authorization: Bearer.

Two entry points mirror the two service endpoints:

• :meth:Client.solve — POST /v1/solve, block for the result (synchronous).
• :meth:Client.submit — POST /v1/jobs, return a :class:Job to poll.

A non-2xx response raises :class:QuicoptError, carrying the service's stable reason code and the framed display text.

Result dataclass

Result(
    job_id: str,
    status: str,
    objective: Optional[float],
    feasible: Optional[bool],
    solution: Dict[str, float],
    solve_time_seconds: float,
    solver_data: Dict[str, Any],
    display: str,
)

A finished solve, parsed from the service's result JSON. objective and feasible are None when the class or outcome leaves them undefined (e.g. an unconstrained heuristic, or no incumbent). display is the framed, ready-to-print summary the service renders for every backend alike.

model_class property

model_class: Optional[str]

The class the service routed the model to (LP/MILP/QUBO/…).

Returns:

Type	Description
Optional[str]	The model_class recorded in solver_data, or None if the
Optional[str]	service did not report one.

QuicoptError

QuicoptError(status_code: int, body: Dict[str, Any])

Bases: Exception

A non-2xx service response. reason is the service's stable snake_case code (size_exceeded, unsupported_model, quota_exhausted, …), display the framed message to print, status_code the HTTP status.

Source code in src/quicopt/client.py

def __init__(self, status_code: int, body: Dict[str, Any]):
    self.status_code = status_code
    self.body = body if isinstance(body, dict) else {"error": str(body)}
    self.reason = self.body.get("reason")
    self.display = self.body.get("display")
    super().__init__(self.body.get("error") or f"HTTP {status_code}")

Client

Client(
    base_url: str,
    api_key: Optional[str] = None,
    *,
    timeout: float = 60.0
)

A connection to a Quicopt service at base_url. Holds the API key: pass a known one, or let the first keyless call mint one (then read it back from client.api_key to persist).

Bind a client to a service endpoint.

Parameters:

Name	Type	Description	Default
base_url	str	The service base URL; a trailing slash is stripped.	required
api_key	Optional[str]	A known API key, or None to let the first keyless call mint one (afterwards readable back from self.api_key).	None
timeout	float	Per-request socket timeout, in seconds.	60.0

Source code in src/quicopt/client.py

def __init__(self, base_url: str, api_key: Optional[str] = None, *, timeout: float = 60.0):
    """Bind a client to a service endpoint.

    Args:
        base_url: The service base URL; a trailing slash is stripped.
        api_key: A known API key, or ``None`` to let the first keyless call mint
            one (afterwards readable back from ``self.api_key``).
        timeout: Per-request socket timeout, in seconds.
    """
    self.base_url = base_url.rstrip("/")
    self.api_key = api_key
    self.timeout = timeout

solve

solve(
    model: _Model,
    *,
    config: Optional[Dict[str, Any]] = None,
    gzip: bool = False
) -> Result

Solve model synchronously, blocking until the result returns.

Parameters:

Name	Type	Description	Default
model	_Model	A front-end model (Pyomo, OR-Tools MathOpt) — imported to the wire IR here — or a Program / pre-encoded wire bytes if you built them yourself.	required
config	Optional[Dict[str, Any]]	Optional service parameters, sent as the query string.	None
gzip	bool	If True, gzip-compress the request body.	False

Returns:

Name	Type	Description
Result	Result	The finished solve.

Raises:

Type	Description
QuicoptError	On a non-2xx response.

Source code in src/quicopt/client.py

def solve(self, model: _Model, *, config: Optional[Dict[str, Any]] = None,
          gzip: bool = False) -> Result:
    """Solve ``model`` synchronously, blocking until the result returns.

    Args:
        model: A front-end model (Pyomo, OR-Tools MathOpt) — imported to the
            wire IR here — or a ``Program`` / pre-encoded wire bytes if you built
            them yourself.
        config: Optional service parameters, sent as the query string.
        gzip: If ``True``, gzip-compress the request body.

    Returns:
        Result: The finished solve.

    Raises:
        QuicoptError: On a non-2xx response.
    """
    return Result._from_json(
        self._request("POST", "/v1/solve", _to_wire(model), config=config, gzip=gzip))

submit

submit(
    model: _Model,
    *,
    config: Optional[Dict[str, Any]] = None,
    gzip: bool = False
) -> "Job"

Submit model for asynchronous solving and return a handle to poll.

Parameters:

Name	Type	Description	Default
model	_Model	A Pyomo/MathOpt model, a Program, or pre-encoded wire bytes.	required
config	Optional[Dict[str, Any]]	Optional service parameters, sent as the query string.	None
gzip	bool	If True, gzip-compress the request body.	False

Returns:

Name	Type	Description
Job	'Job'	A handle to the queued job; call :meth:Job.result to await it.

Raises:

Type	Description
QuicoptError	On a non-2xx response.

Source code in src/quicopt/client.py

def submit(self, model: _Model, *, config: Optional[Dict[str, Any]] = None,
           gzip: bool = False) -> "Job":
    """Submit ``model`` for asynchronous solving and return a handle to poll.

    Args:
        model: A Pyomo/MathOpt model, a ``Program``, or pre-encoded wire bytes.
        config: Optional service parameters, sent as the query string.
        gzip: If ``True``, gzip-compress the request body.

    Returns:
        Job: A handle to the queued job; call :meth:`Job.result` to await it.

    Raises:
        QuicoptError: On a non-2xx response.
    """
    body = self._request("POST", "/v1/jobs", _to_wire(model), config=config, gzip=gzip)
    return Job(self, body["job_id"])

Job dataclass

Job(client: Client, job_id: str)

A handle to an async job. :meth:result polls until it finishes.

status

status() -> Dict[str, Any]

Fetch the job's metadata and framed log_tail.

Returns:

Name	Type	Description
dict	Dict[str, Any]	The job state (queued/running/done/failed) and its
	Dict[str, Any]	log tail, as returned by the service.

Source code in src/quicopt/client.py

def status(self) -> Dict[str, Any]:
    """Fetch the job's metadata and framed ``log_tail``.

    Returns:
        dict: The job state (``queued``/``running``/``done``/``failed``) and its
        log tail, as returned by the service.
    """
    return self.client._request("GET", f"/v1/jobs/{self.job_id}")

result

result(
    *,
    wait: bool = True,
    timeout: float = 120.0,
    poll: float = 0.5
) -> Result

Fetch the job's result, optionally polling until it is ready.

Parameters:

Name	Type	Description	Default
wait	bool	If True, poll past the service's not_done reason until the worker finishes; if False, fetch once.	True
timeout	float	Maximum time to poll, in seconds, before giving up.	120.0
poll	float	Delay between polls, in seconds.	0.5

Returns:

Name	Type	Description
Result	Result	The finished solve.

Raises:

Type	Description
QuicoptError	If wait is False and the job is not yet done, on any non-not_done error, or once timeout elapses.

Source code in src/quicopt/client.py

def result(self, *, wait: bool = True, timeout: float = 120.0, poll: float = 0.5) -> Result:
    """Fetch the job's result, optionally polling until it is ready.

    Args:
        wait: If ``True``, poll past the service's ``not_done`` reason until the
            worker finishes; if ``False``, fetch once.
        timeout: Maximum time to poll, in seconds, before giving up.
        poll: Delay between polls, in seconds.

    Returns:
        Result: The finished solve.

    Raises:
        QuicoptError: If ``wait`` is ``False`` and the job is not yet done, on
            any non-``not_done`` error, or once ``timeout`` elapses.
    """
    deadline = time.monotonic() + timeout
    while True:
        try:
            return Result._from_json(
                self.client._request("GET", f"/v1/jobs/{self.job_id}/result"))
        except QuicoptError as e:
            if not wait or e.reason != "not_done" or time.monotonic() > deadline:
                raise
            time.sleep(poll)

log

log() -> str

Fetch the job's plain-text log.

Returns:

Name	Type	Description
str	str	The framed log view on success, or the error text on failure.

Source code in src/quicopt/client.py

def log(self) -> str:
    """Fetch the job's plain-text log.

    Returns:
        str: The framed log view on success, or the error text on failure.
    """
    return self.client._open("GET", f"/v1/jobs/{self.job_id}/log").decode("utf-8", "replace")

delete

delete() -> None

Delete the job and its stored blob/result/log on the server.

Returns:

Type	Description
None	None.

Source code in src/quicopt/client.py

def delete(self) -> None:
    """Delete the job and its stored blob/result/log on the server.

    Returns:
        None.
    """
    self.client._open("DELETE", f"/v1/jobs/{self.job_id}")