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