Coverage for netbox_data_import/inference_backend.py: 100%
61 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"""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 validate_credential_reference,
21 validate_file_fallback,
22)
23from .inference_trust import InvalidInferenceConfiguration, validate_api_root
25PLUGIN_NAME = "netbox_data_import"
27SOURCE_DATABASE = "database"
28SOURCE_FILE_FALLBACK = "file-fallback"
31class NoActiveInferenceBackend(Exception):
32 """No enabled database row exists and no file fallback is configured."""
35def plugin_settings() -> dict[str, Any]:
36 """Return this plugin's PLUGINS_CONFIG entry."""
37 from django.conf import settings
39 return dict(settings.PLUGINS_CONFIG.get(PLUGIN_NAME, {}))
42def origin_allowlist() -> tuple[str, ...]:
43 """Return the deployment's approved origins."""
44 return tuple(plugin_settings().get(ORIGIN_ALLOWLIST_SETTING, ()))
47def validate_backend_fields(api_root: str, authentication: str, credential_reference: Any) -> None:
48 """Reject backend fields the deployment may not use, as a Django field-keyed ValidationError."""
49 try:
50 validate_credential_reference(credential_reference)
51 except InvalidInferenceConfiguration as exc:
52 raise ValidationError({"credential_reference": str(exc)}) from exc
53 try:
54 validate_api_root(api_root, allowlist=origin_allowlist(), authentication=authentication)
55 except InvalidInferenceConfiguration as exc:
56 raise ValidationError({"api_root": str(exc)}) from exc
59@dataclass(frozen=True)
60class ResolvedInferenceBackend:
61 """One whole backend, from exactly one source."""
63 backend_key: str
64 display_name: str
65 adapter_type: str
66 api_root: str
67 model: str
68 authentication: str
69 response_mode: str
70 credential_reference: CredentialReference
71 connect_timeout: int
72 read_timeout: int
73 source: str
75 def metadata(self) -> dict[str, str]:
76 """Return the backend metadata a job may record: no credential reference, no secret."""
77 return {
78 "backend_key": self.backend_key,
79 "backend_source": self.source,
80 "backend_adapter_type": self.adapter_type,
81 "backend_model": self.model,
82 }
85def _from_row(row, allowlist) -> ResolvedInferenceBackend:
86 """Return the resolved backend one enabled database row describes."""
87 # A saved row outlives the allowlist that approved it, so spec 8.3 validates both sources alike.
88 validate_api_root(row.api_root, allowlist=allowlist, authentication=row.authentication)
89 return ResolvedInferenceBackend(
90 backend_key=row.backend_key,
91 display_name=row.display_name,
92 adapter_type=row.adapter_type,
93 api_root=row.api_root,
94 model=row.model,
95 authentication=row.authentication,
96 response_mode=row.response_mode,
97 credential_reference=CredentialReference.from_mapping(row.credential_reference),
98 connect_timeout=row.connect_timeout,
99 read_timeout=row.read_timeout,
100 source=SOURCE_DATABASE,
101 )
104def _from_file_fallback(mapping, allowlist) -> ResolvedInferenceBackend:
105 """Return the resolved backend the file fallback describes, under its fixed key."""
106 validated = validate_file_fallback(mapping, allowlist)
107 return ResolvedInferenceBackend(
108 backend_key=FILE_FALLBACK_KEY,
109 display_name=validated["display_name"],
110 adapter_type=validated["adapter_type"],
111 api_root=validated["api_root"],
112 model=validated["model"],
113 authentication=validated["authentication"],
114 response_mode=validated["response_mode"],
115 credential_reference=CredentialReference.from_mapping(validated["credential_reference"]),
116 connect_timeout=validated["connect_timeout"],
117 read_timeout=validated["read_timeout"],
118 source=SOURCE_FILE_FALLBACK,
119 )
122def resolve_backend_by_id(pk: int) -> ResolvedInferenceBackend:
123 """Return exactly the authorized row, even when it is disabled.
125 An operator tests a backend to decide whether to enable it, so `enabled` is not a filter.
126 Rows only, never the file fallback: an editable key must not let a scoped operator resolve
127 the deployment's own credential reference. A deleted row cannot select its replacement.
128 """
129 from .models import InferenceBackend
131 row = InferenceBackend.objects.filter(pk=pk).first()
132 if row is None:
133 raise NoActiveInferenceBackend(f"No Inference Backend row has ID {pk}.")
134 return _from_row(row, origin_allowlist())
137def resolve_active_backend() -> ResolvedInferenceBackend:
138 """Return the active backend, naming the one source it came from."""
139 from .models import InferenceBackend
141 row = InferenceBackend.objects.filter(enabled=True).first()
142 if row is not None:
143 return _from_row(row, origin_allowlist())
144 config = plugin_settings()
145 if FILE_FALLBACK_SETTING not in config:
146 raise NoActiveInferenceBackend(
147 "No Inference Backend row is enabled and no 'inference_backend' fallback is configured."
148 )
149 return _from_file_fallback(config[FILE_FALLBACK_SETTING], config.get(ORIGIN_ALLOWLIST_SETTING, ()))
152__all__ = (
153 "SOURCE_DATABASE",
154 "SOURCE_FILE_FALLBACK",
155 "NoActiveInferenceBackend",
156 "ResolvedInferenceBackend",
157 "origin_allowlist",
158 "plugin_settings",
159 "resolve_active_backend",
160 "resolve_backend_by_id",
161 "validate_backend_fields",
162)