Coverage for netbox_data_import/inference_adapter.py: 95%
121 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-09 20:50 +0000
1# SPDX-License-Identifier: Apache-2.0
2# SPDX-FileCopyrightText: 2026 Marcin Zieba <marcinpsk@gmail.com>
3"""The OpenAI-compatible Inference Backend adapter (specification 8.1, 8.3, 8.4).
5One non-streaming Chat Completions call, built entirely from configuration. The adapter knows
6nothing about NetBox, candidates, proposals or jobs: it takes an InferenceRequest and returns an
7InferenceCompletion, or raises a typed backend error.
9A returned completion means the call reached the backend and the envelope parsed. Deciding what the
10content means belongs to the application service, not here.
11"""
13import json
15from dataclasses import dataclass
16from collections.abc import Sequence
18import requests
20from .inference_trust import (
21 InvalidInferenceConfiguration,
22 assert_resolved_address_allowed,
23 resolve_addresses,
24 validate_api_root,
25)
27CHAT_COMPLETIONS_PATH = "/chat/completions"
29# Section 8.4 rejects every other terminal reason: only a completed answer is a completion.
30ACCEPTED_FINISH_REASONS = ("stop",)
33class InferenceBackendError(Exception):
34 """A typed Inference Backend failure. No message carries the API key."""
36 category = "backend_error"
39class TransportFailure(InferenceBackendError):
40 """The backend could not be reached, or answered with a server error."""
42 category = "transport_failure"
45class BackendTimeout(InferenceBackendError):
46 """The backend did not answer inside the configured limits."""
48 category = "timeout"
51class AuthenticationFailure(InferenceBackendError):
52 """The backend refused the credential."""
54 category = "authentication_failure"
57class RateLimited(InferenceBackendError):
58 """The backend refused the call for rate reasons."""
60 category = "rate_limit"
62 def __init__(self, message, retry_after=None):
63 super().__init__(message)
64 self.retry_after = retry_after
67class InvalidBackendConfiguration(InferenceBackendError):
68 """The backend configuration cannot be used as given."""
70 category = "invalid_configuration"
73class MalformedEnvelope(InferenceBackendError):
74 """The backend answered, but not with one usable completion."""
76 category = "malformed_envelope"
79@dataclass(frozen=True)
80class InferenceRequest:
81 """What the application asks of a backend."""
83 system_instruction: str
84 user_payload_json: str
85 requested_response_mode: str
88@dataclass(frozen=True)
89class InferenceCompletion:
90 """What one completed backend call returned."""
92 content_text: str | None
93 is_refusal: bool
94 finish_reason: str
95 backend_request_id: str | None
96 backend_response_id: str | None
97 backend_model: str | None
100def _retry_after(response) -> int | None:
101 """Return the Retry-After seconds a rate-limited answer states, when it states one."""
102 raw = response.headers.get("Retry-After")
103 try:
104 return int(raw) if raw is not None else None
105 except ValueError:
106 return None
109class OpenAICompatibleAdapter:
110 """Call one OpenAI-compatible Chat Completions endpoint, without streaming or tools."""
112 def __init__(
113 self,
114 api_root: str,
115 model: str,
116 allowlist: Sequence[str],
117 authentication: str = "bearer",
118 response_mode: str = "prompt_json",
119 connect_timeout: int = 5,
120 read_timeout: int = 60,
121 session: "requests.Session | None" = None,
122 ):
123 # Verbatim: normalizing here would pass the request-time recheck a value the form refuses.
124 self.api_root = api_root
125 self.model = model
126 self.allowlist = tuple(allowlist)
127 self.authentication = authentication
128 self.response_mode = response_mode
129 self.connect_timeout = connect_timeout
130 self.read_timeout = read_timeout
131 self._session = session or requests.Session()
133 def _check_response_mode(self, request: InferenceRequest) -> None:
134 """Reject a request for a mode this backend is not configured to serve."""
135 if request.requested_response_mode != self.response_mode:
136 raise InvalidBackendConfiguration(
137 f"This backend is configured for '{self.response_mode}', "
138 f"but the request asked for '{request.requested_response_mode}'."
139 )
140 if self.response_mode == "json_schema":
141 raise InvalidBackendConfiguration(
142 "The json_schema response mode needs a schema to send, and this delivery stores none."
143 )
145 def _body(self, request: InferenceRequest) -> dict:
146 """Return the Chat Completions body: one choice, no streaming, no tools."""
147 body = {
148 "model": self.model,
149 "n": 1,
150 "stream": False,
151 "messages": [
152 {"role": "system", "content": request.system_instruction},
153 {"role": "user", "content": request.user_payload_json},
154 ],
155 }
156 if self.response_mode == "json_object":
157 body["response_format"] = {"type": "json_object"}
158 return body
160 def _check_destination(self) -> None:
161 """Reject a destination the deployment has not approved, rechecked at request time."""
162 try:
163 validate_api_root(self.api_root, self.allowlist, self.authentication)
164 assert_resolved_address_allowed(self.api_root, self.allowlist, resolve_addresses(self.api_root))
165 except InvalidInferenceConfiguration as exc:
166 raise InvalidBackendConfiguration(str(exc)) from None
168 def complete(self, request: InferenceRequest, api_key: str) -> InferenceCompletion:
169 """Return one completion, or raise the typed error the backend condition maps to."""
170 self._check_response_mode(request)
171 self._check_destination()
172 headers = {"Content-Type": "application/json", "Accept": "application/json"}
173 if self.authentication == "bearer":
174 headers["Authorization"] = f"Bearer {api_key}"
175 try:
176 response = self._session.post(
177 f"{self.api_root}{CHAT_COMPLETIONS_PATH}",
178 json=self._body(request),
179 headers=headers,
180 timeout=(self.connect_timeout, self.read_timeout),
181 # A redirect is a different destination, so it is refused rather than followed.
182 allow_redirects=False,
183 )
184 except requests.Timeout:
185 raise BackendTimeout("The backend did not answer inside the configured limits.") from None
186 except requests.RequestException as exc:
187 raise TransportFailure(f"The backend could not be reached ({type(exc).__name__}).") from None
188 return self._read(response)
190 def _read(self, response) -> InferenceCompletion:
191 """Classify the answer, then parse the one envelope a completed call returns."""
192 if response.status_code in (301, 302, 303, 307, 308):
193 raise InvalidBackendConfiguration(
194 f"The backend redirected the call (HTTP {response.status_code}), which is not followed."
195 )
196 if response.status_code in (401, 403):
197 raise AuthenticationFailure(f"The backend refused the credential (HTTP {response.status_code}).")
198 if response.status_code == 429:
199 raise RateLimited("The backend rate limited the call.", retry_after=_retry_after(response))
200 if response.status_code >= 400:
201 raise TransportFailure(f"The backend answered HTTP {response.status_code}.")
202 try:
203 envelope = response.json()
204 except ValueError:
205 raise MalformedEnvelope("The backend answered with a body that is not JSON.") from None
206 return self._completion(envelope)
208 @staticmethod
209 def _completion(envelope) -> InferenceCompletion:
210 """Return the completion one envelope describes, rejecting anything else."""
211 if not isinstance(envelope, dict):
212 raise MalformedEnvelope("The backend answered with no completion object.")
213 choices = envelope.get("choices")
214 if not isinstance(choices, list) or len(choices) != 1:
215 raise MalformedEnvelope("A completion carries exactly one choice.")
216 choice = choices[0]
217 if not isinstance(choice, dict):
218 raise MalformedEnvelope("The backend answered with an unreadable choice.")
219 finish_reason = choice.get("finish_reason")
220 if finish_reason not in ACCEPTED_FINISH_REASONS:
221 raise MalformedEnvelope(f"The backend stopped for reason '{finish_reason}', so no answer was produced.")
222 message = choice.get("message")
223 if not isinstance(message, dict):
224 raise MalformedEnvelope("The backend answered with an unreadable message.")
225 content = message.get("content")
226 refusal = message.get("refusal")
227 # Content parts are common on OpenAI-compatible servers, and this delivery reads text only.
228 if content is not None and not isinstance(content, str):
229 raise MalformedEnvelope("The backend answered with content this delivery cannot read as text.")
230 # A refusal is a completed call that produced no answer, not a failure to call.
231 is_refusal = bool(refusal) or not (content or "").strip()
232 return InferenceCompletion(
233 content_text=content,
234 is_refusal=is_refusal,
235 finish_reason=finish_reason,
236 backend_request_id=envelope.get("request_id"),
237 backend_response_id=envelope.get("id"),
238 backend_model=envelope.get("model"),
239 )
242def encode_payload(value) -> str:
243 """Return the compact JSON one user payload travels as."""
244 return json.dumps(value, separators=(",", ":"), sort_keys=True)
247__all__ = (
248 "CHAT_COMPLETIONS_PATH",
249 "AuthenticationFailure",
250 "BackendTimeout",
251 "InferenceBackendError",
252 "InferenceCompletion",
253 "InferenceRequest",
254 "InvalidBackendConfiguration",
255 "MalformedEnvelope",
256 "OpenAICompatibleAdapter",
257 "RateLimited",
258 "TransportFailure",
259 "encode_payload",
260)