Coverage for netbox_data_import/inference_backend.py: 98.39%
62 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"""Resolve the active Inference Backend (specification 8.2).
5The enabled database row is the active backend. The `inference_backend` setting is the whole-backend
6fallback and acts only when no enabled row exists. The two sources are never merged field by field,
7so a resolved backend always names the one source it came from.
8"""
10from dataclasses import dataclass
11from typing import Any
13from django.core.exceptions import ValidationError
15from .inference_credentials import CredentialReference
16from .inference_settings import (
17 FILE_FALLBACK_KEY,
18 FILE_FALLBACK_SETTING,
19 ORIGIN_ALLOWLIST_SETTING,
20 PROPOSAL_CANDIDATE_LIMIT_DEFAULT,
21 PROPOSAL_CANDIDATE_LIMIT_MAX,
22 PROPOSAL_CANDIDATE_LIMIT_SETTING,
23 validate_credential_reference,
24 validate_file_fallback,
25 validate_proposal_candidate_limit,
26)
27from .inference_trust import InvalidInferenceConfiguration, validate_api_root
29PLUGIN_NAME = "netbox_data_import"
31SOURCE_DATABASE = "database"
32SOURCE_FILE_FALLBACK = "file-fallback"
35class NoActiveInferenceBackend(Exception):
36 """No enabled database row exists and no file fallback is configured."""
39def plugin_settings() -> dict[str, Any]:
40 """Return this plugin's PLUGINS_CONFIG entry."""
41 from django.conf import settings
43 return dict(settings.PLUGINS_CONFIG.get(PLUGIN_NAME, {}))
46def proposal_candidate_limit() -> int:
47 """Return how many candidates one request offers the backend, its page size.
49 Defaulted only when the key is omitted. A denser Device is searched a page at a time, so this
50 holds prompt cost down without bounding what the Device may have.
51 """
52 config = plugin_settings()
53 if PROPOSAL_CANDIDATE_LIMIT_SETTING not in config:
54 return PROPOSAL_CANDIDATE_LIMIT_DEFAULT
55 return validate_proposal_candidate_limit(config[PROPOSAL_CANDIDATE_LIMIT_SETTING])
58def proposal_eligible_set_limit() -> int:
59 """Return the ceiling on the eligible set one proposal may freeze, however it is paged.
61 The set is stored, not prompted, so the bound here is the row and the retrieval, not the
62 prompt. Past it the answer is authoritative pre-filtering, not more pages.
63 """
64 return PROPOSAL_CANDIDATE_LIMIT_MAX
67def origin_allowlist() -> tuple[str, ...]:
68 """Return the deployment's approved origins."""
69 return tuple(plugin_settings().get(ORIGIN_ALLOWLIST_SETTING, ()))
72def validate_backend_fields(api_root: str, authentication: str, credential_reference: Any) -> None:
73 """Reject backend fields the deployment may not use, as a Django field-keyed ValidationError."""
74 try:
75 validate_credential_reference(credential_reference)
76 except InvalidInferenceConfiguration as exc:
77 raise ValidationError({"credential_reference": str(exc)}) from exc
78 try:
79 validate_api_root(api_root, allowlist=origin_allowlist(), authentication=authentication)
80 except InvalidInferenceConfiguration as exc:
81 raise ValidationError({"api_root": str(exc)}) from exc
84@dataclass(frozen=True)
85class ResolvedInferenceBackend:
86 """One whole backend, from exactly one source."""
88 backend_key: str
89 display_name: str
90 adapter_type: str
91 api_root: str
92 model: str
93 authentication: str
94 response_mode: str
95 credential_reference: CredentialReference
96 connect_timeout: int
97 read_timeout: int
98 max_concurrent_requests: int
99 source: str
101 def metadata(self) -> dict[str, str]:
102 """Return the backend metadata a job may record: no credential reference, no secret."""
103 return {
104 "backend_key": self.backend_key,
105 "backend_source": self.source,
106 "backend_adapter_type": self.adapter_type,
107 "backend_model": self.model,
108 }
111def adapter_for_backend(backend: ResolvedInferenceBackend, *, deadline=None):
112 """Build the adapter that both proposal jobs and foreground tests use for this backend."""
113 if backend.adapter_type != "openai_compatible":
114 raise InvalidInferenceConfiguration(f"Unsupported Inference Backend adapter '{backend.adapter_type}'.")
115 from .inference_adapter import OpenAICompatibleAdapter
117 return OpenAICompatibleAdapter(
118 api_root=backend.api_root,
119 model=backend.model,
120 allowlist=origin_allowlist(),
121 authentication=backend.authentication,
122 response_mode=backend.response_mode,
123 connect_timeout=backend.connect_timeout,
124 read_timeout=backend.read_timeout,
125 deadline=deadline,
126 )
129def _from_row(row, allowlist) -> ResolvedInferenceBackend:
130 """Return the resolved backend one enabled database row describes."""
131 # A saved row outlives the allowlist that approved it, so spec 8.3 validates both sources alike.
132 validate_api_root(row.api_root, allowlist=allowlist, authentication=row.authentication)
133 return ResolvedInferenceBackend(
134 backend_key=row.backend_key,
135 display_name=row.display_name,
136 adapter_type=row.adapter_type,
137 api_root=row.api_root,
138 model=row.model,
139 authentication=row.authentication,
140 response_mode=row.response_mode,
141 credential_reference=CredentialReference.from_mapping(row.credential_reference),
142 connect_timeout=row.connect_timeout,
143 read_timeout=row.read_timeout,
144 max_concurrent_requests=row.max_concurrent_requests,
145 source=SOURCE_DATABASE,
146 )
149def _from_file_fallback(mapping, allowlist) -> ResolvedInferenceBackend:
150 """Return the resolved backend the file fallback describes, under its fixed key."""
151 validated = validate_file_fallback(mapping, allowlist)
152 return ResolvedInferenceBackend(
153 backend_key=FILE_FALLBACK_KEY,
154 display_name=validated["display_name"],
155 adapter_type=validated["adapter_type"],
156 api_root=validated["api_root"],
157 model=validated["model"],
158 authentication=validated["authentication"],
159 response_mode=validated["response_mode"],
160 credential_reference=CredentialReference.from_mapping(validated["credential_reference"]),
161 connect_timeout=validated["connect_timeout"],
162 read_timeout=validated["read_timeout"],
163 max_concurrent_requests=validated["max_concurrent_requests"],
164 source=SOURCE_FILE_FALLBACK,
165 )
168def resolve_backend_by_id(pk: int) -> ResolvedInferenceBackend:
169 """Return exactly the authorized row, even when it is disabled.
171 An operator tests a backend to decide whether to enable it, so `enabled` is not a filter.
172 Rows only, never the file fallback: an editable key must not let a scoped operator resolve
173 the deployment's own credential reference. A deleted row cannot select its replacement.
174 """
175 from .models import InferenceBackend
177 row = InferenceBackend.objects.filter(pk=pk).first()
178 if row is None:
179 raise NoActiveInferenceBackend(f"No Inference Backend row has ID {pk}.")
180 return _from_row(row, origin_allowlist())
183def resolve_active_backend() -> ResolvedInferenceBackend:
184 """Return the active backend, naming the one source it came from."""
185 from .models import InferenceBackend
187 row = InferenceBackend.objects.filter(enabled=True).first()
188 if row is not None:
189 return _from_row(row, origin_allowlist())
190 config = plugin_settings()
191 if FILE_FALLBACK_SETTING not in config:
192 raise NoActiveInferenceBackend(
193 "No Inference Backend row is enabled and no 'inference_backend' fallback is configured."
194 )
195 return _from_file_fallback(config[FILE_FALLBACK_SETTING], config.get(ORIGIN_ALLOWLIST_SETTING, ()))
198__all__ = (
199 "SOURCE_DATABASE",
200 "SOURCE_FILE_FALLBACK",
201 "NoActiveInferenceBackend",
202 "ResolvedInferenceBackend",
203 "origin_allowlist",
204 "plugin_settings",
205 "proposal_candidate_limit",
206 "proposal_eligible_set_limit",
207 "resolve_active_backend",
208 "resolve_backend_by_id",
209 "validate_backend_fields",
210)