Coverage for netbox_data_import/inference_adapter.py: 99.17%
242 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-10-30 09:47 +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).
5The adapter makes one non-streaming Chat Completions call and can discover optional model-id
6suggestions. It knows nothing about NetBox, candidates, proposals or jobs: it takes an
7InferenceRequest and returns an InferenceCompletion, 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
17from contextlib import suppress
19import requests
20from urllib3.exceptions import ReadTimeoutError
22from .inference_transport import (
23 ResponseBodyTooLarge,
24 ResponseProcessingFailure,
25 WallClockDeadline,
26 WallClockDeadlineExceeded,
27 is_preconnect_failure,
28 request_to_resolved_address,
29)
30from .inference_settings import MODEL_MAX_LENGTH
31from .inference_trust import (
32 InvalidInferenceConfiguration,
33 assert_resolved_address_allowed,
34 resolve_addresses,
35 validate_api_root,
36)
38CHAT_COMPLETIONS_PATH = "/chat/completions"
39MODELS_PATH = "/models"
40CHAT_COMPLETION_RESPONSE_LIMIT = 65_536
41MODEL_DISCOVERY_LIMIT = 100
42MODEL_DISCOVERY_RESPONSE_LIMIT = 65_536
43DIAGNOSTIC_TEXT_LIMIT = 4096
45# Section 8.4 rejects every other terminal reason: only a completed answer is a completion.
46ACCEPTED_FINISH_REASONS = ("stop",)
48# A caller must tell an absent body from a cut-short one, an empty one and a present one.
49BODY_ABSENT = "absent"
50BODY_INTERRUPTED = "interrupted"
51BODY_EMPTY = "empty"
52BODY_PRESENT = "present"
54# Specification 13.3 enumerates the transient statuses. Every other error status is the request.
55TRANSIENT_STATUSES = (500, 502, 503, 504)
57_REDACTED = "[redacted: the backend echoed the credential]"
58_BODY_CLEAN = "clean"
59_BODY_ECHOES = "echoes"
62@dataclass(frozen=True)
63class ResponseDiagnostic:
64 """What one call received, for an operator to read when it failed.
66 `text` is retained only for an empty or unauthenticated response, bounded by
67 DIAGNOSTIC_TEXT_LIMIT, or as a fixed redaction marker. `redacted` records that the body was
68 replaced for credential safety. `withheld` records that an authenticated response body was not
69 retained. `truncated` records that retained unauthenticated text is incomplete.
70 """
72 receipt: str
73 text: str | None = None
74 status_code: int | None = None
75 redacted: bool = False
76 withheld: bool = False
77 truncated: bool = False
80ABSENT_DIAGNOSTIC = ResponseDiagnostic(receipt=BODY_ABSENT)
83def _inspect(text: str, api_key: str) -> str:
84 """Return whether the body contains a complete literal or decoded key."""
85 pending: list[object] = [text]
86 while pending:
87 value = pending.pop()
88 if isinstance(value, str):
89 if api_key in value:
90 return _BODY_ECHOES
91 with suppress(ValueError, RecursionError):
92 # Members are flattened into the list so a key echoed as a member name is seen too.
93 pending.append(
94 json.loads(value, object_pairs_hook=lambda pairs: [item for pair in pairs for item in pair])
95 )
96 elif isinstance(value, list):
97 pending.extend(value)
98 return _BODY_CLEAN
101def _diagnostic(response, api_key: str) -> ResponseDiagnostic:
102 """Return the diagnostic one answered call carries, with the credential taken out of it."""
103 try:
104 text = response.text
105 except Exception: # noqa: BLE001 - a body that cannot be decoded is a lost body, not a new failure
106 return ResponseDiagnostic(receipt=BODY_INTERRUPTED, status_code=response.status_code)
107 verdict = _inspect(text, api_key) if api_key else _BODY_CLEAN
108 if verdict == _BODY_ECHOES:
109 return ResponseDiagnostic(receipt=BODY_PRESENT, text=_REDACTED, status_code=response.status_code, redacted=True)
110 receipt = BODY_PRESENT if text else BODY_EMPTY
111 if api_key and text:
112 return ResponseDiagnostic(receipt=receipt, status_code=response.status_code, withheld=True)
113 return ResponseDiagnostic(
114 receipt=receipt,
115 text=text[:DIAGNOSTIC_TEXT_LIMIT],
116 status_code=response.status_code,
117 truncated=len(text) > DIAGNOSTIC_TEXT_LIMIT,
118 )
121class InferenceBackendError(Exception):
122 """A typed Inference Backend failure. No message carries the API key."""
124 category = "backend_error"
125 # A caller must never parse a message or an HTTP object to decide whether to try again.
126 retryable = False
128 def __init__(self, *args, diagnostic: ResponseDiagnostic | None = None):
129 super().__init__(*args)
130 self.diagnostic = diagnostic or ABSENT_DIAGNOSTIC
133class TransportFailure(InferenceBackendError):
134 """The backend could not be reached, or answered with a server error."""
136 category = "transport_failure"
137 retryable = True
140class BackendTimeout(InferenceBackendError):
141 """The backend did not answer inside the configured limits."""
143 category = "timeout"
144 retryable = True
147class AuthenticationFailure(InferenceBackendError):
148 """The backend refused the credential."""
150 category = "authentication_failure"
153class RateLimited(InferenceBackendError):
154 """The backend refused the call for rate reasons."""
156 category = "rate_limit"
157 retryable = True
159 def __init__(self, message, retry_after=None, diagnostic=None):
160 super().__init__(message, diagnostic=diagnostic)
161 self.retry_after = retry_after
164class InvalidBackendConfiguration(InferenceBackendError):
165 """The backend configuration cannot be used as given."""
167 category = "invalid_configuration"
170class MalformedEnvelope(InferenceBackendError):
171 """The backend answered, but not with one usable completion."""
173 category = "malformed_envelope"
176@dataclass(frozen=True)
177class InferenceRequest:
178 """What the application asks of a backend."""
180 system_instruction: str
181 user_payload_json: str
182 requested_response_mode: str
185@dataclass(frozen=True)
186class InferenceCompletion:
187 """What one completed backend call returned."""
189 content_text: str | None
190 is_refusal: bool
191 finish_reason: str
192 backend_request_id: str | None
193 backend_response_id: str | None
194 backend_model: str | None
195 # The adapter computes `is_refusal` while parsing. It does not retain the raw refusal body.
196 diagnostic: ResponseDiagnostic = ABSENT_DIAGNOSTIC
199def _retry_after(response) -> int | None:
200 """Return the Retry-After seconds a rate-limited answer states, when it states one."""
201 raw = response.headers.get("Retry-After")
202 try:
203 return int(raw) if raw is not None else None
204 except ValueError:
205 return None
208def _status_failure(status: int, diagnostic, retry_after: int | None) -> InferenceBackendError | None:
209 """Return the typed failure one HTTP status maps to, or None when the status carries none."""
210 if 300 <= status < 400:
211 return InvalidBackendConfiguration(
212 f"The backend redirected the call (HTTP {status}), which is not followed.", diagnostic=diagnostic
213 )
214 if status in (401, 403):
215 return AuthenticationFailure(f"The backend refused the credential (HTTP {status}).", diagnostic=diagnostic)
216 if status == 429:
217 return RateLimited("The backend rate limited the call.", retry_after=retry_after, diagnostic=diagnostic)
218 if status in TRANSIENT_STATUSES:
219 return TransportFailure(f"The backend answered HTTP {status}.", diagnostic=diagnostic)
220 if status >= 400:
221 return InvalidBackendConfiguration(
222 f"The backend rejected the request (HTTP {status}), which repeating cannot fix.",
223 diagnostic=diagnostic,
224 )
225 return None
228def _processing_failure(exc: ResponseProcessingFailure, api_key: str) -> InferenceBackendError:
229 """Return the typed failure one interrupted or refused response body maps to."""
230 diagnostic = ResponseDiagnostic(receipt=BODY_INTERRUPTED, status_code=exc.response.status_code)
231 if isinstance(exc.cause, ResponseBodyTooLarge):
232 # The body is refused unread, so the status is the only classification left.
233 return _status_failure(exc.response.status_code, diagnostic, _retry_after(exc.response)) or MalformedEnvelope(
234 "The backend response is too large.", diagnostic=diagnostic
235 )
236 if isinstance(exc.cause, ValueError):
237 return InvalidBackendConfiguration(
238 f"The backend answered with a location this delivery cannot use ({type(exc.cause).__name__}).",
239 diagnostic=_diagnostic(exc.response, api_key),
240 )
241 if isinstance(exc.cause, requests.Timeout) or _is_response_read_timeout(exc.cause):
242 return BackendTimeout(
243 "The backend did not finish its answer inside the configured limits.", diagnostic=diagnostic
244 )
245 return TransportFailure(f"The backend answer was interrupted ({type(exc.cause).__name__}).", diagnostic=diagnostic)
248def _is_response_read_timeout(exc: Exception) -> bool:
249 """Return whether Requests timed out after it had received response headers."""
250 return isinstance(exc, requests.ConnectionError) and bool(exc.args) and isinstance(exc.args[0], ReadTimeoutError)
253class OpenAICompatibleAdapter:
254 """Call one OpenAI-compatible Chat Completions endpoint, without streaming or tools."""
256 def __init__(
257 self,
258 api_root: str,
259 model: str,
260 allowlist: Sequence[str],
261 authentication: str = "bearer",
262 response_mode: str = "prompt_json",
263 connect_timeout: int = 5,
264 read_timeout: int = 60,
265 session: "requests.Session | None" = None,
266 deadline: WallClockDeadline | None = None,
267 ):
268 # Verbatim: normalizing here would pass the request-time recheck a value the form refuses.
269 self.api_root = api_root
270 self.model = model
271 self.allowlist = tuple(allowlist)
272 self.authentication = authentication
273 self.response_mode = response_mode
274 self.connect_timeout = connect_timeout
275 self.read_timeout = read_timeout
276 self._session = session or requests.Session()
277 self._deadline = deadline
279 def _check_response_mode(self, request: InferenceRequest) -> None:
280 """Reject a request for a mode this backend is not configured to serve."""
281 if request.requested_response_mode != self.response_mode:
282 raise InvalidBackendConfiguration(
283 f"This backend is configured for '{self.response_mode}', "
284 f"but the request asked for '{request.requested_response_mode}'."
285 )
286 if self.response_mode == "json_schema":
287 raise InvalidBackendConfiguration(
288 "The json_schema response mode needs a schema to send, and this delivery stores none."
289 )
291 def _body(self, request: InferenceRequest) -> dict:
292 """Return the Chat Completions body: one choice, no streaming, no tools."""
293 body = {
294 "model": self.model,
295 "n": 1,
296 "stream": False,
297 "messages": [
298 {"role": "system", "content": request.system_instruction},
299 {"role": "user", "content": request.user_payload_json},
300 ],
301 }
302 if self.response_mode == "json_object":
303 body["response_format"] = {"type": "json_object"}
304 return body
306 def _resolved_destinations(self) -> tuple[str, ...]:
307 """Return all approved addresses, rechecking the destination at request time."""
308 try:
309 validate_api_root(self.api_root, self.allowlist, self.authentication)
310 addresses = resolve_addresses(self.api_root, deadline=self._deadline)
311 assert_resolved_address_allowed(self.api_root, self.allowlist, addresses)
312 except WallClockDeadlineExceeded:
313 raise BackendTimeout("The connection test exceeded its overall time limit.") from None
314 except InvalidInferenceConfiguration as exc:
315 raise InvalidBackendConfiguration(str(exc)) from None
316 return addresses
318 def _send(
319 self,
320 method: str,
321 path: str,
322 api_key: str,
323 *,
324 response_body_limit: int,
325 **kwargs,
326 ) -> requests.Response:
327 """Send one authenticated request through the checked and pinned destination boundary."""
328 resolved_addresses = self._resolved_destinations()
329 headers = {"Accept": "application/json"}
330 if "json" in kwargs:
331 headers["Content-Type"] = "application/json"
332 if self.authentication == "bearer":
333 headers["Authorization"] = f"Bearer {api_key}"
334 connection_failure: requests.RequestException | None = None
335 for resolved_address in resolved_addresses:
336 try:
337 response = request_to_resolved_address(
338 self._session,
339 method,
340 f"{self.api_root}{path}",
341 resolved_address,
342 headers=headers,
343 timeout=(self.connect_timeout, self.read_timeout),
344 deadline=self._deadline,
345 response_body_limit=response_body_limit,
346 # A redirect is a different destination, so it is refused rather than followed.
347 allow_redirects=False,
348 **kwargs,
349 )
350 except ResponseProcessingFailure as exc:
351 raise _processing_failure(exc, api_key) from None
352 except requests.ConnectTimeout as exc:
353 connection_failure = exc
354 continue
355 except requests.ConnectionError as exc:
356 if is_preconnect_failure(exc):
357 connection_failure = exc
358 continue
359 raise TransportFailure(
360 f"The backend could not be reached ({type(exc).__name__}).",
361 diagnostic=ABSENT_DIAGNOSTIC,
362 ) from None
363 except requests.Timeout:
364 raise BackendTimeout("The backend did not answer inside the configured limits.") from None
365 except requests.RequestException as exc:
366 # A cut-short body raises with no response attached, so its bytes cannot be recovered.
367 receipt = BODY_INTERRUPTED if isinstance(exc, requests.exceptions.ChunkedEncodingError) else BODY_ABSENT
368 raise TransportFailure(
369 f"The backend could not be reached ({type(exc).__name__}).",
370 diagnostic=ResponseDiagnostic(receipt=receipt),
371 ) from None
372 except ValueError as exc:
373 # `requests` raises this bare while preparing an unparsable redirect target.
374 raise InvalidBackendConfiguration(
375 f"The backend answered with a location this delivery cannot use ({type(exc).__name__})."
376 ) from None
377 return response
379 if isinstance(connection_failure, requests.ConnectTimeout):
380 raise BackendTimeout("The backend did not answer inside the configured limits.") from None
381 raise TransportFailure(
382 f"The backend could not be reached ({type(connection_failure).__name__}).",
383 diagnostic=ABSENT_DIAGNOSTIC,
384 ) from None
386 @staticmethod
387 def _raise_for_status(response: requests.Response, api_key: str) -> ResponseDiagnostic:
388 """Return a safe diagnostic for success, or raise the typed HTTP failure."""
389 diagnostic = _diagnostic(response, api_key)
390 failure = _status_failure(response.status_code, diagnostic, _retry_after(response))
391 if failure is not None:
392 raise failure
393 return diagnostic
395 def complete(self, request: InferenceRequest, api_key: str) -> InferenceCompletion:
396 """Return one completion, or raise the typed error the backend condition maps to."""
397 self._check_response_mode(request)
398 response = self._send(
399 "POST",
400 CHAT_COMPLETIONS_PATH,
401 api_key,
402 response_body_limit=CHAT_COMPLETION_RESPONSE_LIMIT,
403 json=self._body(request),
404 )
405 return self._read(response, api_key)
407 def discover_models(self, api_key: str) -> tuple[str, ...]:
408 """Return bounded model-id suggestions from a compatible optional models endpoint."""
409 response = self._send(
410 "GET",
411 MODELS_PATH,
412 api_key,
413 response_body_limit=MODEL_DISCOVERY_RESPONSE_LIMIT,
414 )
415 diagnostic = self._raise_for_status(response, api_key)
416 if diagnostic.redacted:
417 raise MalformedEnvelope("The backend model list contained credential material.", diagnostic=diagnostic)
418 try:
419 envelope = response.json()
420 except (ValueError, RecursionError):
421 raise MalformedEnvelope("The backend model list is not JSON.", diagnostic=diagnostic) from None
422 if not isinstance(envelope, dict) or not isinstance(envelope.get("data"), list):
423 raise MalformedEnvelope("The backend did not provide a compatible model list.", diagnostic=diagnostic)
425 models = []
426 seen = set()
427 for item in envelope["data"]:
428 model_id = item.get("id") if isinstance(item, dict) else None
429 if (
430 not isinstance(model_id, str)
431 or not model_id
432 or model_id != model_id.strip()
433 or not model_id.isprintable()
434 or len(model_id) > MODEL_MAX_LENGTH
435 or model_id in seen
436 ):
437 continue
438 seen.add(model_id)
439 models.append(model_id)
440 if len(models) == MODEL_DISCOVERY_LIMIT:
441 break
442 return tuple(models)
444 def _read(self, response, api_key: str = "") -> InferenceCompletion:
445 """Classify the answer, then parse the one envelope a completed call returns."""
446 # Captured before any parsing, so every failure below can carry what it received.
447 diagnostic = self._raise_for_status(response, api_key)
448 try:
449 envelope = response.json()
450 except (ValueError, RecursionError):
451 # Below Python 3.14 the decoder recurses, so deep nesting raises outside the ValueError tree.
452 raise MalformedEnvelope(
453 "The backend answered with a body that is not JSON.", diagnostic=diagnostic
454 ) from None
455 return self._completion(envelope, diagnostic)
457 @staticmethod
458 def _completion(envelope, diagnostic: ResponseDiagnostic = ABSENT_DIAGNOSTIC) -> InferenceCompletion:
459 """Return the completion one envelope describes, rejecting anything else."""
460 if not isinstance(envelope, dict):
461 raise MalformedEnvelope("The backend answered with no completion object.", diagnostic=diagnostic)
462 choices = envelope.get("choices")
463 if not isinstance(choices, list) or len(choices) != 1:
464 raise MalformedEnvelope("A completion carries exactly one choice.", diagnostic=diagnostic)
465 choice = choices[0]
466 if not isinstance(choice, dict):
467 raise MalformedEnvelope("The backend answered with an unreadable choice.", diagnostic=diagnostic)
468 finish_reason = choice.get("finish_reason")
469 if finish_reason not in ACCEPTED_FINISH_REASONS:
470 raise MalformedEnvelope("The backend stopped for a reason that produces no answer.", diagnostic=diagnostic)
471 message = choice.get("message")
472 if not isinstance(message, dict):
473 raise MalformedEnvelope("The backend answered with an unreadable message.", diagnostic=diagnostic)
474 content = message.get("content")
475 refusal = message.get("refusal")
476 # Content parts are common on OpenAI-compatible servers, and this delivery reads text only.
477 if content is not None and not isinstance(content, str):
478 raise MalformedEnvelope(
479 "The backend answered with content this delivery cannot read as text.", diagnostic=diagnostic
480 )
481 # A refusal is a completed call that produced no answer, not a failure to call.
482 is_refusal = bool(refusal) or not (content or "").strip()
483 return InferenceCompletion(
484 content_text=content,
485 is_refusal=is_refusal,
486 finish_reason=finish_reason,
487 backend_request_id=envelope.get("request_id"),
488 backend_response_id=envelope.get("id"),
489 backend_model=envelope.get("model"),
490 diagnostic=diagnostic,
491 )
494def encode_payload(value) -> str:
495 """Return the compact JSON one user payload travels as."""
496 return json.dumps(value, separators=(",", ":"), sort_keys=True)
499__all__ = (
500 "CHAT_COMPLETIONS_PATH",
501 "MODELS_PATH",
502 "MODEL_DISCOVERY_LIMIT",
503 "AuthenticationFailure",
504 "BackendTimeout",
505 "InferenceBackendError",
506 "InferenceCompletion",
507 "InferenceRequest",
508 "InvalidBackendConfiguration",
509 "MalformedEnvelope",
510 "OpenAICompatibleAdapter",
511 "RateLimited",
512 "TransportFailure",
513 "encode_payload",
514)