quicopt.client

client

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

Encode a model (quicopt.wire), POST it, read the result back. Standard library only (urllib): sending adds no dependency, like the rest of the package. The request body is the encoded model; 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; it is cached at $XDG_CACHE_HOME/quicopt/free_key and replayed as Authorization: Bearer on every later call — including from a later process, so one caller keeps one key rather than minting a fresh one per run.

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.

Example

Maximize 3x + 5y subject to x + 2y ≤ 5, with 0 ≤ x ≤ 4 and y binary:

import pulp
from quicopt import Client

prob = pulp.LpProblem("mix", pulp.LpMaximize)
x = pulp.LpVariable("x", lowBound=0, upBound=4)
y = pulp.LpVariable("y", cat="Binary")
prob += 3 * x + 5 * y
prob += x + 2 * y <= 5

result = Client().solve(prob)                 # read, encode, POST, wait
print(result.status, result.objective)        # optimal 14.0
print(result.solution)                        # {'x': 3.0, 'y': 1.0}
print(result.model_class)                     # milp
print(result.display)                         # the service's own summary

The model is passed as it stands — Pyomo, MathOpt or PuLP; the matching importer is chosen by type, so nothing here changes with the library, and a :class:~quicopt.ir.Program or already-encoded bytes are accepted too. No solver runs locally and none need be installed. Which kind of model it was is the service's finding, reported back as model_class, not something to declare.

For a long solve, :meth:Client.submit returns immediately with a :class:Job and :meth:Job.result waits on it:

job = Client().submit(prob, project="pricing-study")
print(job.status()["status"])                 # queued
result = job.result(timeout=600)              # poll until done

KEY_PATH_ENV module-attribute

KEY_PATH_ENV = 'QUICOPT_KEY_PATH'

Environment variable overriding where the free key is cached. Point it at durable storage in an environment whose home directory does not survive the run (CI, containers, Colab), where the default location is wiped between sessions and every run would otherwise mint a fresh key.

DEFAULT_BASE_URL module-attribute

DEFAULT_BASE_URL = (
    "https://try.quicoptapi.pgi.fz-juelich.de"
)

The public Quicopt free-tier endpoint a :class:Client targets when no base_url is given. Mirrors the Julia client's DEFAULT_BASE_URL so both clients reach the same server out of the box.

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 where the outcome leaves them undefined — a model with nothing to feasibility-check, or a run that found no solution. display is the framed, ready-to-print summary the service renders the same way for every kind of model.

model_class property

model_class: Optional[str]

The kind of model the service recognised (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 = DEFAULT_BASE_URL,
    api_key: Optional[str] = None,
    *,
    timeout: float = 60.0,
    key_path: Optional[Union[str, Path]] = None,
    cache: bool = True
)

A connection to a Quicopt service at base_url — the public free tier (:data:DEFAULT_BASE_URL) unless another URL is given. Holds the API key: pass a known one, or let the first keyless call mint one — which is then cached on disk and reused by later runs, so one caller keeps one key.

Bind a client to a service endpoint.

Parameters:

Name	Type	Description	Default
base_url	str	The service base URL; a trailing slash is stripped. Defaults to :data:DEFAULT_BASE_URL, the public free-tier endpoint; pass another URL to target a different server.	DEFAULT_BASE_URL
api_key	Optional[str]	A known API key, or None to reuse the cached free key (minting one on the first keyless call). A key passed here is used as-is and never written to the cache — so authenticating with a specific key cannot clobber the free key of the user running you.	None
timeout	float	Per-request socket timeout, in seconds.	60.0
key_path	Optional[Union[str, Path]]	Where to cache the free key. Defaults to :func:_default_key_path.	None
cache	bool	Set False to keep the key in memory only, neither reading nor writing the cache file.	True

Source code in src/quicopt/client.py

def __init__(self, base_url: str = DEFAULT_BASE_URL, api_key: Optional[str] = None,
             *, timeout: float = 60.0, key_path: Optional[Union[str, Path]] = None,
             cache: bool = True):
    """Bind a client to a service endpoint.

    Args:
        base_url: The service base URL; a trailing slash is stripped. Defaults
            to :data:`DEFAULT_BASE_URL`, the public free-tier endpoint; pass
            another URL to target a different server.
        api_key: A known API key, or ``None`` to reuse the cached free key
            (minting one on the first keyless call). A key passed here is used
            as-is and never written to the cache — so authenticating with a
            specific key cannot clobber the free key of the user running you.
        timeout: Per-request socket timeout, in seconds.
        key_path: Where to cache the free key. Defaults to
            :func:`_default_key_path`.
        cache: Set ``False`` to keep the key in memory only, neither reading nor
            writing the cache file.
    """
    self.base_url = base_url.rstrip("/")
    self.timeout = timeout
    self.key_path = Path(key_path).expanduser() if key_path is not None else _default_key_path()
    self.cache = cache
    self._explicit = api_key is not None
    self.api_key = api_key if self._explicit else (_read_key(self.key_path) if cache else None)
    # Only a key that came off disk may be discarded and re-minted on a 401 —
    # see `_open`, which relies on this to bound re-minting to one per run.
    self._from_cache = self.api_key is not None and not self._explicit

solve

solve(
    model: _Model,
    *,
    project: Optional[str] = None,
    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 model from a modeling library (Pyomo, OR-Tools MathOpt, PuLP) — read and encoded here — or a Program / already-encoded bytes if you built them yourself.	required
project	Optional[str]	Optional project tag for the call, so calls on one key can be invoiced per project. Sent as a query param, not baked into the model.	None
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, *, project: Optional[str] = None,
          config: Optional[Dict[str, Any]] = None, gzip: bool = False) -> Result:
    """Solve ``model`` synchronously, blocking until the result returns.

    Args:
        model: A model from a modeling library (Pyomo, OR-Tools MathOpt, PuLP)
            — read and encoded here — or a ``Program`` / already-encoded bytes
            if you built them yourself.
        project: Optional project tag for the call, so calls on one key can be
            invoiced per project. Sent as a query param, not baked into the model.
        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=_meta_config(model, project, config), gzip=gzip))

submit

submit(
    model: _Model,
    *,
    project: Optional[str] = None,
    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/PuLP model, a Program, or encoded bytes.	required
project	Optional[str]	Optional project tag for the call (per-project invoicing), sent as a query param, not baked into the model.	None
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, *, project: Optional[str] = None,
           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/PuLP model, a ``Program``, or encoded bytes.
        project: Optional project tag for the call (per-project invoicing), sent
            as a query param, not baked into the model.
        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=_meta_config(model, project, config), gzip=gzip)
    self._remember(body.get("api_key"))     # /v1/jobs echoes a minted key in the 202 body too
    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}")