Coverage for netbox_data_import/contact_resolution.py: 99%
314 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# Copyright (C) 2026 Marcin Zieba <marcinpsk@gmail.com>
3"""Resolve, review, and apply one primary Contact decision."""
5from __future__ import annotations
7import re
8from copy import deepcopy
9from dataclasses import dataclass
11from django.contrib.contenttypes.models import ContentType
12from django.core.exceptions import ValidationError
13from django.core.validators import validate_email
14from django.db import connection, transaction
15from django.db.models import Q
17from .models import CONTACT_RESOLUTION_FIELDS, stored_import_source, validate_contact_candidate_resolution
18from .object_permissions import ObjectPermissionDenied, enforce_saved_object_permission
21def _text(value) -> str:
22 """Return a stripped string for one optional source value."""
23 if value is None:
24 return ""
25 return str(value).strip()
28# `name` is proposed from its header, because any text is a valid name. The `email` and `phone`
29# entries only keep a column that names a different field out of the name proposal.
30_ROLE_HEADER_HINTS = {
31 "email": ("email", "mail"),
32 "phone": ("phone", "number", "tel", "mobile", "cell"),
33 "name": ("name", "contact", "person"),
34}
35_PHONE_PUNCTUATION = re.compile(r"[\s()\-./]")
36_PHONE_SHAPE = re.compile(r"\+?\d{7,}")
39def _looks_like_email(value: str) -> bool:
40 try:
41 validate_email(value)
42 except ValidationError:
43 return False
44 return True
47def _looks_like_phone(value: str) -> bool:
48 return bool(_PHONE_SHAPE.fullmatch(_PHONE_PUNCTUATION.sub("", value)))
51def _header_hints_at(source_column: str, role: str) -> bool:
52 lowered = source_column.lower()
53 return any(hint in lowered for hint in _ROLE_HEADER_HINTS[role])
56def suggest_contact_roles(candidate_values: dict[str, str]) -> dict[str, str]:
57 """Map each Contact field to the candidate column that most likely supplies it.
59 Returns ``{role: source_column}`` for the roles it can recognize, and leaves out the rest
60 so the operator decides. The value shape settles ``email`` and ``phone``. Only a header
61 keyword settles ``name``, because any text is a valid name and an organization looks the
62 same as a person.
63 """
64 by_shape = {
65 "email": [column for column, value in candidate_values.items() if _looks_like_email(value)],
66 "phone": [column for column, value in candidate_values.items() if _looks_like_phone(value)],
67 }
68 suggestions = {}
69 for role, columns in by_shape.items():
70 # A header says which field a value can feed, never which of several same-shaped values
71 # is the right one: `Backup Email` carries the keyword that `Primary Contact` does not.
72 # So a second value of the same shape means the operator decides, because the collapsed
73 # modal turns a proposal into a one-click save.
74 if len(columns) == 1:
75 suggestions[role] = columns[0]
77 recognized = set(by_shape["email"]) | set(by_shape["phone"])
78 claimed = set(suggestions.values())
79 # A header that says "email" or "phone" holds a malformed value of that type, never a name.
80 named = [
81 column
82 for column in candidate_values
83 if column not in claimed
84 and column not in recognized
85 and _header_hints_at(column, "name")
86 and not _header_hints_at(column, "email")
87 and not _header_hints_at(column, "phone")
88 ]
89 if len(named) == 1:
90 suggestions["name"] = named[0]
91 return suggestions
94@dataclass(frozen=True)
95class ContactSelection:
96 """The resolved Contact values and optional selected NetBox identity."""
98 values: dict[str, str]
99 contact_id: int | None = None
102@dataclass(frozen=True)
103class ContactReview:
104 """A read-only Contact decision shared by preview and write paths."""
106 selection: ContactSelection | None
107 extra_columns: dict
108 plan: dict | None
109 candidate_values: dict[str, str]
110 suggestion: dict | None
113class DanglingProfileReference(ValidationError):
114 """Report an `adapter_config` natural key that no longer resolves to a NetBox object."""
117class ContactResolutionRequired(ValidationError):
118 """Require an operator decision for ambiguous or invalid Contact values."""
120 candidate_target = "contact"
122 def __init__(self, candidate_values: dict[str, str], message: str | None = None, suggestion=None):
123 self.candidate_values = candidate_values
124 self.suggestion = suggestion
125 super().__init__(
126 {
127 "contact": message
128 or "Select which candidate values supply Contact fields, enter Contact details, or select no contact."
129 }
130 )
133def contact_identity(contact) -> dict:
134 """Return the persisted Contact fields the preview picker needs to show it again."""
135 return {"id": contact.pk, "name": contact.name, "email": contact.email, "phone": contact.phone}
138class PrimaryContactResolver:
139 """Hide Contact resolution, lookup, assignment, and JSON migration behind one interface."""
141 @staticmethod
142 def candidate_source_columns(profile) -> dict[str, frozenset[str]]:
143 """Return the source columns each candidate target collects, grouped by target."""
144 grouped: dict[str, set[str]] = {}
145 for mapping in profile.column_mappings.filter(target_field__startswith="candidate:"):
146 target = mapping.target_field.removeprefix("candidate:")
147 grouped.setdefault(target, set()).add(mapping.source_column)
148 return {target: frozenset(columns) for target, columns in grouped.items()}
150 @staticmethod
151 def _candidate_values(row, candidate_source_columns, extra_columns) -> dict[str, str]:
152 row_values = row.get("_candidate_values", {}).get("contact", {})
153 values = (
154 {str(source): _text(value) for source, value in row_values.items() if _text(value)}
155 if isinstance(row_values, dict)
156 else {}
157 )
158 for source_column in candidate_source_columns.get("contact", ()):
159 value = _text(extra_columns.pop(source_column, ""))
160 if value and source_column not in values:
161 values[source_column] = value
162 return values
164 @classmethod
165 def selection_for_resolution(cls, profile, resolved_fields, candidate_values) -> ContactSelection | None:
166 """Return the Contact one saved resolution names, without planning an assignment."""
167 normalized = validate_contact_candidate_resolution(
168 {
169 "contact_resolution_applied": True,
170 "contact_field_sources": resolved_fields.get("contact_field_sources", {}),
171 "contact_field_values": resolved_fields.get("contact_field_values", {}),
172 "contact_id": resolved_fields.get("contact_id"),
173 },
174 profile.adapter_settings.primary_contact_lookup_field,
175 candidate_values,
176 )
177 values = dict(normalized["field_values"])
178 for field_name, source_column in normalized["field_sources"].items():
179 values[field_name] = candidate_values[source_column]
180 if not values and normalized["contact_id"] is None:
181 return None
182 return ContactSelection(values=values, contact_id=normalized["contact_id"])
184 @classmethod
185 def _selection(cls, row, profile, candidate_values, legacy_primary_contact) -> ContactSelection | None:
186 if row.get("contact_resolution_applied") is True:
187 return cls.selection_for_resolution(profile, row, candidate_values)
189 if legacy_primary_contact:
190 candidate_values.setdefault("Legacy primary contact", legacy_primary_contact)
191 return ContactSelection(
192 values={
193 "name": legacy_primary_contact,
194 profile.adapter_settings.primary_contact_lookup_field: legacy_primary_contact,
195 }
196 )
197 if not candidate_values:
198 return None
199 raise ContactResolutionRequired(candidate_values)
201 @classmethod
202 def review(
203 cls,
204 obj,
205 row: dict,
206 profile,
207 user=None,
208 *,
209 candidate_source_columns: dict[str, frozenset[str]] | None = None,
210 ) -> ContactReview:
211 """Return the effective Contact plan without writing database state."""
212 extra_columns = {}
213 import_source = stored_import_source(obj)
214 if import_source is not None and isinstance(import_source.extra_columns, dict):
215 extra_columns.update(import_source.extra_columns)
216 row_extra = row.get("_extra_columns")
217 if isinstance(row_extra, dict):
218 extra_columns.update(row_extra)
220 legacy_primary_contact = _text(row.get("primary_contact")) or _text(extra_columns.get("primary_contact"))
221 extra_columns.pop("primary_contact", None)
222 source_columns = (
223 cls.candidate_source_columns(profile) if candidate_source_columns is None else candidate_source_columns
224 )
225 candidate_values = cls._candidate_values(row, source_columns, extra_columns)
226 try:
227 selection = cls._selection(row, profile, candidate_values, legacy_primary_contact)
228 plan = None if selection is None else cls._plan(obj, profile, selection, user)
229 except ContactResolutionRequired as exc:
230 exc.suggestion = cls.suggest(candidate_values, profile, user)
231 raise
232 except DanglingProfileReference:
233 # The profile is at fault, so candidate values cannot turn this into a row decision.
234 raise
235 except ValidationError as exc:
236 if not candidate_values:
237 raise
238 suggestion = cls.suggest(candidate_values, profile, user)
239 raise ContactResolutionRequired(candidate_values, "; ".join(exc.messages), suggestion) from exc
241 suggestion = cls._suggestion_from_plan(plan)
242 return ContactReview(selection, extra_columns, plan, candidate_values, suggestion)
244 @staticmethod
245 def _suggestion_from_plan(plan) -> dict | None:
246 if not plan or plan["contact_action"] != "reuse":
247 return None
248 return {
249 "id": plan["contact_id"],
250 "name": plan["contact_name"],
251 "email": plan["contact_email"],
252 "phone": plan["contact_phone"],
253 }
255 @staticmethod
256 def _unique_contact(values, fields, user) -> dict | None:
257 """Return the one visible Contact that an exact match on *fields* identifies."""
258 from tenancy.models import Contact
260 query = Q()
261 for field in fields:
262 for value in values:
263 query |= Q(**{f"{field}__iexact": value})
264 contacts = Contact.objects.filter(query)
265 if user is not None:
266 contacts = contacts.restrict(user, "view")
267 matches = list(contacts.order_by("pk")[:2])
268 if len(matches) != 1:
269 return None
270 contact = matches[0]
271 return {
272 "id": contact.pk,
273 "name": contact.name,
274 "email": contact.email,
275 "phone": contact.phone,
276 }
278 @classmethod
279 def suggest(cls, candidate_values: dict[str, str], profile, user=None) -> dict | None:
280 """Return the one visible Contact that a candidate value in this row identifies."""
281 values = [value for value in map(_text, candidate_values.values()) if value]
282 if not values:
283 return None
284 lookup_field = profile.adapter_settings.primary_contact_lookup_field
285 # The configured lookup field answers first. A row carrying only a name matches nothing
286 # there, so the remaining identity fields answer rather than leaving the picker empty.
287 for fields in ([lookup_field], sorted(CONTACT_RESOLUTION_FIELDS - {lookup_field})):
288 match = cls._unique_contact(values, fields, user)
289 if match is not None:
290 return match
291 return None
293 @classmethod
294 def _plan_assignment(cls, obj, role, contact, user, lock):
295 from tenancy.models import ContactAssignment
297 if obj is None:
298 if user is not None and not user.has_perm("tenancy.add_contactassignment"):
299 raise ObjectPermissionDenied("tenancy.add_contactassignment")
300 return None, None, "create"
302 scope = {
303 "object_type": ContentType.objects.get_for_model(obj),
304 "object_id": obj.pk,
305 "role": role,
306 }
307 assignments = ContactAssignment.objects.select_for_update() if lock else ContactAssignment.objects
308 primary_assignments = list(assignments.filter(**scope, priority="primary")[:2])
309 if len(primary_assignments) > 1:
310 raise ValidationError(
311 {"primary_contact": "More than one primary assignment exists for the selected contact role."}
312 )
313 primary_assignment = primary_assignments[0] if primary_assignments else None
314 assignment = assignments.filter(**scope, contact=contact).first() if contact.pk is not None else None
316 if primary_assignment is not None and primary_assignment.contact_id != contact.pk:
317 enforce_saved_object_permission(primary_assignment, user, "change")
318 if assignment is None:
319 return primary_assignment, None, "replace"
320 enforce_saved_object_permission(assignment, user, "change")
321 return primary_assignment, assignment, "demote_and_promote"
322 if assignment is None:
323 if user is not None and not user.has_perm("tenancy.add_contactassignment"):
324 raise ObjectPermissionDenied("tenancy.add_contactassignment")
325 return primary_assignment, None, "create"
326 if assignment.priority != "primary":
327 enforce_saved_object_permission(assignment, user, "change")
328 return primary_assignment, assignment, "promote"
329 return primary_assignment, assignment, "unchanged"
331 @classmethod
332 def _contact_for_values(cls, contact_values, lookup_field, user, contact_queryset):
333 """Return the stored Contact these values identify, or an unsaved one to create."""
334 from tenancy.models import Contact
336 value = _text(contact_values.get(lookup_field))
337 if not value:
338 raise ValidationError(
339 {"primary_contact": f"Select or enter a value for the Contact {lookup_field} lookup field."}
340 )
341 if lookup_field == "email":
342 validate_email(value)
343 proposed_contact = Contact(**contact_values)
344 proposed_contact.full_clean()
345 contacts = list(contact_queryset.filter(**{f"{lookup_field}__iexact": value})[:2])
346 if len(contacts) > 1:
347 raise ValidationError({"primary_contact": f"More than one contact has the {lookup_field} value '{value}'."})
348 if contacts:
349 contact = contacts[0]
350 if user is not None and not Contact.objects.restrict(user, "view").filter(pk=contact.pk).exists():
351 raise ObjectPermissionDenied("tenancy.view_contact")
352 return contact
353 if user is not None and not user.has_perm("tenancy.add_contact"):
354 raise ObjectPermissionDenied("tenancy.add_contact")
355 return proposed_contact
357 @staticmethod
358 def _reject_moved_lookup(contact, selection: ContactSelection, lookup_field) -> None:
359 """Refuse a selected Contact whose lookup value no longer matches the saved decision."""
360 selected = _text(selection.values.get(lookup_field))
361 if selected and selected.casefold() != _text(getattr(contact, lookup_field)).casefold():
362 raise ValidationError(
363 {"primary_contact": f"The selected Contact no longer has the chosen {lookup_field} value."}
364 )
366 @classmethod
367 def create_contact(cls, profile, selection: ContactSelection, user=None):
368 """Store the Contact one resolution names, so it exists before any import runs.
370 Returns ``(contact, created)``. A row whose values already name a stored Contact reuses it,
371 which is what keeps a second save from making a duplicate identity.
372 """
373 from tenancy.models import Contact
375 lookup_field = profile.adapter_settings.primary_contact_lookup_field
376 if selection.contact_id is not None:
377 contacts = Contact.objects.restrict(user, "view") if user is not None else Contact.objects
378 contact = contacts.filter(pk=selection.contact_id).first()
379 if contact is None:
380 raise ValidationError({"primary_contact": "The selected NetBox Contact no longer exists."})
381 # `_plan` refuses this later, so storing it here would report success on a doomed decision.
382 cls._reject_moved_lookup(contact, selection, lookup_field)
383 return contact, False
385 with transaction.atomic():
386 cls.lock_imports()
387 contact = cls._contact_for_values(selection.values, lookup_field, user, Contact.objects.select_for_update())
388 if contact.pk is not None:
389 # atomic-exit-safe: stored-contact-reused
390 return contact, False
391 contact.save()
392 enforce_saved_object_permission(contact, user, "add")
393 # atomic-exit-safe: success-commit-intended
394 return contact, True
396 @classmethod
397 def _plan(cls, obj, profile, selection: ContactSelection, user=None, lock=False) -> dict:
398 role_name = profile.adapter_settings.primary_contact_role
399 if not role_name:
400 raise ValidationError({"primary_contact": "Select a primary contact role on the import profile."})
401 # The profile memoizes the role, and one instance serves both review and apply. Re-read it
402 # under the apply lock so a role deleted in between is refused here, not by the FK check.
403 if lock:
404 from tenancy.models import ContactRole
406 role = ContactRole.objects.select_for_update().filter(name=role_name).first()
407 else:
408 role = profile.resolved_primary_contact_role
409 if role is None:
410 raise DanglingProfileReference(
411 {
412 "primary_contact": f"The import profile references Contact Role '{role_name}', which no longer exists."
413 }
414 )
416 from tenancy.models import Contact
418 lookup_field = profile.adapter_settings.primary_contact_lookup_field
419 contact_queryset = Contact.objects.select_for_update() if lock else Contact.objects
420 if selection.contact_id is not None:
421 contact = contact_queryset.filter(pk=selection.contact_id).first()
422 if contact is None:
423 raise ValidationError({"primary_contact": "The selected NetBox Contact no longer exists."})
424 if user is not None and not Contact.objects.restrict(user, "view").filter(pk=contact.pk).exists():
425 raise ObjectPermissionDenied("tenancy.view_contact")
426 cls._reject_moved_lookup(contact, selection, lookup_field)
427 contact_values = {
428 "name": contact.name,
429 "email": contact.email,
430 "phone": contact.phone,
431 }
432 else:
433 contact_values = selection.values
434 contact = cls._contact_for_values(contact_values, lookup_field, user, contact_queryset)
436 primary_assignment, assignment, assignment_action = cls._plan_assignment(obj, role, contact, user, lock)
437 return {
438 "lookup_field": lookup_field,
439 "value": _text(getattr(contact, lookup_field, "")) or _text(contact_values.get(lookup_field)),
440 "contact_values": contact_values,
441 "role_id": role.pk,
442 "contact_id": contact.pk,
443 "contact_name": contact.name,
444 "contact_email": contact.email,
445 "contact_phone": contact.phone,
446 "contact_action": "reuse" if contact.pk is not None else "create",
447 "primary_assignment_id": primary_assignment.pk if primary_assignment is not None else None,
448 "assignment_id": assignment.pk if assignment is not None else None,
449 "assignment_action": assignment_action,
450 }
452 @classmethod
453 def apply(cls, obj, profile, review: ContactReview, user=None) -> dict | None:
454 """Apply one reviewed Contact decision and remove its legacy JSON value."""
455 if review.selection is None:
456 cls._remove_legacy_json(obj)
457 return None
459 from tenancy.models import Contact, ContactAssignment
461 with transaction.atomic():
462 cls.lock_imports()
463 plan = cls._plan(obj, profile, review.selection, user, lock=True)
464 if plan["contact_id"] is None:
465 contact = Contact(**plan["contact_values"])
466 contact.full_clean()
467 contact.save()
468 enforce_saved_object_permission(contact, user, "add")
469 else:
470 contact = Contact.objects.get(pk=plan["contact_id"])
472 scope = {
473 "object_type": ContentType.objects.get_for_model(obj),
474 "object_id": obj.pk,
475 "role_id": plan["role_id"],
476 }
477 action = plan["assignment_action"]
478 assignment = None
479 if action == "replace":
480 assignment = ContactAssignment.objects.get(pk=plan["primary_assignment_id"])
481 assignment.contact = contact
482 assignment.full_clean()
483 assignment.save(update_fields=["contact"])
484 enforce_saved_object_permission(assignment, user, "change")
485 elif action == "demote_and_promote":
486 previous = ContactAssignment.objects.get(pk=plan["primary_assignment_id"])
487 previous.priority = "secondary"
488 previous.full_clean()
489 previous.save(update_fields=["priority"])
490 enforce_saved_object_permission(previous, user, "change")
491 assignment = ContactAssignment.objects.get(pk=plan["assignment_id"])
492 elif action in ("promote", "unchanged"):
493 assignment = ContactAssignment.objects.get(pk=plan["assignment_id"])
495 if assignment is None:
496 assignment = ContactAssignment(contact=contact, priority="primary", **scope)
497 assignment.full_clean()
498 assignment.save()
499 enforce_saved_object_permission(assignment, user, "add")
500 elif assignment.priority != "primary":
501 assignment.priority = "primary"
502 assignment.full_clean()
503 assignment.save(update_fields=["priority"])
504 enforce_saved_object_permission(assignment, user, "change")
506 cls._remove_legacy_json(obj)
507 plan["saved_contact"] = contact_identity(contact)
508 # atomic-exit-safe: success-commit-intended
509 return plan
511 @staticmethod
512 def _remove_legacy_json(obj) -> None:
513 """Drop the legacy primary_contact value once a native Contact assignment holds it."""
514 import_source = stored_import_source(obj)
515 if import_source is None or "primary_contact" not in import_source.extra_columns:
516 return
517 extra_columns = deepcopy(import_source.extra_columns)
518 extra_columns.pop("primary_contact")
519 import_source.extra_columns = extra_columns
520 import_source.save(update_fields=["extra_columns"])
522 @staticmethod
523 def lock_imports() -> None:
524 """Serialize import jobs that can create shared Contact identities."""
525 with connection.cursor() as cursor:
526 cursor.execute("SELECT pg_advisory_xact_lock(hashtext(%s))", ["netbox_data_import.contact_sync"])