« prev ^ index » next coverage.py v7.14.3, created at 2026-07-01 19:09 +0000
1# SPDX-License-Identifier: Apache-2.0
2# Copyright (C) 2025 Marcin Zieba <marcinpsk@gmail.com>
3"""Core renaming engine — rule lookup and interface rename logic.
5This module is imported lazily by signals.py so that model imports happen
6after Django is fully initialised.
7"""
9import ast
10import contextlib
11import logging
12import re
13import threading
15from django.core.exceptions import ValidationError
16from django.db import IntegrityError, transaction
17from django.db.models import Aggregate, F, TextField, Value
18from django.db.models.functions import Cast, Coalesce, Concat, Length
20logger = logging.getLogger(__name__)
22# In-process cache of the enabled InterfaceNameRule set, keyed by a cheap fingerprint
23# of that set. find_matching_rule() is called once per module row when a device's
24# module-sync table is rendered, and each call used to run a DB query per scope
25# candidate per tier; loading the (small) rule set once and matching in memory removes
26# that per-row query storm. The fingerprint (see _enabled_rules_version) is re-read once
27# per call and the set reloaded only when it changes, so it self-invalidates on any
28# create/delete/edit — including raw bulk ``.update()`` of any matching field and SET_NULL
29# cascades that bypass auto_now, and across test transactions — with no signal wiring.
30# A reload publishes the new set by rebinding this name to a fresh dict in one atomic assignment
31# (see _get_enabled_rules), so a concurrent reader on another worker thread sees either the whole
32# old set or the whole new one — never exact/regex/memo torn across two versions.
33_RULE_CACHE = {"version": None, "exact": (), "regex": (), "memo": {}}
35# Per-version cap on the find_matching_rule memo. A long-lived worker that sees many distinct
36# (module_type, scope) contexts under a stable rule set would otherwise grow it without bound;
37# on overflow the memo is cleared wholesale and rebuilt lazily. The cap is far above the number
38# of contexts in any single module-sync render, so it never churns mid-render.
39_MEMO_MAX = 4096
41# Sentinel for the memo read. A single ``memo.get(sig, _MEMO_MISS)`` is one atomic dict lookup, so a
42# concurrent ``memo.clear()`` at the cap can't wedge between a membership test and the subscript the
43# way ``if sig in memo: return memo[sig]`` could — that compound read can raise a sporadic KeyError
44# under threaded workers sharing one per-version memo. A real "no rule matched" is memoized as None,
45# so the miss sentinel must be a distinct object, not None.
46_MEMO_MISS = object()
48# Per-thread pin depth (see pinned_rule_cache). While > 0 on the current thread, _get_enabled_rules
49# trusts the loaded set without re-reading the fingerprint, so an internal batch that wraps its loop
50# turns N per-call fingerprint queries into one. Thread-local so concurrent requests don't affect
51# each other; the depth defaults to 0, so any path that does not pin (the per-row signal handler,
52# single applies) re-reads the fingerprint on every call exactly as before.
53_pin = threading.local()
56@contextlib.contextmanager
57def pinned_rule_cache():
58 """Pin the enabled-rule set for the duration of the block, skipping the per-call fingerprint query.
60 find_matching_rule() normally re-reads a cheap fingerprint on every call to detect rule-set
61 changes. An internal batch that makes many lookups against an unchanging rule set — e.g.
62 _apply_rules_for_device_deferred(), which re-applies rules to every module on a device after a
63 virtual-chassis change — can wrap its loop in this context manager so the set is loaded and
64 fingerprinted once and reused for every call inside the block, turning N fingerprint queries
65 into one::
67 with pinned_rule_cache():
68 for module in modules:
69 apply_interface_name_rules(module, module.module_bay, force_reapply=True)
71 This is an INR-internal helper: only code that owns the batch loop pins, so it never becomes a
72 cross-plugin dependency. Other plugins integrate through the ``predict_module_interface_names``
73 signal, which is dispatched one row at a time — so an external caller neither can nor needs to
74 pin, and INR exposes no batch API across the plugin boundary.
76 Scope is explicit and per-thread: outside the block (and in any other thread/request) the normal
77 self-invalidating behaviour is unchanged, so there is no global staleness window. Nesting is
78 safe — only the outermost block manages the pin. Priming is lazy: the first find_matching_rule
79 inside the block does the one real fingerprint read + reload and captures that rule set into
80 thread-local state; the rest reuse that *snapshot* — so a block that makes no lookups does no
81 query at all, and a concurrent request reloading the shared cache on another thread cannot switch
82 this batch's rule set mid-loop. The loop may rename interfaces, but it must not edit rules: edits
83 to InterfaceNameRule made *inside* the block are not observed until it exits.
84 """
85 depth = getattr(_pin, "depth", 0)
86 _pin.depth = depth + 1
87 if depth == 0: 87 ↛ 89line 87 didn't jump to line 89 because the condition on line 87 was always true
88 _pin.primed = False # the first lookup inside primes the cache (lazily — an empty block stays query-free)
89 try:
90 yield
91 finally:
92 _pin.depth -= 1
93 if _pin.depth == 0: 93 ↛ exitline 93 didn't return from function 'pinned_rule_cache' because the condition on line 93 was always true
94 # Release the per-thread snapshot so the next block re-primes from the live cache.
95 _pin.primed = False
96 for attr in ("exact", "regex", "memo"):
97 _pin.__dict__.pop(attr, None)
100def _compile_pattern(pattern):
101 """Compile a regex *pattern* once, returning the compiled object or None for an invalid pattern.
103 A None result is skipped at match time, mirroring the previous per-match ``try/except re.error``
104 without recompiling the pattern on every lookup.
105 """
106 try:
107 return re.compile(pattern)
108 except re.error:
109 return None
112# Rule columns that affect matching or output, in a fixed order. The enabled-set fingerprint
113# (see _enabled_rules_version) is an md5 over these for every enabled rule, so it changes on ANY
114# edit that could change a lookup result. ``description`` is excluded — it is operator notes that
115# never affect matching. ``id`` anchors each row to its identity, so a compensating swap between
116# two rules (which leaves count + column sums unchanged) still changes the hash.
117_VERSION_COLUMNS = (
118 "id",
119 "module_type_id",
120 "module_type_is_regex",
121 "module_type_pattern",
122 "parent_module_type_id",
123 "device_type_id",
124 "platform_id",
125 "name_template",
126 "channel_count",
127 "channel_start",
128 "applies_to_device_interfaces",
129)
132class _Md5OrderedStringAgg(Aggregate):
133 """``md5(string_agg(<row>, <delim> ORDER BY id))`` expressed as one ORM aggregate.
135 Built through the ORM rather than a hand-formatted SQL string, so the table/column identifiers are
136 quoted by Django's compiler and the delimiter is a bound parameter — there is no string-interpolated
137 SQL to audit for injection. The template is ours, so it does not depend on ``StringAgg``'s Python
138 signature, which differs across the Django 5.x–6.x versions in CI. ``ORDER BY id`` makes string_agg
139 deterministic — a stable row order yields a stable hash for an unchanged set.
140 """
142 function = "STRING_AGG"
143 template = "MD5(%(function)s(%(expressions)s ORDER BY id))"
144 output_field = TextField()
147def _version_row_signature():
148 """Build the per-rule text signature expression for the fingerprint.
150 Each matching/output column is cast to text and emitted length-prefixed as ``<char length>:<value>``.
151 That makes the row encoding self-delimiting: distinct column tuples can never serialize to the same
152 string, even if a text column (name_template / module_type_pattern) contains digits, a colon, or the
153 control characters a plain separator scheme would rely on being absent. A nullable FK is coalesced to
154 '' (length 0), so null stays distinct from any value while keeping the column's slot.
155 """
156 empty = Value("", output_field=TextField())
157 colon = Value(":", output_field=TextField())
158 parts = []
159 for column in _VERSION_COLUMNS:
160 cast = Cast(F(column), output_field=TextField())
161 value = Coalesce(cast, empty, output_field=TextField()) if column.endswith("_id") else cast
162 parts.append(Cast(Length(value), output_field=TextField()))
163 parts.append(colon)
164 parts.append(value)
165 return Concat(*parts, output_field=TextField())
168# The signature expression is constant, so build it once and reuse it across calls.
169_ROW_SIGNATURE = _version_row_signature()
172def _enabled_rules_version():
173 """Return a deterministic content fingerprint of the enabled-rule set.
175 Computed server-side as an md5 over every enabled rule's matching/output columns, row-ordered
176 by pk, so the fingerprint changes on ANY edit that could change a lookup — including a raw bulk
177 ``.update()`` of a text field (name_template / module_type_pattern) or a boolean
178 (module_type_is_regex), which an aggregate of counts/sums cannot see, and a compensating edit
179 that keeps the column sums constant. It is one query returning a single 32-char hash, so it
180 stays cheap enough to re-read on every call.
182 Each column is length-prefixed (see _version_row_signature), so the per-row encoding is
183 self-delimiting and rows concatenate unambiguously — distinct rule sets cannot collide even if a
184 text column contains arbitrary bytes. A nullable FK renders as the empty string (a real id never
185 does), keeping null distinct from any value. The empty set aggregates to NULL → coalesced to a
186 stable empty fingerprint, so "no enabled rules" is fixed.
187 """
188 from .models import InterfaceNameRule
190 return InterfaceNameRule.objects.filter(enabled=True).aggregate(
191 fingerprint=Coalesce(
192 _Md5OrderedStringAgg(_ROW_SIGNATURE, Value("", output_field=TextField())),
193 Value("", output_field=TextField()),
194 )
195 )["fingerprint"]
198def _get_enabled_rules():
199 """Return ``(exact_rules, regex_rules, memo)``, reloading only when the rule set changes.
201 ``exact_rules`` are ordered by ``(module_type__model, pk)`` to mirror the model's default
202 ordering (so an in-memory ``first match`` equals the previous ``.first()``). ``regex_rules``
203 is a tuple of ``(compiled_pattern, rule)`` pairs, pre-sorted once by ``(-pattern length, pk)``
204 and with each pattern compiled once, so the regex tier neither re-sorts nor recompiles per
205 call. ``memo`` caches find_matching_rule results for the current rule-set version.
207 Inside a ``pinned_rule_cache()`` block on this thread, once the set has been primed (by the first
208 lookup in the block) the fingerprint query is skipped and the snapshot captured at prime time is
209 returned — never the live ``_RULE_CACHE``, which another thread may reload mid-block.
210 """
211 global _RULE_CACHE
213 pinned = getattr(_pin, "depth", 0) > 0
214 if pinned and getattr(_pin, "primed", False):
215 # Serve the snapshot captured when this block primed, not the shared cache: a concurrent
216 # request on another thread may reload _RULE_CACHE to a different version while we iterate, and
217 # a pinned batch must match every item against one consistent rule set.
218 return _pin.exact, _pin.regex, _pin.memo
220 from .models import InterfaceNameRule
222 # Read the module global exactly once. A reload below publishes the new set by rebinding
223 # _RULE_CACHE to a brand-new dict (a single atomic name assignment) rather than mutating this
224 # one in place, so this local is a consistent snapshot: exact/regex/memo can never be torn
225 # across two rule-set versions even if another thread reloads between the reads at the end.
226 cache = _RULE_CACHE
227 version = _enabled_rules_version()
228 if cache["version"] != version:
229 rules = list(InterfaceNameRule.objects.filter(enabled=True).order_by("module_type__model", "pk"))
230 exact = tuple(r for r in rules if not r.module_type_is_regex)
231 regex_rules = sorted(
232 (r for r in rules if r.module_type_is_regex),
233 key=lambda r: (-len(r.module_type_pattern or ""), r.pk),
234 )
235 regex = tuple((_compile_pattern(r.module_type_pattern), r) for r in regex_rules)
236 # Publish the whole new version atomically: a reader either sees the old dict or this one,
237 # never a mix of the two. Last writer wins; a concurrent reload to the same version just
238 # rebuilds redundantly, never corrupts.
239 cache = {"version": version, "exact": exact, "regex": regex, "memo": {}}
240 _RULE_CACHE = cache
241 if pinned:
242 # Pin this thread to the freshly-resolved set for the rest of the block. The tuples are
243 # immutable and the memo is COPIED into thread-local state — not aliased — so the pinned batch
244 # neither shares nor races the global memo: another thread clearing the shared memo at the cap
245 # can't evict our warmed entries (or wedge a KeyError) mid-loop, and entries we add stay private.
246 _pin.exact = cache["exact"]
247 _pin.regex = cache["regex"]
248 _pin.memo = dict(cache["memo"])
249 _pin.primed = True
250 return _pin.exact, _pin.regex, _pin.memo
251 return cache["exact"], cache["regex"], cache["memo"]
254def _get_parent_module_type(module_bay):
255 """Return the module type of the module installed in the parent bay, or None.
257 Used by ``apply_interface_name_rules`` to scope rules to a specific parent
258 module type (e.g., SFP inside a CVR-X2-SFP converter).
259 """
260 if module_bay.parent:
261 parent_bay = module_bay.parent
262 if hasattr(parent_bay, "installed_module") and parent_bay.installed_module: 262 ↛ 264line 262 didn't jump to line 264 because the condition on line 262 was always true
263 return parent_bay.installed_module.module_type
264 return None
267def _collect_unrenamed(interfaces, rule, raw_names, force_reapply):
268 """Return the subset of *interfaces* that should be processed by the rule.
270 Normal (non-force) mode: only interfaces whose current name is still in the
271 raw template names (idempotency guard).
273 force_reapply, non-channel: all interfaces (e.g. vc_position changed).
275 force_reapply, channel rule: one interface per base name, matching when the
276 full base OR its last path segment is in *raw_names*; prefers the ":0"
277 channel to avoid duplicate-name errors on re-apply.
278 """
279 if not force_reapply:
280 return [i for i in interfaces if i.name in raw_names]
281 if rule.channel_count == 0:
282 return interfaces
283 # Breakout + force_reapply: deduplicate by base, prefer ":0"
284 seen_bases: dict = {}
285 for i in interfaces:
286 base = i.name.rsplit(":", 1)[0]
287 # Also match when the last path segment equals a raw name (already-renamed bases).
288 if base in raw_names or base.rsplit("/", 1)[-1] in raw_names: 288 ↛ 285line 288 didn't jump to line 285 because the condition on line 288 was always true
289 if base not in seen_bases or i.name.endswith(":0"):
290 seen_bases[base] = i
291 return list(seen_bases.values())
294def apply_interface_name_rules(module, module_bay, force_reapply=False):
295 """Apply InterfaceNameRule rename after module installation.
297 Looks up a matching rule for (module_type, parent_module_type, device_type, platform)
298 and renames interfaces created by NetBox's template instantiation.
300 Only processes interfaces whose name still matches the raw bay position
301 (i.e., haven't been renamed yet), ensuring idempotency. Pass
302 ``force_reapply=True`` to skip this check and re-apply rules to ALL
303 module interfaces (used when vc_position or other variables change).
305 Returns:
306 Number of interfaces renamed/created, or 0 if no rule matched.
308 """
309 from dcim.models import Interface
311 device_type = module.device.device_type if module.device else None
312 platform = module.device.platform if module.device else None
313 rule = find_matching_rule(module.module_type, _get_parent_module_type(module_bay), device_type, platform)
315 if not rule:
316 return 0
318 variables = build_variables(module_bay, device=module.device)
319 interfaces = list(Interface.objects.filter(module=module))
321 if not interfaces:
322 return 0
324 # Determine raw names NetBox assigned from templates; fall back to bay_position.
325 raw_names = _get_raw_interface_names(module) or {variables["bay_position"]}
326 unrenamed = _collect_unrenamed(interfaces, rule, raw_names, force_reapply)
328 if not unrenamed:
329 return 0 # Already renamed (idempotent guard)
331 renamed = 0
332 conflicts: list = []
333 for iface in unrenamed:
334 vars_copy = dict(variables)
335 vars_copy["base"] = iface.name
336 try:
337 renamed += _apply_rule_to_interface(rule, iface, vars_copy, module, conflicts=conflicts)
338 except (ValueError, ValidationError, IntegrityError):
339 # The collision pre-check closes the common case, but a concurrent
340 # insert can still win between that check and the save — surfacing
341 # here as IntegrityError/ValidationError out of the per-interface
342 # atomic block (which has already rolled back cleanly). Log and keep
343 # going so one racing interface never aborts the whole install batch,
344 # mirroring apply_rule_to_existing().
345 logger.exception(
346 "Failed to apply rule '%s' to interface '%s' (id=%s); skipping.",
347 rule,
348 iface.name,
349 iface.pk,
350 )
352 if unrenamed and renamed == 0 and not conflicts:
353 # All interfaces already have the names the rule would produce — flag as
354 # potentially obsolete (e.g., newer NetBox generates correct names natively).
355 # Skipped when the 0-count was caused by name collisions (a different reason
356 # than a no-op rule), so a collision never mislabels the rule as deprecated.
357 _flag_rule_potentially_deprecated(rule)
359 return renamed
362def predict_rule_output(module, module_bay, raw_names):
363 """Predict the names apply_interface_name_rules would produce for raw_names.
365 Pure function — does not save, mutate, or query the Interface table. Used
366 by external integrations (e.g., netbox-librenms-plugin) that need to know
367 the post-rename names without actually applying any rule.
369 For breakout rules (channel_count > 0), each raw name expands to
370 channel_count predicted names. For simple renames, one name in → one name
371 out. Returns raw_names unchanged when no rule matches or evaluation fails.
372 """
373 device_type = module.device.device_type if module.device else None
374 platform = module.device.platform if module.device else None
375 rule = find_matching_rule(module.module_type, _get_parent_module_type(module_bay), device_type, platform)
376 if not rule:
377 return list(raw_names)
379 variables = build_variables(module_bay, device=module.device)
381 output = []
382 for raw_name in raw_names:
383 vars_copy = dict(variables)
384 vars_copy["base"] = raw_name
385 try:
386 if rule.channel_count > 0:
387 temp = []
388 for ch in range(rule.channel_count):
389 vars_copy["channel"] = str(rule.channel_start + ch)
390 temp.append(evaluate_name_template(rule.name_template, vars_copy))
391 output.extend(temp)
392 else:
393 output.append(evaluate_name_template(rule.name_template, vars_copy))
394 except (ValueError, TypeError, re.error):
395 # Template eval failed; apply path would also fail and leave the
396 # interface alone, so the predicted name is the raw name.
397 output.append(raw_name)
399 return output
402def _try_rename_device_interface(rule, iface, vc_position, device, renamed_pks, conflicts=None):
403 """Attempt to rename a single device-level interface using *rule*.
405 Returns ``True`` if the interface was successfully renamed, ``False`` otherwise.
406 Mutates ``renamed_pks`` on success.
408 A computed name already taken by another interface on the device is skipped
409 with a tidy WARNING (no traceback), mirroring the module-install path; pass a
410 list as *conflicts* to also collect them. ``full_clean()`` remains the
411 backstop for the rarer cross-member (VC) uniqueness violation.
412 """
413 if iface.pk in renamed_pks:
414 return False # Already renamed by a higher-priority rule
416 if rule.module_type_pattern:
417 try:
418 if not re.fullmatch(rule.module_type_pattern, iface.name):
419 return False
420 except re.error:
421 return False
423 port = iface.name.rsplit("/", 1)[-1] if "/" in iface.name else iface.name
424 variables = {"vc_position": vc_position, "base": iface.name, "port": port}
426 try:
427 new_name = evaluate_name_template(rule.name_template, variables)
428 except (ValueError, TypeError, re.error):
429 logger.exception(
430 "Failed to evaluate template %r for interface %s (rule %s)",
431 rule.name_template,
432 iface.name,
433 rule.pk,
434 )
435 return False
437 if new_name == iface.name:
438 return False
440 # Pre-check device-scope name uniqueness so an expected collision is a clean
441 # WARNING + skip instead of an ERROR traceback out of full_clean().
442 if _name_exists_on_device(device, new_name, exclude_pk=iface.pk):
443 _record_conflict(conflicts, device, iface.name, new_name, iface.pk)
444 return False
446 old_name = iface.name
447 iface.name = new_name
448 try:
449 iface.full_clean()
450 except ValidationError as exc:
451 logger.warning(
452 "Validation failed renaming device interface %r → %r (rule %s, device %s); skipping: %s",
453 old_name,
454 new_name,
455 rule.pk,
456 device.pk,
457 exc,
458 )
459 iface.name = old_name
460 return False
461 try:
462 iface.save()
463 except (IntegrityError, ValidationError):
464 logger.exception(
465 "DB save failed for device interface %s → %s (rule %s, device %s)",
466 old_name,
467 new_name,
468 rule.pk,
469 device.pk,
470 )
471 iface.name = old_name
472 return False
474 renamed_pks.add(iface.pk)
475 logger.debug("Renamed device interface %s → %s (rule %s, device %s)", old_name, new_name, rule.pk, device.pk)
476 return True
479def apply_device_interface_rules(device):
480 """Rename device-level interfaces (module=None) when a device joins/changes position in a VC.
482 Finds all enabled rules with ``applies_to_device_interfaces=True`` that match the device's
483 type and platform, then renames any matching interfaces using the name_template.
485 Template variables available: ``{vc_position}``, ``{base}`` (full current name),
486 ``{port}`` (segment after the last ``/``, or the full name if no ``/`` present).
488 Returns the number of interfaces renamed.
489 """
490 from dcim.models import Interface
492 from .models import InterfaceNameRule
494 if not getattr(device, "virtual_chassis_id", None):
495 return 0 # Only rename for VC members (vc_position must be set)
497 if device.vc_position is None:
498 return 0 # vc_position unset (e.g. VC master before position assigned)
500 vc_position = str(device.vc_position)
501 device_type = getattr(device, "device_type", None)
502 platform = getattr(device, "platform", None)
504 from django.db.models import Q
506 rules = list(
507 InterfaceNameRule.objects.filter(
508 applies_to_device_interfaces=True,
509 enabled=True,
510 )
511 .filter(Q(device_type=device_type) | Q(device_type__isnull=True))
512 .filter(Q(platform=platform) | Q(platform__isnull=True))
513 )
514 # Sort Python-side: specificity_score descending, then module_type_pattern length
515 # descending (for device-interface rules with ties), then pk ascending for stability.
516 # (InterfaceNameRule has no DB 'priority' field; specificity_score is a property.)
517 rules.sort(
518 key=lambda r: (
519 -r.specificity_score,
520 -(len(r.module_type_pattern or "") if r.applies_to_device_interfaces else 0),
521 r.pk,
522 )
523 )
525 if not rules:
526 return 0
528 interfaces = list(Interface.objects.filter(device=device, module=None))
529 if not interfaces:
530 return 0
532 total = 0
533 renamed_pks: set[int] = set()
534 for rule in rules:
535 for iface in interfaces:
536 if _try_rename_device_interface(rule, iface, vc_position, device, renamed_pks):
537 total += 1
539 return total
542def _get_raw_interface_names(module):
543 """Return the original interface names NetBox assigned from templates.
545 Prefetches module relationships to avoid per-template queries when
546 InterfaceTemplate.resolve_name() dereferences the module bay chain.
547 """
548 from dcim.models import InterfaceTemplate, Module
550 module_fresh = Module.objects.select_related(
551 "module_bay",
552 "module_bay__parent",
553 "module_bay__module",
554 "module_bay__module__module_bay",
555 "module_bay__module__module_bay__parent",
556 "module_bay__module__module_bay__module",
557 ).get(pk=module.pk)
558 templates = InterfaceTemplate.objects.filter(module_type=module_fresh.module_type)
559 return {tmpl.resolve_name(module_fresh) for tmpl in templates}
562def _flag_rule_potentially_deprecated(rule):
563 """Tag a rule as 'potentially-deprecated' when its rename is a no-op.
565 Called from apply_interface_name_rules when a matching rule produces no
566 renames because NetBox already generates the correct interface names. This
567 may indicate the rule is no longer needed (e.g. after a NetBox upgrade that
568 improved template resolution), or only needed for a subset of module types.
570 Adds a NetBox Tag 'potentially-deprecated' so the rule is visually flagged
571 in the UI for operator review. Failures are logged but never re-raised so
572 the install path is not disrupted.
573 """
574 try:
575 from extras.models import Tag
577 tag, _ = Tag.objects.get_or_create(
578 slug="potentially-deprecated",
579 defaults={"name": "potentially-deprecated", "color": "ffc107"},
580 )
581 rule.tags.add(tag)
582 logger.info(
583 "Rule '%s' flagged as potentially-deprecated: NetBox already generates the correct interface names.",
584 rule,
585 )
586 except Exception:
587 logger.exception("Failed to flag rule '%s' as potentially-deprecated.", rule)
590def _scope_ids(parent_module_type, device_type, platform):
591 """Map the (parent_module_type, device_type, platform) scope objects to their FK ids.
593 ``None`` (no constraint) maps to ``None`` so it compares equal to a rule's unset scope FK.
594 Centralises the ``x.pk if x is not None else None`` coalescing used by both match tiers and
595 the memo key.
596 """
597 return (
598 parent_module_type.pk if parent_module_type is not None else None,
599 device_type.pk if device_type is not None else None,
600 platform.pk if platform is not None else None,
601 )
604def _rule_scope_matches(rule, scope_ids):
605 """Return True when *rule*'s (parent_module_type, device_type, platform) FKs equal *scope_ids*."""
606 pmt_id, dt_id, pl_id = scope_ids
607 return rule.parent_module_type_id == pmt_id and rule.device_type_id == dt_id and rule.platform_id == pl_id
610def _build_candidates(parent_module_type, device_type, platform) -> list:
611 """Build ordered list of (pmt, dt, pl) tuples from most to least specific.
613 Each argument expands to ``[value, None]`` when provided, or ``[None]``
614 when already absent. Deduplication ensures no key appears twice (which
615 would happen when multiple inputs are None).
616 """
617 seen: set = set()
618 candidates = []
619 pmt_opts = [parent_module_type, None] if parent_module_type else [None]
620 dt_opts = [device_type, None] if device_type else [None]
621 pl_opts = [platform, None] if platform else [None]
622 for pmt in pmt_opts:
623 for dt in dt_opts:
624 for pl in pl_opts:
625 key = (pmt, dt, pl)
626 if key not in seen: 626 ↛ 624line 626 didn't jump to line 624 because the condition on line 626 was always true
627 seen.add(key)
628 candidates.append(key)
629 return candidates
632def _find_exact_match(module_type, candidates, exact_rules=None):
633 """Tier 1: return the first enabled exact-FK rule in specificity order, or None.
635 ``exact_rules`` is the preloaded, ``(module_type__model, pk)``-ordered enabled
636 exact-rule set (the hot path passes it to avoid a DB query per call); when omitted
637 it is loaded on demand so direct callers keep working.
638 """
639 if exact_rules is None:
640 exact_rules, _, _ = _get_enabled_rules()
642 # module_type fixed below → the (module_type__model, pk) ordering reduces to pk, so the
643 # first matching rule equals the previous ``.filter(...).first()``.
644 scoped = [r for r in exact_rules if r.module_type_id == module_type.pk]
645 for candidate in candidates:
646 scope_ids = _scope_ids(*candidate)
647 for rule in scoped:
648 if _rule_scope_matches(rule, scope_ids):
649 return rule
650 return None
653def _find_regex_match(model_name: str, candidates, regex_rules=None):
654 """Tier 2: return the first enabled regex rule whose pattern fullmatches *model_name*, or None.
656 Tries candidates in specificity order; within each level longer patterns are tried first
657 (more specific). ``regex_rules`` is the preloaded ``(compiled_pattern, rule)`` set — pre-sorted
658 by ``(-pattern length, pk)`` with each pattern compiled once (a None compile is an invalid
659 pattern, silently skipped); loaded on demand when omitted.
660 """
661 if regex_rules is None:
662 _, regex_rules, _ = _get_enabled_rules()
664 for candidate in candidates:
665 scope_ids = _scope_ids(*candidate)
666 for compiled, rule in regex_rules:
667 if compiled is not None and _rule_scope_matches(rule, scope_ids) and compiled.fullmatch(model_name):
668 return rule
669 return None
672def find_matching_rule(module_type, parent_module_type, device_type, platform=None):
673 """Find the most specific InterfaceNameRule matching the context.
675 Uses a two-tier strategy:
676 Tier 1 — Exact FK match (priority order, most specific first):
677 Iterates all combinations of (parent_module_type, device_type, platform)
678 from fully-constrained to fully-unconstrained (None = any).
679 Tier 2 — Regex pattern match (same priority order, longer patterns first):
680 Same specificity cascade, but module_type_pattern is matched via
681 re.fullmatch() against module_type.model. Patterns are iterated
682 from longest to shortest to prefer more specific patterns.
684 The enabled rule set is loaded once and matched in memory, and the per-context
685 result is memoized for the current rule-set version, so repeated calls (e.g. one
686 per module row in a module-sync render) don't re-query the database.
688 Returns the first matching rule, or None if no rule matches.
689 """
690 if module_type is None:
691 # Module rules are always keyed on a module type; both tiers dereference it
692 # (module_type.pk / .model), so there is nothing to match without one.
693 return None
695 exact_rules, regex_rules, memo = _get_enabled_rules()
696 # The regex tier matches against module_type.model (a live string), so the memo must key on
697 # it too — otherwise a ModuleType.model rename (same pk) would return a stale regex result.
698 sig = (module_type.pk, module_type.model, *_scope_ids(parent_module_type, device_type, platform))
699 # One atomic lookup, not `if sig in memo: return memo[sig]`: another thread sharing this per-version
700 # memo can clear it at the cap between a membership test and the subscript, raising KeyError.
701 cached = memo.get(sig, _MEMO_MISS)
702 if cached is not _MEMO_MISS:
703 return cached
705 candidates = _build_candidates(parent_module_type, device_type, platform)
706 result = _find_exact_match(module_type, candidates, exact_rules) or _find_regex_match(
707 module_type.model, candidates, regex_rules
708 )
709 if len(memo) >= _MEMO_MAX:
710 memo.clear() # bound per-version memory; entries are rebuilt lazily on the next miss
711 memo[sig] = result
712 return result
715def _extract_trailing_digits(s: str) -> str:
716 r"""Return the trailing digit run of *s* without regex backtracking.
718 Pure O(n) string scan — eliminates the polynomial backtracking risk that
719 arises from using ``re.search(r"(\d+)$", ...)`` on strings ending in a
720 non-digit character (e.g. ``"1" * n + "x"`` would cause O(n²) steps).
722 Returns an empty string when *s* has no trailing digits.
723 """
724 i = len(s)
725 while i > 0 and s[i - 1].isdigit():
726 i -= 1
727 return s[i:]
730def _resolve_bay_position(module_bay):
731 """Return (bay_position, bay_position_num) from a module bay's position field.
733 Handles template expressions like ``{module}`` by extracting the trailing
734 digit from the bay name. Falls back to ``"0"`` if no digit is found.
735 """
736 bay_position = module_bay.position or "0"
737 if bay_position.startswith("{"):
738 digits = _extract_trailing_digits(module_bay.name)
739 bay_position = digits if digits else "0"
740 digits = _extract_trailing_digits(bay_position)
741 bay_position_num = digits if digits else "0"
742 return bay_position, bay_position_num
745def _resolve_slot(module_bay, bay_position_num, parent_bay_position):
746 """Return the ``slot`` variable from the module bay hierarchy.
748 When the bay has a parent bay, slot comes from the parent (or grandparent
749 when two levels of nesting exist). When the bay belongs to an installed
750 module with its own bay, slot comes from that module's bay position.
751 Falls back to ``bay_position_num``.
752 """
753 if module_bay.parent:
754 parent_bay = module_bay.parent
755 if parent_bay.parent and hasattr(parent_bay.parent, "installed_module"):
756 return parent_bay.parent.position or parent_bay_position
757 return parent_bay_position
758 if hasattr(module_bay, "module") and module_bay.module: 758 ↛ 759line 758 didn't jump to line 759 because the condition on line 758 was never true
759 owner_module = module_bay.module
760 if hasattr(owner_module, "module_bay") and owner_module.module_bay:
761 return owner_module.module_bay.position or bay_position_num
762 return bay_position_num
765def build_variables(module_bay, device=None):
766 """Build template variable dict from a module bay's position context.
768 Extracts numeric and raw position values from the bay and its parent chain,
769 producing the variables available for name_template substitution.
771 Returns a dict with keys: slot, bay_position, bay_position_num,
772 parent_bay_position, sfp_slot, and optionally vc_position.
774 ``vc_position`` is only injected when *device* is a Virtual Chassis member
775 (device.virtual_chassis_id is set). Templates using ``{vc_position}`` on a
776 non-VC device will raise ValueError during evaluation — this is intentional.
777 Note: Juniper VC positions start at 0, so 0 is a valid real-world value and
778 cannot be used as a "not in VC" sentinel.
779 """
780 bay_position, bay_position_num = _resolve_bay_position(module_bay)
782 parent_bay_position = "0"
783 if module_bay.parent:
784 parent_bay_position = module_bay.parent.position or "0"
786 slot = _resolve_slot(module_bay, bay_position_num, parent_bay_position)
788 result = {
789 "slot": slot,
790 "bay_position": bay_position,
791 "bay_position_num": bay_position_num,
792 "parent_bay_position": parent_bay_position,
793 "sfp_slot": bay_position_num,
794 }
795 if (
796 device is not None
797 and getattr(device, "virtual_chassis_id", None) is not None
798 and device.vc_position is not None
799 ):
800 result["vc_position"] = str(device.vc_position)
801 return result
804def _name_exists_on_device(device, name, exclude_pk=None):
805 """Return True if another interface on *device* already uses *name*.
807 Pre-checks the per-device interface-name uniqueness NetBox enforces so a
808 rename/create that would collide is skipped cleanly instead of raising
809 mid-transaction. (VC-wide uniqueness is not pre-checked here; full_clean()
810 remains the authoritative validator for that rarer cross-member case.)
811 """
812 from dcim.models import Interface
814 qs = Interface.objects.filter(device=device, name=name)
815 if exclude_pk is not None:
816 qs = qs.exclude(pk=exclude_pk)
817 return qs.exists()
820def _record_conflict(conflicts, device, current_name, attempted_name, interface_pk=None):
821 """Log a name collision at WARNING and append it to *conflicts* when collecting.
823 Collisions are expected during automatic renaming (module install, type
824 change, VC change) when the computed name is already taken on the device;
825 they must never abort the batch, so callers skip the rename and carry on.
826 """
827 logger.warning(
828 "Interface name %r already exists on device %s — skipping rename of %r → %r",
829 attempted_name,
830 device,
831 current_name,
832 attempted_name,
833 )
834 if conflicts is not None:
835 conflicts.append(
836 {
837 "device": str(device),
838 "current_name": current_name,
839 "attempted_name": attempted_name,
840 "interface_pk": interface_pk,
841 }
842 )
845def _rename_in_place(iface, new_name, device, conflicts):
846 """Rename *iface* to *new_name*; return 1 if renamed, 0 if no-op or collision."""
847 if new_name == iface.name:
848 return 0
849 if _name_exists_on_device(device, new_name, exclude_pk=iface.pk):
850 _record_conflict(conflicts, device, iface.name, new_name, iface.pk)
851 return 0
852 iface.name = new_name
853 iface.full_clean()
854 iface.save()
855 return 1
858def _create_channel(iface, module, new_name, device, conflicts):
859 """Create a breakout channel interface *new_name*; return 1 if created, else 0.
861 Silently skips when this module already has the channel (idempotent
862 re-apply); records a conflict when *new_name* is taken by a different
863 interface on the device.
864 """
865 from dcim.models import Interface
867 if Interface.objects.filter(module=module, name=new_name).exists():
868 return 0 # idempotent: channel already created on this module
869 if _name_exists_on_device(device, new_name):
870 _record_conflict(conflicts, device, iface.name, new_name, iface.pk)
871 return 0
872 breakout_iface = Interface(
873 device=device,
874 module=module,
875 name=new_name,
876 type=iface.type,
877 enabled=iface.enabled,
878 )
879 breakout_iface.full_clean()
880 breakout_iface.save()
881 return 1
884def _apply_rule_to_interface(rule, iface, variables, module, conflicts=None):
885 """Apply a single rule to an interface, handling breakout channels.
887 All saves are wrapped in a transaction so a failure mid-breakout rolls
888 back any partially created interfaces. A computed name that already exists
889 on the device is skipped (logged, and recorded in *conflicts* when a list
890 is passed) instead of raising — so automatic renaming (module install,
891 module-type change, VC change) never aborts the rest of the batch on a
892 name collision.
894 Returns the number of interfaces renamed/created.
895 """
896 count = 0
897 device = module.device
899 with transaction.atomic():
900 if rule.channel_count > 0:
901 # Breakout: rename base interface and create additional channel interfaces
902 for ch in range(rule.channel_count):
903 variables["channel"] = str(rule.channel_start + ch)
904 new_name = evaluate_name_template(rule.name_template, variables)
905 if ch == 0:
906 count += _rename_in_place(iface, new_name, device, conflicts)
907 else:
908 count += _create_channel(iface, module, new_name, device, conflicts)
909 else:
910 # Simple rename (converter offset, platform naming, etc.)
911 new_name = evaluate_name_template(rule.name_template, variables)
912 count += _rename_in_place(iface, new_name, device, conflicts)
914 return count
917def _find_channel_base(rule, ifaces, variables):
918 """Find the best 'base' interface for a channel rule on a single module.
920 Prefers an interface whose current name already equals the expected ch=0 name
921 (i.e. it has already been renamed to channel 0 and is safe to re-process).
922 Falls back to the first interface (alphabetically) so that on first apply,
923 the template-created base interface becomes channel 0.
925 This ensures apply_rule_to_existing / find_interfaces_for_rule call
926 _apply_rule_to_interface exactly ONCE per module for channel rules, preventing
927 duplicate-name IntegrityErrors when channels already exist.
928 """
929 if not ifaces:
930 return None
931 for iface in ifaces:
932 vars_copy = dict(variables)
933 vars_copy["base"] = iface.name
934 vars_copy["channel"] = str(rule.channel_start) # ch=0
935 try:
936 ch0_name = evaluate_name_template(rule.name_template, vars_copy)
937 if iface.name == ch0_name:
938 return iface
939 except ValueError:
940 pass
941 return ifaces[0]
944def _matching_moduletype_pks(module_type_pattern):
945 """Return PKs of ModuleTypes whose model name matches the given regex pattern.
947 Raises ValueError for invalid regex patterns, mirroring evaluate_name_template's
948 error-handling convention so callers can treat both as ValueError.
949 """
950 from dcim.models import ModuleType
952 try:
953 compiled = re.compile(module_type_pattern)
954 except re.error as exc:
955 raise ValueError(f"Invalid module_type_pattern regex '{module_type_pattern}': {exc}") from exc
956 return [mt.pk for mt in ModuleType.objects.only("pk", "model") if compiled.fullmatch(mt.model)]
959def has_applicable_interfaces(rule) -> bool:
960 """Check whether applying this rule right now would rename at least one interface.
962 Calls find_interfaces_for_rule(limit=1) to determine if any currently installed
963 interface would receive a new name. Returns False when:
964 - no matching modules/interfaces are installed, OR
965 - all matching interfaces are already correctly named.
967 This is more expensive than a plain EXISTS query but ensures the Applicable
968 column in the Apply Rules list accurately reflects "would something change?"
969 rather than the misleading "do interfaces exist?".
970 """
971 try:
972 results, _ = find_interfaces_for_rule(rule, limit=1)
973 return len(results) > 0
974 except (ValueError, re.error):
975 return False
978def _build_module_qs(rule):
979 """Return a Module queryset filtered to the rule's scope (module type, parent, device, platform).
981 Shared by ``find_interfaces_for_rule`` and ``apply_rule_to_existing`` to avoid
982 duplicating the filtering logic.
983 """
984 from dcim.models import Module
986 if rule.module_type_is_regex:
987 qs = Module.objects.filter(module_type__in=_matching_moduletype_pks(rule.module_type_pattern))
988 else:
989 qs = Module.objects.filter(module_type=rule.module_type)
990 if rule.parent_module_type:
991 qs = qs.filter(module_bay__parent__installed_module__module_type=rule.parent_module_type)
992 if rule.device_type:
993 qs = qs.filter(device__device_type=rule.device_type)
994 if rule.platform:
995 qs = qs.filter(device__platform=rule.platform)
996 return qs
999def _evaluate_plain_interface(rule, module, iface, variables) -> dict | None:
1000 """Return a result dict if *iface* would be renamed by *rule*, else None."""
1001 vars_copy = {**variables, "base": iface.name}
1002 try:
1003 new_name = evaluate_name_template(rule.name_template, vars_copy)
1004 except ValueError as exc:
1005 new_name = f"<error: {exc}>"
1006 if new_name == iface.name:
1007 return None
1008 return {"module": module, "interface": iface, "current_name": iface.name, "new_names": [new_name]}
1011def _channel_rule_entry(rule, module, ifaces, variables) -> dict | None:
1012 """Return a result dict if the channel rule would change any name for this module, else None."""
1013 base_iface = _find_channel_base(rule, ifaces, variables)
1014 if base_iface is None: 1014 ↛ 1015line 1014 didn't jump to line 1015 because the condition on line 1014 was never true
1015 return None
1016 vars_copy = {**variables, "base": base_iface.name}
1017 expected_names = []
1018 try:
1019 for ch in range(rule.channel_count):
1020 expected_names.append(
1021 evaluate_name_template(rule.name_template, {**vars_copy, "channel": str(rule.channel_start + ch)})
1022 )
1023 except ValueError as exc:
1024 expected_names = [f"<error: {exc}>"]
1025 existing_names = {i.name for i in ifaces}
1026 # Report if any channel name is missing or the base itself needs renaming
1027 if any(n not in existing_names for n in expected_names) or (
1028 expected_names and expected_names[0] != base_iface.name
1029 ):
1030 return {"module": module, "interface": base_iface, "current_name": base_iface.name, "new_names": expected_names}
1031 return None
1034def _count_remaining_interfaces(module_qs, processed_pks) -> int:
1035 """Count interfaces in modules not yet visited during a find_interfaces_for_rule scan."""
1036 from dcim.models import Interface
1038 return Interface.objects.filter(module__in=module_qs.exclude(pk__in=processed_pks)).count()
1041def _process_channel_module(rule, module, ifaces, variables, limit, results, module_qs, processed_pks):
1042 """Process one module for a channel rule. Returns (checked_count, should_stop)."""
1043 checked = len(ifaces)
1044 if not ifaces:
1045 return checked, False
1046 entry = _channel_rule_entry(rule, module, ifaces, variables)
1047 if entry:
1048 results.append(entry)
1049 if limit is not None and len(results) >= limit:
1050 return checked + _count_remaining_interfaces(module_qs, processed_pks), True
1051 return checked, False
1054def _process_plain_module(rule, module, ifaces, variables, limit, results, module_qs, processed_pks):
1055 """Process one module for a plain (non-channel) rule. Returns (checked_count, should_stop)."""
1056 checked = 0
1057 for iface_idx, iface in enumerate(ifaces):
1058 checked += 1
1059 entry = _evaluate_plain_interface(rule, module, iface, variables)
1060 if entry:
1061 results.append(entry)
1062 if limit is not None and len(results) >= limit:
1063 checked += len(ifaces) - (iface_idx + 1)
1064 checked += _count_remaining_interfaces(module_qs, processed_pks)
1065 return checked, True
1066 return checked, False
1069def find_interfaces_for_rule(rule, limit=None):
1070 """Find interfaces that would be renamed by applying the given rule retroactively.
1072 Searches for all Module instances matching the rule's criteria and computes
1073 what their interfaces would be renamed to.
1075 Returns a tuple ``(results, total_checked)`` where *results* is a list of dicts::
1077 {
1078 "module": Module instance,
1079 "interface": Interface instance,
1080 "current_name": str,
1081 "new_names": list[str], # one entry per channel, or single-element
1082 }
1084 Only includes entries where at least one new_name differs from current_name.
1085 If *limit* is set the list is truncated after that many changed entries, but
1086 *total_checked* always reflects the full count of interfaces examined.
1087 """
1088 from collections import defaultdict
1090 from dcim.models import Interface
1092 module_qs = _build_module_qs(rule).select_related(
1093 "module_type",
1094 "device",
1095 "device__device_type",
1096 "device__platform",
1097 "device__virtual_chassis",
1098 "module_bay",
1099 "module_bay__parent",
1100 )
1101 process_fn = _process_channel_module if rule.channel_count > 0 else _process_plain_module
1103 # Batch-load all interfaces for matching modules to avoid N+1 queries.
1104 ifaces_by_module = defaultdict(list)
1105 for iface in Interface.objects.filter(module__in=module_qs).order_by("module_id", "name"):
1106 ifaces_by_module[iface.module_id].append(iface)
1108 processed_pks = set()
1109 results = []
1110 total_checked = 0
1111 for module in module_qs:
1112 processed_pks.add(module.pk)
1113 variables = build_variables(module.module_bay, device=module.device)
1114 ifaces = ifaces_by_module.get(module.pk, [])
1115 checked, stop = process_fn(rule, module, ifaces, variables, limit, results, module_qs, processed_pks)
1116 total_checked += checked
1117 if stop:
1118 return results, total_checked
1120 return results, total_checked
1123def _apply_channel_rule_to_module(rule, module, ifaces, variables, id_set, conflicts):
1124 """Apply a channel rule to one module via its base interface; return the rename count.
1126 A channel rule is processed ONCE per module (not per interface) so existing
1127 channel names are not re-created. An unexpected failure (e.g. a save race)
1128 is logged and skipped so it never aborts the surrounding batch.
1129 """
1130 if not ifaces:
1131 return 0
1132 base_iface = _find_channel_base(rule, ifaces, variables)
1133 if id_set is not None and base_iface.pk not in id_set:
1134 return 0
1135 vars_copy = dict(variables)
1136 vars_copy["base"] = base_iface.name
1137 try:
1138 return _apply_rule_to_interface(rule, base_iface, vars_copy, module, conflicts=conflicts)
1139 except (ValueError, ValidationError, IntegrityError):
1140 logger.exception(
1141 "Failed to apply channel rule '%s' to module '%s' (id=%s); skipping.",
1142 rule,
1143 module,
1144 module.pk,
1145 )
1146 return 0
1149def _apply_plain_rule_to_module(rule, module, ifaces, variables, id_set, conflicts):
1150 """Apply a non-channel rule to each selected interface on one module; return the rename count.
1152 Each interface is independent: an unexpected failure on one is logged and
1153 skipped so the rest of the module (and batch) still process.
1154 """
1155 count = 0
1156 for iface in ifaces:
1157 if id_set is not None and iface.pk not in id_set:
1158 continue
1159 vars_copy = dict(variables)
1160 vars_copy["base"] = iface.name
1161 try:
1162 count += _apply_rule_to_interface(rule, iface, vars_copy, module, conflicts=conflicts)
1163 except (ValueError, ValidationError, IntegrityError):
1164 logger.exception(
1165 "Failed to apply rule '%s' to interface '%s' (id=%s); skipping.",
1166 rule,
1167 iface.name,
1168 iface.pk,
1169 )
1170 return count
1173def apply_rule_to_existing(rule, limit=None, interface_ids=None, conflicts=None):
1174 """Apply a rule retroactively to all matching installed modules.
1176 Unlike apply_interface_name_rules(), this does not skip already-renamed
1177 interfaces — it re-evaluates every interface on each matching module.
1179 For channel rules (channel_count > 0), each module is processed as a single
1180 unit using _find_channel_base() to pick the base interface. Calling
1181 _apply_rule_to_interface for every interface in the module would produce
1182 duplicate-name IntegrityErrors when channel interfaces already exist.
1184 If *interface_ids* is provided (list/set of Interface PKs), only those
1185 interfaces are processed; all others are skipped. For channel rules the
1186 base interface PK is used as the selector. An empty *interface_ids*
1187 collection returns 0 immediately without touching the database.
1189 If *conflicts* is a list, each interface skipped because its target name is
1190 already taken on the device is appended to it (and logged) — letting the
1191 caller report how many renames were dropped. Collisions never raise.
1193 Returns the number of interfaces renamed/created.
1194 """
1195 from collections import defaultdict
1197 from dcim.models import Interface
1199 id_set = frozenset(interface_ids) if interface_ids is not None else None
1200 if id_set is not None and not id_set:
1201 return 0
1203 if not rule.enabled:
1204 return 0
1206 module_qs = _build_module_qs(rule)
1208 # Batch-load interfaces to avoid N+1 queries in the module loop.
1209 ifaces_by_module = defaultdict(list)
1210 for iface in Interface.objects.filter(module__in=module_qs).order_by("module_id", "name"):
1211 ifaces_by_module[iface.module_id].append(iface)
1213 count = 0
1214 for module in module_qs.select_related("module_bay", "module_type", "device", "device__virtual_chassis"):
1215 variables = build_variables(module.module_bay, device=module.device)
1216 ifaces = ifaces_by_module.get(module.pk, [])
1218 if rule.channel_count > 0:
1219 count += _apply_channel_rule_to_module(rule, module, ifaces, variables, id_set, conflicts)
1220 else:
1221 count += _apply_plain_rule_to_module(rule, module, ifaces, variables, id_set, conflicts)
1223 if limit is not None and count >= limit:
1224 return count
1226 return count
1229def evaluate_name_template(template: str, variables: dict) -> str:
1230 """Evaluate a name template with variable substitution and safe arithmetic.
1232 Supports templates like:
1233 "GigabitEthernet{slot}/{8 + ({parent_bay_position} - 1) * 2 + {sfp_slot}}"
1235 Variables are substituted first, then any brace-enclosed expression
1236 containing arithmetic operators is safely evaluated via AST. True division
1237 (/) is not allowed — use floor division (//) instead. Results are cast to
1238 int to ensure interface names are always whole numbers.
1239 """
1240 # First pass: substitute all simple variables
1241 result = template
1242 for key, value in variables.items():
1243 result = result.replace(f"{{{key}}}", str(value))
1245 # Second pass: evaluate any remaining brace-enclosed arithmetic expressions
1246 def _eval_expr(match):
1247 expr = match.group(1).strip()
1248 # Allow digits, arithmetic operators (excluding lone /), parens, whitespace.
1249 # Negative lookahead disallows a single / that is not part of //.
1250 if not re.match(r"^(?!.*(?<!/)/(?!/))[\d\s\+\-\*\(\/\)]+$", expr):
1251 raise ValueError(f"Unsafe expression in name template: {expr}")
1252 try:
1253 node = ast.parse(expr, mode="eval")
1254 for child in ast.walk(node):
1255 if not isinstance(
1256 child,
1257 (
1258 ast.Expression,
1259 ast.BinOp,
1260 ast.UnaryOp,
1261 ast.Constant,
1262 ast.Add,
1263 ast.Sub,
1264 ast.Mult,
1265 ast.FloorDiv,
1266 ast.USub,
1267 ast.UAdd,
1268 ),
1269 ):
1270 raise ValueError(f"Unsafe AST node in expression: {type(child).__name__}")
1271 return str(int(eval(compile(node, "<template>", "eval")))) # noqa: S307
1272 except (SyntaxError, TypeError) as e:
1273 raise ValueError(f"Invalid arithmetic expression '{expr}': {e}") from e
1275 return re.sub(r"\{([^}]+)\}", _eval_expr, result)