Skip to content

nautobot.apps.utils

Nautobot utility functions.

nautobot.apps.utils.ChangeLoggedModelsQuery

Bases: FeaturedQueryMixin

Helper class to get ContentType for models that implements the to_objectchange method for change logging.

Source code in nautobot/extras/utils.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
@deconstructible
class ChangeLoggedModelsQuery(FeaturedQueryMixin):
    """
    Helper class to get ContentType for models that implements the to_objectchange method for change logging.
    """

    def list_subclasses(self):
        """
        Return a list of classes that implement the to_objectchange method
        """
        return [_class for _class in apps.get_models() if hasattr(_class, "to_objectchange")]

list_subclasses()

Return a list of classes that implement the to_objectchange method

Source code in nautobot/extras/utils.py 
100
101
102
103
104
def list_subclasses(self):
    """
    Return a list of classes that implement the to_objectchange method
    """
    return [_class for _class in apps.get_models() if hasattr(_class, "to_objectchange")]

nautobot.apps.utils.FeatureQuery

Helper class that delays evaluation of the registry contents for the functionality store until it has been populated.

Source code in nautobot/extras/utils.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
@deconstructible
class FeatureQuery:
    """
    Helper class that delays evaluation of the registry contents for the functionality store
    until it has been populated.
    """

    def __init__(self, feature):
        self.feature = feature

    def __call__(self):
        return self.get_query()

    def get_query(self):
        """
        Given an extras feature, return a Q object for content type lookup
        """

        # The `populate_model_features_registry` function is called in the `FeatureQuery().get_query` method instead of
        # `ExtrasConfig.ready` because `FeatureQuery().get_query` is called before `ExtrasConfig.ready`.
        # This is because `FeatureQuery` is a helper class used in `Forms` and `Serializers` that are called during the
        # initialization of the application, before `ExtrasConfig.ready` is called.
        # Calling `populate_model_features_registry` in `ExtrasConfig.ready` would lead to an outdated `model_features`
        # `registry` record being used by `FeatureQuery`.

        populate_model_features_registry()
        query = Q()
        for app_label, models in self.as_dict():
            query |= Q(app_label=app_label, model__in=models)

        return query

    def as_dict(self):
        """
        Given an extras feature, return a dict of app_label: [models] for content type lookup
        """
        return registry["model_features"][self.feature].items()

    def get_choices(self):
        """
        Given an extras feature, return a list of 2-tuple of `(model_label, pk)`
        suitable for use as `choices` on a choice field:

            >>> FeatureQuery('statuses').get_choices()
            [('dcim.device', 13), ('dcim.rack', 34)]
        """
        return [(f"{ct.app_label}.{ct.model}", ct.pk) for ct in ContentType.objects.filter(self.get_query())]

    def list_subclasses(self):
        """Return a list of model classes that declare this feature."""
        return [ct.model_class() for ct in ContentType.objects.filter(self.get_query())]

as_dict()

Given an extras feature, return a dict of app_label: [models] for content type lookup

Source code in nautobot/extras/utils.py 
139
140
141
142
143
def as_dict(self):
    """
    Given an extras feature, return a dict of app_label: [models] for content type lookup
    """
    return registry["model_features"][self.feature].items()

get_choices()

Given an extras feature, return a list of 2-tuple of (model_label, pk) suitable for use as choices on a choice field:

>>> FeatureQuery('statuses').get_choices()
[('dcim.device', 13), ('dcim.rack', 34)]
Source code in nautobot/extras/utils.py 
145
146
147
148
149
150
151
152
153
def get_choices(self):
    """
    Given an extras feature, return a list of 2-tuple of `(model_label, pk)`
    suitable for use as `choices` on a choice field:

        >>> FeatureQuery('statuses').get_choices()
        [('dcim.device', 13), ('dcim.rack', 34)]
    """
    return [(f"{ct.app_label}.{ct.model}", ct.pk) for ct in ContentType.objects.filter(self.get_query())]

get_query()

Given an extras feature, return a Q object for content type lookup

Source code in nautobot/extras/utils.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def get_query(self):
    """
    Given an extras feature, return a Q object for content type lookup
    """

    # The `populate_model_features_registry` function is called in the `FeatureQuery().get_query` method instead of
    # `ExtrasConfig.ready` because `FeatureQuery().get_query` is called before `ExtrasConfig.ready`.
    # This is because `FeatureQuery` is a helper class used in `Forms` and `Serializers` that are called during the
    # initialization of the application, before `ExtrasConfig.ready` is called.
    # Calling `populate_model_features_registry` in `ExtrasConfig.ready` would lead to an outdated `model_features`
    # `registry` record being used by `FeatureQuery`.

    populate_model_features_registry()
    query = Q()
    for app_label, models in self.as_dict():
        query |= Q(app_label=app_label, model__in=models)

    return query

list_subclasses()

Return a list of model classes that declare this feature.

Source code in nautobot/extras/utils.py 
155
156
157
def list_subclasses(self):
    """Return a list of model classes that declare this feature."""
    return [ct.model_class() for ct in ContentType.objects.filter(self.get_query())]

nautobot.apps.utils.FeaturedQueryMixin

Mixin class that gets a list of featured models.

Source code in nautobot/extras/utils.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
@deconstructible
class FeaturedQueryMixin:
    """Mixin class that gets a list of featured models."""

    def list_subclasses(self):
        """Return a list of classes that has implements this `name`."""
        raise NotImplementedError("list_subclasses is not implemented")

    def __call__(self):
        """
        Given an extras feature, return a Q object for content type lookup
        """
        query = Q()
        for model in self.list_subclasses():
            query |= Q(app_label=model._meta.app_label, model=model.__name__.lower())

        return query

    def as_queryset(self):
        return ContentType.objects.filter(self()).order_by("app_label", "model")

    def get_choices(self):
        return [(f"{ct.app_label}.{ct.model}", ct.pk) for ct in self.as_queryset()]

__call__()

Given an extras feature, return a Q object for content type lookup

Source code in nautobot/extras/utils.py 
77
78
79
80
81
82
83
84
85
def __call__(self):
    """
    Given an extras feature, return a Q object for content type lookup
    """
    query = Q()
    for model in self.list_subclasses():
        query |= Q(app_label=model._meta.app_label, model=model.__name__.lower())

    return query

list_subclasses()

Return a list of classes that has implements this name.

Source code in nautobot/extras/utils.py 
73
74
75
def list_subclasses(self):
    """Return a list of classes that has implements this `name`."""
    raise NotImplementedError("list_subclasses is not implemented")

nautobot.apps.utils.GitRepo

Source code in nautobot/core/utils/git.py 
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
class GitRepo:
    def __init__(self, path, url, clone_initially=True):
        """
        Ensure that we have a clone of the given remote Git repository URL at the given local directory path.

        Args:
            path (str): path to git repo
            url (str): git repo url
            clone_initially (bool): True if the repo needs to be cloned
        """
        if os.path.isdir(path) and os.path.isdir(os.path.join(path, ".git")):
            self.repo = Repo(path=path)
        elif clone_initially:
            # Don't log `url` as it may include authentication details.
            logger.debug("Cloning git repository to %s...", path)
            self.repo = Repo.clone_from(url, to_path=path, env=GIT_ENVIRONMENT)
        else:
            self.repo = Repo.init(path)
            self.repo.create_remote("origin", url=url)

        if url not in self.repo.remotes.origin.urls:
            self.repo.remotes.origin.set_url(url)

    @property
    def head(self):
        """Current checked out repository head commit."""
        return self.repo.head.commit.hexsha

    def fetch(self):
        with self.repo.git.custom_environment(**GIT_ENVIRONMENT):
            self.repo.remotes.origin.fetch()

    def checkout(self, branch, commit_hexsha=None):
        """
        Check out the given branch, and optionally the specified commit within that branch.

        Returns:
            (str, bool): commit_hexsha the repo contains now, whether any change occurred
        """
        # Short-circuit logic - do we already have this commit checked out?
        if commit_hexsha and commit_hexsha == self.head:
            logger.debug(f"Commit {commit_hexsha} is already checked out.")
            return (commit_hexsha, False)

        self.fetch()
        if commit_hexsha:
            # Sanity check - GitPython doesn't provide a handy API for this so we just call a raw Git command:
            # $ git branch origin/<branch> --remotes --contains <commit>
            # prints the branch name if it DOES contain the commit, and nothing if it DOES NOT contain the commit.
            # Since we did a `fetch` and not a `pull` above, we need to check for the commit in the remote origin
            # branch, not the local (not-yet-updated) branch.
            if branch not in self.repo.git.branch(f"origin/{branch}", "--remotes", "--contains", commit_hexsha):
                raise RuntimeError(f"Requested to check out commit `{commit_hexsha}`, but it's not in branch {branch}!")
            logger.info(f"Checking out commit `{commit_hexsha}` on branch `{branch}`...")
            self.repo.git.checkout(commit_hexsha)
            return (commit_hexsha, True)

        if branch in self.repo.heads:
            branch_head = self.repo.heads[branch]
        else:
            try:
                branch_head = self.repo.create_head(branch, self.repo.remotes.origin.refs[branch])
                branch_head.set_tracking_branch(self.repo.remotes.origin.refs[branch])
            except IndexError as git_error:
                logger.error(
                    "Branch %s does not exist at %s. %s", branch, list(self.repo.remotes.origin.urls)[0], git_error
                )
                raise BranchDoesNotExist(
                    f"Please create branch '{branch}' in upstream and try again."
                    f" If this is a new repo, please add a commit before syncing. {git_error}"
                )

        logger.info(f"Checking out latest commit on branch `{branch}`...")
        branch_head.checkout()
        # No specific commit hash was given, so make sure we get the latest from origin
        # We would use repo.remotes.origin.pull() here, but that will fail in the case where someone has
        # force-pushed to the upstream repo since the last time we did a pull. To be safe, we reset instead.
        self.repo.head.reset(f"origin/{branch}", index=True, working_tree=True)
        commit_hexsha = self.repo.head.reference.commit.hexsha
        logger.info(f"Latest commit on branch `{branch}` is `{commit_hexsha}`")
        return (commit_hexsha, True)

    def diff_remote(self, branch):
        logger.debug("Fetching from remote.")
        self.fetch()

        try:
            self.repo.remotes.origin.refs[branch]
        except IndexError as git_error:
            logger.error(
                "Branch %s does not exist at %s. %s", branch, list(self.repo.remotes.origin.urls)[0], git_error
            )
            raise BranchDoesNotExist(
                f"Please create branch '{branch}' in upstream and try again."
                f" If this is a new repo, please add a commit before syncing. {git_error}"
            )

        logger.debug("Getting diff between local branch and remote branch")
        diff = self.repo.git.diff("--name-status", f"origin/{branch}")
        if diff:  # if diff is not empty
            return convert_git_diff_log_to_list(diff)
        logger.debug("No Difference")
        return []

head property

Current checked out repository head commit.

__init__(path, url, clone_initially=True)

Ensure that we have a clone of the given remote Git repository URL at the given local directory path.

Parameters:

Name Type Description Default
path str

path to git repo

 required
url str

git repo url

 required
clone_initially bool

True if the repo needs to be cloned

 True
Source code in nautobot/core/utils/git.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def __init__(self, path, url, clone_initially=True):
    """
    Ensure that we have a clone of the given remote Git repository URL at the given local directory path.

    Args:
        path (str): path to git repo
        url (str): git repo url
        clone_initially (bool): True if the repo needs to be cloned
    """
    if os.path.isdir(path) and os.path.isdir(os.path.join(path, ".git")):
        self.repo = Repo(path=path)
    elif clone_initially:
        # Don't log `url` as it may include authentication details.
        logger.debug("Cloning git repository to %s...", path)
        self.repo = Repo.clone_from(url, to_path=path, env=GIT_ENVIRONMENT)
    else:
        self.repo = Repo.init(path)
        self.repo.create_remote("origin", url=url)

    if url not in self.repo.remotes.origin.urls:
        self.repo.remotes.origin.set_url(url)

checkout(branch, commit_hexsha=None)

Check out the given branch, and optionally the specified commit within that branch.

Returns:

Type Description
(str, bool)

commit_hexsha the repo contains now, whether any change occurred
Source code in nautobot/core/utils/git.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def checkout(self, branch, commit_hexsha=None):
    """
    Check out the given branch, and optionally the specified commit within that branch.

    Returns:
        (str, bool): commit_hexsha the repo contains now, whether any change occurred
    """
    # Short-circuit logic - do we already have this commit checked out?
    if commit_hexsha and commit_hexsha == self.head:
        logger.debug(f"Commit {commit_hexsha} is already checked out.")
        return (commit_hexsha, False)

    self.fetch()
    if commit_hexsha:
        # Sanity check - GitPython doesn't provide a handy API for this so we just call a raw Git command:
        # $ git branch origin/<branch> --remotes --contains <commit>
        # prints the branch name if it DOES contain the commit, and nothing if it DOES NOT contain the commit.
        # Since we did a `fetch` and not a `pull` above, we need to check for the commit in the remote origin
        # branch, not the local (not-yet-updated) branch.
        if branch not in self.repo.git.branch(f"origin/{branch}", "--remotes", "--contains", commit_hexsha):
            raise RuntimeError(f"Requested to check out commit `{commit_hexsha}`, but it's not in branch {branch}!")
        logger.info(f"Checking out commit `{commit_hexsha}` on branch `{branch}`...")
        self.repo.git.checkout(commit_hexsha)
        return (commit_hexsha, True)

    if branch in self.repo.heads:
        branch_head = self.repo.heads[branch]
    else:
        try:
            branch_head = self.repo.create_head(branch, self.repo.remotes.origin.refs[branch])
            branch_head.set_tracking_branch(self.repo.remotes.origin.refs[branch])
        except IndexError as git_error:
            logger.error(
                "Branch %s does not exist at %s. %s", branch, list(self.repo.remotes.origin.urls)[0], git_error
            )
            raise BranchDoesNotExist(
                f"Please create branch '{branch}' in upstream and try again."
                f" If this is a new repo, please add a commit before syncing. {git_error}"
            )

    logger.info(f"Checking out latest commit on branch `{branch}`...")
    branch_head.checkout()
    # No specific commit hash was given, so make sure we get the latest from origin
    # We would use repo.remotes.origin.pull() here, but that will fail in the case where someone has
    # force-pushed to the upstream repo since the last time we did a pull. To be safe, we reset instead.
    self.repo.head.reset(f"origin/{branch}", index=True, working_tree=True)
    commit_hexsha = self.repo.head.reference.commit.hexsha
    logger.info(f"Latest commit on branch `{branch}` is `{commit_hexsha}`")
    return (commit_hexsha, True)

nautobot.apps.utils.RoleModelsQuery

Bases: FeaturedQueryMixin

Helper class to get ContentType models that implements role.

Source code in nautobot/extras/utils.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
@deconstructible
class RoleModelsQuery(FeaturedQueryMixin):
    """
    Helper class to get ContentType models that implements role.
    """

    def list_subclasses(self):
        """
        Return a list of classes that implements roles e.g roles = ...
        """
        # Avoid circular imports
        from nautobot.extras.models.roles import RoleField

        model_classes = []
        for model_class in apps.get_models():
            if hasattr(model_class, "role") and isinstance(model_class._meta.get_field("role"), RoleField):
                model_classes.append(model_class)
        return model_classes

list_subclasses()

Return a list of classes that implements roles e.g roles = ...

Source code in nautobot/extras/utils.py 
187
188
189
190
191
192
193
194
195
196
197
198
def list_subclasses(self):
    """
    Return a list of classes that implements roles e.g roles = ...
    """
    # Avoid circular imports
    from nautobot.extras.models.roles import RoleField

    model_classes = []
    for model_class in apps.get_models():
        if hasattr(model_class, "role") and isinstance(model_class._meta.get_field("role"), RoleField):
            model_classes.append(model_class)
    return model_classes

nautobot.apps.utils.TaggableClassesQuery

Bases: FeaturedQueryMixin

Helper class to get ContentType models that implements tags(TagsField)

Source code in nautobot/extras/utils.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
@deconstructible
class TaggableClassesQuery(FeaturedQueryMixin):
    """
    Helper class to get ContentType models that implements tags(TagsField)
    """

    def list_subclasses(self):
        """
        Return a list of classes that has implements tags e.g tags = TagsField(...)
        """
        return [
            _class
            for _class in apps.get_models()
            if (
                hasattr(_class, "tags")
                and isinstance(_class.tags, TagsManager)
                and ".tests." not in _class.__module__  # avoid leakage from nautobot.core.tests.test_filters
            )
        ]

list_subclasses()

Return a list of classes that has implements tags e.g tags = TagsField(...)

Source code in nautobot/extras/utils.py 
166
167
168
169
170
171
172
173
174
175
176
177
178
def list_subclasses(self):
    """
    Return a list of classes that has implements tags e.g tags = TagsField(...)
    """
    return [
        _class
        for _class in apps.get_models()
        if (
            hasattr(_class, "tags")
            and isinstance(_class.tags, TagsManager)
            and ".tests." not in _class.__module__  # avoid leakage from nautobot.core.tests.test_filters
        )
    ]

nautobot.apps.utils.build_lookup_label(field_name, _verbose_name)

Return lookup expr with its verbose name

Parameters:

Name Type Description Default
field_name str

Field name e.g name__iew

 required
_verbose_name str

The verbose name for the lookup expr which is suffixed to the field name e.g iew -> iendswith

 required

Examples:

>>> build_lookup_label("name__iew", "iendswith")
>>> "ends-with (iew)"
Source code in nautobot/core/utils/filtering.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def build_lookup_label(field_name, _verbose_name):
    """
    Return lookup expr with its verbose name

    Args:
        field_name (str): Field name e.g name__iew
        _verbose_name (str): The verbose name for the lookup expr which is suffixed to the field name e.g iew -> iendswith

    Examples:
        >>> build_lookup_label("name__iew", "iendswith")
        >>> "ends-with (iew)"
    """
    verbose_name = verbose_lookup_expr(_verbose_name) or "exact"
    label = ""
    search = CONTAINS_LOOKUP_EXPR_RE.search(field_name)
    if search:
        label = f" ({search.group()})"

    verbose_name = "not " + verbose_name if label.startswith(" (n") else verbose_name

    return verbose_name + label

nautobot.apps.utils.check_if_key_is_graphql_safe(model_name, key, field_name='key')

Helper method to check if a key field is Python/GraphQL safe. Used in CustomField, ComputedField and Relationship models.

Source code in nautobot/extras/utils.py 
475
476
477
478
479
480
481
482
483
484
485
486
def check_if_key_is_graphql_safe(model_name, key, field_name="key"):
    """
    Helper method to check if a key field is Python/GraphQL safe.
    Used in CustomField, ComputedField and Relationship models.
    """
    graphql_safe_pattern = re.compile("[_A-Za-z][_0-9A-Za-z]*")
    if not graphql_safe_pattern.fullmatch(key):
        raise ValidationError(
            {
                f"{field_name}": f"This {field_name} is not Python/GraphQL safe. Please do not start the {field_name} with a digit and do not use hyphens or whitespace"
            }
        )

nautobot.apps.utils.class_deprecated(message)

Decorator to mark a class as deprecated with a custom message about what to do instead of subclassing it.

Source code in nautobot/core/utils/deprecation.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def class_deprecated(message):
    """Decorator to mark a class as deprecated with a custom message about what to do instead of subclassing it."""

    def decorate(cls):
        def init_subclass(new_subclass):
            # Walk the stack up to the class declaration in question.
            stacklevel = 0
            for fs in reversed(traceback.extract_stack()):
                stacklevel += 1
                if new_subclass.__name__ in fs.line:
                    break
            else:
                stacklevel = 1
            warnings.warn(
                f"Class {cls.__name__} is deprecated, and will be removed in a future Nautobot release. "
                f"Instead of deriving {new_subclass.__name__} from {cls.__name__}, {message}.",
                DeprecationWarning,
                stacklevel=stacklevel,
            )
            if LOG_DEPRECATION_WARNINGS:
                # Since DeprecationWarnings are silenced by default, also log a traditional warning.
                logger.warning(
                    f"Class {cls.__name__} is deprecated, and will be removed in a future Nautobot release. "
                    f"Instead of deriving {new_subclass.__name__} from {cls.__name__}, {message}.",
                    stacklevel=stacklevel,
                )

        cls.__init_subclass__ = classmethod(init_subclass)
        return cls

    return decorate

nautobot.apps.utils.class_deprecated_in_favor_of(replacement_class)

Decorator to mark a class as deprecated and suggest a replacement class if it is subclassed from.

Source code in nautobot/core/utils/deprecation.py 
45
46
47
def class_deprecated_in_favor_of(replacement_class):
    """Decorator to mark a class as deprecated and suggest a replacement class if it is subclassed from."""
    return class_deprecated(f"please migrate your code to inherit from class {replacement_class.__name__} instead")

nautobot.apps.utils.convert_git_diff_log_to_list(logs) 

Convert Git diff log into a list splitted by \n

Example:
    >>> git_log = "M        index.html

R sample.txt" >>> print(convert_git_diff_log_to_list(git_log)) ["Modification - index.html", "Renaming - sample.txt"]

Source code in nautobot/core/utils/git.py 
39
40
41
42
43
44
45
46
47
48
49
def convert_git_diff_log_to_list(logs):
    """
    Convert Git diff log into a list splitted by \\n

    Example:
        >>> git_log = "M\tindex.html\nR\tsample.txt"
        >>> print(convert_git_diff_log_to_list(git_log))
        ["Modification - index.html", "Renaming - sample.txt"]
    """
    logs = logs.split("\n")
    return [swap_status_initials(line) for line in logs]

nautobot.apps.utils.convert_querydict_to_factory_formset_acceptable_querydict(request_querydict, filterset)

Convert request QueryDict/GET into an acceptable factory formset QueryDict while discarding querydict params which are not part of filterset_class params

Parameters:

Name Type Description Default
request_querydict QueryDict

QueryDict to convert

 required
filterset FilterSet

Filterset class

 required

Examples:

>>> convert_querydict_to_factory_formset_acceptable_querydict({"status": ["active", "decommissioning"], "name__ic": ["location"]},)
>>> {
...     'form-TOTAL_FORMS': [3],
...     'form-INITIAL_FORMS': ['0'],
...     'form-MIN_NUM_FORMS': [''],
...     'form-MAX_NUM_FORMS': [''],
...     'form-0-lookup_field': ['status'],
...     'form-0-lookup_type': ['status'],
...     'form-0-value': ['active', 'decommissioning'],
...     'form-1-lookup_field': ['name'],
...     'form-1-lookup_type': ['name__ic'],
...     'form-1-value': ['location']
... }
Source code in nautobot/core/utils/requests.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def convert_querydict_to_factory_formset_acceptable_querydict(request_querydict, filterset):
    """
    Convert request QueryDict/GET into an acceptable factory formset QueryDict
    while discarding `querydict` params which are not part of `filterset_class` params

    Args:
        request_querydict (QueryDict): QueryDict to convert
        filterset (FilterSet): Filterset class

    Examples:
        >>> convert_querydict_to_factory_formset_acceptable_querydict({"status": ["active", "decommissioning"], "name__ic": ["location"]},)
        >>> {
        ...     'form-TOTAL_FORMS': [3],
        ...     'form-INITIAL_FORMS': ['0'],
        ...     'form-MIN_NUM_FORMS': [''],
        ...     'form-MAX_NUM_FORMS': [''],
        ...     'form-0-lookup_field': ['status'],
        ...     'form-0-lookup_type': ['status'],
        ...     'form-0-value': ['active', 'decommissioning'],
        ...     'form-1-lookup_field': ['name'],
        ...     'form-1-lookup_type': ['name__ic'],
        ...     'form-1-value': ['location']
        ... }
    """
    query_dict = QueryDict(mutable=True)
    filterset_class_fields = filterset.filters.keys()

    query_dict.setdefault("form-INITIAL_FORMS", 0)
    query_dict.setdefault("form-MIN_NUM_FORMS", 0)
    query_dict.setdefault("form-MAX_NUM_FORMS", 100)

    lookup_field_placeholder = "form-%d-lookup_field"
    lookup_type_placeholder = "form-%d-lookup_type"
    lookup_value_placeholder = "form-%d-lookup_value"

    num = 0
    request_querydict = request_querydict.copy()
    request_querydict.pop("q", None)
    for filter_field_name, value in request_querydict.items():
        # Discard fields without values
        if value:
            if filter_field_name in filterset_class_fields:
                if hasattr(filterset.filters[filter_field_name], "relationship"):
                    lookup_field = filter_field_name
                else:
                    # convert_querydict_to_factory_formset_acceptable_querydict expects to have a QueryDict as input
                    # which means we may not have the exact field name as defined in the filterset class
                    # it may contain a lookup expression (e.g. `name__ic`), so we need to strip it
                    # this is so we can select the correct field in the formset for the "field" column
                    # TODO: Since we likely need to instantiate the filterset class early in the request anyway
                    # the filterset can handle the QueryDict conversion and we can just pass the QueryDict to the filterset
                    # then use the FilterSet to de-dupe the field names
                    lookup_field = re.sub(r"__\w+", "", filter_field_name)
                lookup_value = request_querydict.getlist(filter_field_name)

                query_dict.setlistdefault(lookup_field_placeholder % num, [lookup_field])
                query_dict.setlistdefault(lookup_type_placeholder % num, [filter_field_name])
                query_dict.setlistdefault(lookup_value_placeholder % num, lookup_value)
                num += 1

    query_dict.setdefault("form-TOTAL_FORMS", max(num, 3))
    return query_dict

nautobot.apps.utils.custom_validator_clean(model_clean_func)

Decorator that wraps a models existing clean method to also execute registered plugin custom validators

:param model_clean_func: The original model clean method which is to be wrapped

Source code in nautobot/extras/plugins/validators.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def custom_validator_clean(model_clean_func):
    """
    Decorator that wraps a models existing clean method to also execute registered plugin custom validators

    :param model_clean_func: The original model clean method which is to be wrapped
    """

    @wraps(model_clean_func)
    def wrapper(model_instance):
        # Run original model clean method
        model_clean_func(model_instance)

        # Run registered plugin custom validators
        model_name = model_instance._meta.label_lower

        # Note this registry holds instances of PluginCustomValidator registered from plugins
        # which is different than the `custom_validators` model features registry
        custom_validators = registry["plugin_custom_validators"].get(model_name, [])

        for custom_validator in custom_validators:
            # If the class has not overridden the specified method, we can skip it (because we know it
            # will raise NotImplementedError).
            if getattr(custom_validator, "clean") == getattr(CustomValidator, "clean"):
                continue
            custom_validator(model_instance).clean()

    return wrapper

nautobot.apps.utils.deepmerge(original, new)

Deep merge two dictionaries (new into original) and return a new dict

Source code in nautobot/core/utils/data.py 
15
16
17
18
19
20
21
22
23
24
25
def deepmerge(original, new):
    """
    Deep merge two dictionaries (new into original) and return a new dict
    """
    merged = OrderedDict(original)
    for key, val in new.items():
        if key in original and isinstance(original[key], dict) and isinstance(val, dict):
            merged[key] = deepmerge(original[key], val)
        else:
            merged[key] = val
    return merged

nautobot.apps.utils.ensure_content_type_and_field_name_in_query_params(query_params)

Ensure query_params includes content_type and field_name and content_type is a valid ContentType.

Return the 'ContentTypes' model and 'field_name' if validation was successful.

Source code in nautobot/core/utils/requests.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def ensure_content_type_and_field_name_in_query_params(query_params):
    """Ensure `query_params` includes `content_type` and `field_name` and `content_type` is a valid ContentType.

    Return the 'ContentTypes' model and 'field_name' if validation was successful.
    """
    if "content_type" not in query_params or "field_name" not in query_params:
        raise ValidationError("content_type and field_name are required parameters", code=400)
    contenttype = query_params.get("content_type")
    app_label, model_name = contenttype.split(".")
    try:
        model_contenttype = ContentType.objects.get(app_label=app_label, model=model_name)
        model = model_contenttype.model_class()
        if model is None:
            raise ValidationError(f"model for content_type: <{model_contenttype}> not found", code=500)
    except ContentType.DoesNotExist:
        raise ValidationError("content_type not found", code=404)
    field_name = query_params.get("field_name")

    return field_name, model

nautobot.apps.utils.fixup_null_statuses(*, model, model_contenttype, status_model)

For instances of model that have an invalid NULL status field, create and use a special status_model instance.

Source code in nautobot/extras/utils.py 
489
490
491
492
493
494
495
496
497
498
499
500
501
502
def fixup_null_statuses(*, model, model_contenttype, status_model):
    """For instances of model that have an invalid NULL status field, create and use a special status_model instance."""
    instances_to_fixup = model.objects.filter(status__isnull=True)
    if instances_to_fixup.exists():
        null_status, _ = status_model.objects.get_or_create(
            name="NULL",
            defaults={
                "color": ColorChoices.COLOR_BLACK,
                "description": "Created by Nautobot to replace invalid null references",
            },
        )
        null_status.content_types.add(model_contenttype)
        updated_count = instances_to_fixup.update(status=null_status)
        print(f"    Found and fixed {updated_count} instances of {model.__name__} that had null 'status' fields.")

nautobot.apps.utils.flatten_dict(d, prefix='', separator='.')

Flatten nested dictionaries into a single level by joining key names with a separator.

:param d: The dictionary to be flattened :param prefix: Initial prefix (if any) :param separator: The character to use when concatenating key names

Source code in nautobot/core/utils/data.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def flatten_dict(d, prefix="", separator="."):
    """
    Flatten nested dictionaries into a single level by joining key names with a separator.

    :param d: The dictionary to be flattened
    :param prefix: Initial prefix (if any)
    :param separator: The character to use when concatenating key names
    """
    ret = {}
    for k, v in d.items():
        key = separator.join([prefix, k]) if prefix else k
        if isinstance(v, dict):
            ret.update(flatten_dict(v, prefix=key))
        else:
            ret[key] = v
    return ret

nautobot.apps.utils.flatten_iterable(iterable)

Flatten a nested iterable such as a list of lists, keeping strings intact.

:param iterable: The iterable to be flattened :returns: generator

Source code in nautobot/core/utils/data.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
def flatten_iterable(iterable):
    """
    Flatten a nested iterable such as a list of lists, keeping strings intact.

    :param iterable: The iterable to be flattened
    :returns: generator
    """
    for i in iterable:
        if hasattr(i, "__iter__") and not isinstance(i, str):
            for j in flatten_iterable(i):
                yield j
        else:
            yield i

nautobot.apps.utils.foreground_color(bg_color)

Return the ideal foreground color (black or white) for a given background color in hexadecimal RGB format.

Source code in nautobot/core/utils/color.py 
18
19
20
21
22
23
24
25
26
27
def foreground_color(bg_color):
    """
    Return the ideal foreground color (black or white) for a given background color in hexadecimal RGB format.
    """
    bg_color = bg_color.strip("#")
    r, g, b = hex_to_rgb(bg_color)
    if r * 0.299 + g * 0.587 + b * 0.114 > 186:
        return "000000"
    else:
        return "ffffff"

nautobot.apps.utils.generate_signature(request_body, secret)

Return a cryptographic signature that can be used to verify the authenticity of webhook data.

Source code in nautobot/extras/utils.py 
272
273
274
275
276
277
def generate_signature(request_body, secret):
    """
    Return a cryptographic signature that can be used to verify the authenticity of webhook data.
    """
    hmac_prep = hmac.new(key=secret.encode("utf8"), msg=request_body, digestmod=hashlib.sha512)
    return hmac_prep.hexdigest()

nautobot.apps.utils.get_all_lookup_expr_for_field(model, field_name)

Return all lookup expressions for field_name in model filterset

Source code in nautobot/core/utils/filtering.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def get_all_lookup_expr_for_field(model, field_name):
    """
    Return all lookup expressions for `field_name` in `model` filterset
    """
    filterset = get_filterset_for_model(model)().filters

    if not filterset.get(field_name):
        raise exceptions.FilterSetFieldNotFound("field_name not found")

    if field_name.startswith("has_"):
        return [{"id": field_name, "name": "exact"}]

    lookup_expr_choices = []

    for name, field in filterset.items():
        # remove the lookup_expr from field_name e.g name__iew -> name
        if re.sub(r"__\w+", "", name) == field_name and not name.startswith("has_"):
            lookup_expr_choices.append(
                {
                    "id": name,
                    "name": build_lookup_label(name, field.lookup_expr),
                }
            )
        elif name == field_name and not name.startswith("has_"):
            lookup_expr_choices.append(
                {
                    "id": name,
                    "name": "exact",
                }
            )

    return lookup_expr_choices

nautobot.apps.utils.get_base_template(base_template, model)

Returns the name of the base template, if the base_template is not None Otherwise, default to using "/.html" as the base template, if it exists. Otherwise, check if "/_retrieve.html" used in NautobotUIViewSet exists. If both templates do not exist, fall back to "base.html".

Source code in nautobot/extras/utils.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def get_base_template(base_template, model):
    """
    Returns the name of the base template, if the base_template is not None
    Otherwise, default to using "<app>/<model>.html" as the base template, if it exists.
    Otherwise, check if "<app>/<model>_retrieve.html" used in `NautobotUIViewSet` exists.
    If both templates do not exist, fall back to "base.html".
    """
    if base_template is None:
        base_template = f"{model._meta.app_label}/{model._meta.model_name}.html"
        # 2.0 TODO(Hanlin): This can be removed once an object view has been established for every model.
        try:
            get_template(base_template)
        except TemplateDoesNotExist:
            base_template = f"{model._meta.app_label}/{model._meta.model_name}_retrieve.html"
            try:
                get_template(base_template)
            except TemplateDoesNotExist:
                base_template = "base.html"
    return base_template

nautobot.apps.utils.get_celery_queues()

Return a dictionary of celery queues and the number of workers active on the queue in the form {queue_name: num_workers}

Source code in nautobot/extras/utils.py 
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
def get_celery_queues():
    """
    Return a dictionary of celery queues and the number of workers active on the queue in
    the form {queue_name: num_workers}
    """
    from nautobot.core.celery import app  # prevent circular import

    celery_queues = {}

    celery_inspect = app.control.inspect()
    active_queues = celery_inspect.active_queues()
    if active_queues is None:
        return celery_queues
    for task_queue_list in active_queues.values():
        distinct_queues = {q["name"] for q in task_queue_list}
        for queue in distinct_queues:
            celery_queues.setdefault(queue, 0)
            celery_queues[queue] += 1

    return celery_queues

nautobot.apps.utils.get_changes_for_model(model)

Return a queryset of ObjectChanges for a model or instance. The queryset will be filtered by the model class. If an instance is provided, the queryset will also be filtered by the instance id.

Source code in nautobot/core/utils/lookup.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
def get_changes_for_model(model):
    """
    Return a queryset of ObjectChanges for a model or instance. The queryset will be filtered
    by the model class. If an instance is provided, the queryset will also be filtered by the instance id.
    """
    from nautobot.extras.models import ObjectChange  # prevent circular import

    if isinstance(model, Model):
        return ObjectChange.objects.filter(
            changed_object_type=ContentType.objects.get_for_model(model._meta.model),
            changed_object_id=model.pk,
        )
    if issubclass(model, Model):
        return ObjectChange.objects.filter(changed_object_type=ContentType.objects.get_for_model(model._meta.model))
    raise TypeError(f"{model!r} is not a Django Model class or instance")

nautobot.apps.utils.get_filter_field_label(filter_field)

Return a label for a given field name and value.

Parameters:

Name Type Description Default
filter_field Filter

The filter to get a label for

 required

Returns:

Type Description
str

The label for the given field
Source code in nautobot/core/utils/filtering.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
def get_filter_field_label(filter_field):
    """
    Return a label for a given field name and value.

    Args:
        filter_field (Filter): The filter to get a label for

    Returns:
        (str): The label for the given field
    """

    if filter_field.label:
        return filter_field.label
    elif hasattr(filter_field, "relationship"):
        return filter_field.relationship.get_label(side=filter_field.side)
    elif hasattr(filter_field, "custom_field"):
        return filter_field.custom_field.label
    else:
        return _field_name_to_display(filter_field.field_name)

nautobot.apps.utils.get_filterable_params_from_filter_params(filter_params, non_filter_params, filterset)

Remove any non_filter_params and fields that are not a part of the filterset from filter_params to return only queryset filterable parameters.

Parameters:

Name Type Description Default
filter_params QueryDict

Filter param querydict

 required
non_filter_params list

Non queryset filterable params

 required
filterset FilterSet

The FilterSet class

 required

Returns:

Type Description
QueryDict

Filter param querydict with only queryset filterable params
Source code in nautobot/core/utils/requests.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def get_filterable_params_from_filter_params(filter_params, non_filter_params, filterset):
    """
    Remove any `non_filter_params` and fields that are not a part of the filterset from  `filter_params`
    to return only queryset filterable parameters.

    Args:
        filter_params (QueryDict): Filter param querydict
        non_filter_params (list): Non queryset filterable params
        filterset (FilterSet): The FilterSet class

    Returns:
        (QueryDict): Filter param querydict with only queryset filterable params
    """
    for non_filter_param in non_filter_params:
        filter_params.pop(non_filter_param, None)

    # Some FilterSet field only accept single choice not multiple choices
    # e.g datetime field, bool fields etc.
    final_filter_params = {}
    for field in filter_params.keys():
        if filter_params.get(field):
            # `is_single_choice_field` implements `get_filterset_field`, which throws an exception if a field is not found.
            # If an exception is thrown, instead of throwing an exception, set `_is_single_choice_field` to 'False'
            # because the fields that were not discovered are still necessary.
            try:
                _is_single_choice_field = is_single_choice_field(filterset, field)
            except exceptions.FilterSetFieldNotFound:
                _is_single_choice_field = False

            final_filter_params[field] = (
                filter_params.get(field) if _is_single_choice_field else filter_params.getlist(field)
            )

    return final_filter_params

nautobot.apps.utils.get_filterset_for_model(model)

Return the FilterSet class associated with a given model.

The FilterSet class is expected to be in the filters module within the application associated with the model and its name is expected to be {ModelName}FilterSet.

If a matching FilterSet is not found, this will return None.

Parameters:

Name Type Description Default
model BaseModel

A model class

 required

Returns:

Type Description
Union[FilterSet, None]

Either the FilterSet class or None
Source code in nautobot/core/utils/lookup.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def get_filterset_for_model(model):
    """Return the `FilterSet` class associated with a given `model`.

    The `FilterSet` class is expected to be in the `filters` module within the application
    associated with the model and its name is expected to be `{ModelName}FilterSet`.

    If a matching `FilterSet` is not found, this will return `None`.

    Args:
        model (BaseModel): A model class

    Returns:
        (Union[FilterSet,None]): Either the `FilterSet` class or `None`
    """
    return get_related_class_for_model(model, module_name="filters", object_suffix="FilterSet")

nautobot.apps.utils.get_filterset_parameter_form_field(model, parameter, filterset=None)

Return the relevant form field instance for a filterset parameter e.g DynamicModelMultipleChoiceField, forms.IntegerField e.t.c

Source code in nautobot/core/utils/filtering.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
def get_filterset_parameter_form_field(model, parameter, filterset=None):
    """
    Return the relevant form field instance for a filterset parameter e.g DynamicModelMultipleChoiceField, forms.IntegerField e.t.c
    """
    # Avoid circular import
    from nautobot.dcim.models import Device
    from nautobot.extras.filters import ContentTypeMultipleChoiceFilter, CustomFieldFilterMixin, StatusFilter
    from nautobot.extras.models import ConfigContext, Role, Status, Tag
    from nautobot.extras.utils import ChangeLoggedModelsQuery, RoleModelsQuery, TaggableClassesQuery
    from nautobot.core.filters import MultiValueDecimalFilter, MultiValueFloatFilter
    from nautobot.core.forms import (
        BOOLEAN_CHOICES,
        DynamicModelMultipleChoiceField,
        MultipleContentTypeField,
        StaticSelect2,
        StaticSelect2Multiple,
    )
    from nautobot.virtualization.models import VirtualMachine

    if filterset is None or filterset.Meta.model != model:
        filterset = get_filterset_for_model(model)()
    field = get_filterset_field(filterset, parameter)
    form_field = field.field

    # TODO(Culver): We are having to replace some widgets here because multivalue_field_factory that generates these isn't smart enough
    if isinstance(field, CustomFieldFilterMixin):
        form_field = field.custom_field.to_filter_form_field(lookup_expr=field.lookup_expr)
    elif isinstance(field, (MultiValueDecimalFilter, MultiValueFloatFilter)):
        form_field = forms.DecimalField()
    elif isinstance(field, NumberFilter):
        form_field = forms.IntegerField()
    elif isinstance(field, ModelMultipleChoiceFilter):
        related_model = Status if isinstance(field, StatusFilter) else field.extra["queryset"].model
        form_attr = {
            "queryset": related_model.objects.all(),
            "to_field_name": field.extra.get("to_field_name", "id"),
        }
        # ConfigContext requires content_type set to Device and VirtualMachine
        if model == ConfigContext:
            form_attr["query_params"] = {"content_types": [Device._meta.label_lower, VirtualMachine._meta.label_lower]}
        # Status and Tag api requires content_type, to limit result to only related content_types
        elif related_model in [Role, Status, Tag]:
            form_attr["query_params"] = {"content_types": model._meta.label_lower}

        form_field = DynamicModelMultipleChoiceField(**form_attr)
    elif isinstance(
        field, ContentTypeMultipleChoiceFilter
    ):  # While there are other objects using `ContentTypeMultipleChoiceFilter`, the case where
        # models that have such a filter and the `verbose_name_plural` has multiple words is ony one: "dynamic groups".
        from nautobot.core.models.fields import slugify_dashes_to_underscores  # Avoid circular import

        plural_name = slugify_dashes_to_underscores(model._meta.verbose_name_plural)
        # Cable-connectable models use "cable_terminations", not "cables", as the feature name
        if plural_name == "cables":
            plural_name == "cable_terminations"
        try:
            form_field = MultipleContentTypeField(choices_as_strings=True, feature=plural_name)
        except KeyError:
            # `MultipleContentTypeField` employs `registry["model features"][feature]`, which may
            # result in an error if `feature` is not found in the `registry["model features"]` dict.
            # In this case use queryset
            queryset_map = {
                "tags": TaggableClassesQuery,
                "job_hooks": ChangeLoggedModelsQuery,
                "roles": RoleModelsQuery,
            }
            form_field = MultipleContentTypeField(
                choices_as_strings=True, queryset=queryset_map[plural_name]().as_queryset()
            )
    elif isinstance(field, (MultipleChoiceFilter, ChoiceFilter)) and "choices" in field.extra:
        form_field = forms.MultipleChoiceField(choices=field.extra.get("choices"), widget=StaticSelect2Multiple)

    elif isinstance(field, BooleanFilter):
        form_field = forms.ChoiceField(choices=BOOLEAN_CHOICES, widget=StaticSelect2)

    form_field.required = False
    form_field.initial = None
    form_field.widget.attrs.pop("required", None)

    css_classes = form_field.widget.attrs.get("class", "")
    form_field.widget.attrs["class"] = "form-control " + css_classes
    return form_field

nautobot.apps.utils.get_form_for_model(model, form_prefix='')

Return the Form class associated with a given model.

The Form class is expected to be in the forms module within the application associated with the model and its name is expected to be {ModelName}{form_prefix}Form.

If a matching Form is not found, this will return None.

Parameters:

Name Type Description Default
form_prefix str

An additional prefix for the form name (e.g. Filter, such as to retrieve FooFilterForm) that will come after the model name.

 ''

Returns:

Type Description
Union[Form, None]

Either the Form class or None
Source code in nautobot/core/utils/lookup.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def get_form_for_model(model, form_prefix=""):
    """Return the `Form` class associated with a given `model`.

    The `Form` class is expected to be in the `forms` module within the application
    associated with the model and its name is expected to be `{ModelName}{form_prefix}Form`.

    If a matching `Form` is not found, this will return `None`.

    Args:
        form_prefix (str):
            An additional prefix for the form name (e.g. `Filter`, such as to retrieve
            `FooFilterForm`) that will come after the model name.

    Returns:
        (Union[Form, None]): Either the `Form` class or `None`
    """
    object_suffix = f"{form_prefix}Form"
    return get_related_class_for_model(model, module_name="forms", object_suffix=object_suffix)

nautobot.apps.utils.get_latest_release(pre_releases=False)

Get latest known Nautobot release from cache, or if not available, queue up a background task to populate the cache.

Returns:

Type Description
(Version, str)

Latest release version and the release URL, if found in the cache
(unknown, None)

If not present in the cache at this time
Source code in nautobot/core/releases.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def get_latest_release(pre_releases=False):
    """
    Get latest known Nautobot release from cache, or if not available, queue up a background task to populate the cache.

    Returns:
        (Version, str): Latest release version and the release URL, if found in the cache
        ("unknown", None): If not present in the cache at this time
    """
    if get_settings_or_config("RELEASE_CHECK_URL"):
        logger.debug("Checking for most recent release")
        latest_release = cache.get("latest_release")
        if latest_release is not None:
            logger.debug(f"Found cached release: {latest_release}")
            return latest_release
        # Get the releases in the background worker, it will fill the cache
        logger.info("Initiating background task to retrieve updated releases list")
        get_releases.delay(pre_releases=pre_releases)

    else:
        logger.debug("Skipping release check; RELEASE_CHECK_URL not defined")

    return "unknown", None

nautobot.apps.utils.get_model_from_name(model_name)

Given a full model name in dotted format (example: dcim.model), a model class is returned if valid.

:param model_name: Full dotted name for a model as a string (ex: dcim.model) :type model_name: str

:raises TypeError: If given model name is not found.

:return: Found model.

Source code in nautobot/core/utils/lookup.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def get_model_from_name(model_name):
    """Given a full model name in dotted format (example: `dcim.model`), a model class is returned if valid.

    :param model_name: Full dotted name for a model as a string (ex: `dcim.model`)
    :type model_name: str

    :raises TypeError: If given model name is not found.

    :return: Found model.
    """
    from django.apps import apps

    try:
        return apps.get_model(model_name)
    except (ValueError, LookupError) as exc:
        raise TypeError(exc) from exc

nautobot.apps.utils.get_only_new_ui_ready_routes(patterns, prefix='')

Recursively traverses Django URL patterns to find routes associated with view classes that have the use_new_ui attribute set to True.

Parameters:

Name Type Description Default
patterns list

List of URL patterns to traverse.

 required
prefix str

URL pattern prefix to include when constructing route patterns.

 ''

Returns:

Type Description
list

A list of route patterns associated with view classes that use the new UI.
Source code in nautobot/core/utils/navigation.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def get_only_new_ui_ready_routes(patterns, prefix=""):
    """
    Recursively traverses Django URL patterns to find routes associated with view classes
    that have the `use_new_ui` attribute set to `True`.

    Args:
        patterns (list): List of URL patterns to traverse.
        prefix (str): URL pattern prefix to include when constructing route patterns.

    Returns:
        (list): A list of route patterns associated with view classes that use the new UI.
    """
    new_ui_routes = set()
    for pattern in patterns:
        if hasattr(pattern, "url_patterns"):
            r_pattern = pattern.pattern.regex.pattern.lstrip("^").rstrip(r"\Z")
            combined_pattern = prefix + r_pattern
            new_ui_routes.update(get_only_new_ui_ready_routes(pattern.url_patterns, combined_pattern))
        else:
            use_new_ui = False
            # There are two types of generic view class ObjectView and NautobotUIViewSet which has different approach to validate if this route is a new_ui_ready route
            if hasattr(pattern.callback, "view_class"):
                # For ObjectView
                use_new_ui = getattr(pattern.callback.view_class, "use_new_ui", False)
            elif hasattr(pattern.callback, "cls"):
                # For NautobotUIViewSet
                use_new_ui_list = getattr(pattern.callback.cls, "use_new_ui", [])
                # Check if the current action is part of the allowed actions in this ViewSet class
                use_new_ui = bool(set(use_new_ui_list) & set(pattern.callback.actions.values()))
            if use_new_ui:
                r_pattern = pattern.pattern.regex.pattern.lstrip("^")
                final_pattern = rf"^{prefix}{r_pattern}"
                new_ui_routes.add(final_pattern)
    return new_ui_routes

nautobot.apps.utils.get_permission_for_model(model, action)

Resolve the named permission for a given model (or instance) and action (e.g. view or add).

:param model: A model or instance :param action: View, add, change, or delete (string)

Source code in nautobot/core/utils/permissions.py 
 5
 6
 7
 8
 9
10
11
12
13
14
15
def get_permission_for_model(model, action):
    """
    Resolve the named permission for a given model (or instance) and action (e.g. view or add).

    :param model: A model or instance
    :param action: View, add, change, or delete (string)
    """
    if action not in ("view", "add", "change", "delete"):
        raise ValueError(f"Unsupported action: {action}")

    return f"{model._meta.app_label}.{action}_{model._meta.model_name}"

Return the appropriate class associated with a given model matching the module_name and object_suffix.

The given model can either be a model class, a model instance, or a dotted representation (ex: dcim.device).

The object class is expected to be in the module within the application associated with the model and its name is expected to be {ModelName}{object_suffix}.

If a matching class is not found, this will return None.

Parameters:

Name Type Description Default
model Union[BaseModel, str]

A model class, instance, or dotted representation

 required
module_name str

The name of the module to search for the object class

 required
object_suffix str

The suffix to append to the model name to find the object class

 required

Returns:

Type Description
Union[BaseModel, str]

Either the matching object class or None
Source code in nautobot/core/utils/lookup.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def get_related_class_for_model(model, module_name, object_suffix):
    """Return the appropriate class associated with a given model matching the `module_name` and
    `object_suffix`.

    The given `model` can either be a model class, a model instance, or a dotted representation (ex: `dcim.device`).

    The object class is expected to be in the module within the application
    associated with the model and its name is expected to be `{ModelName}{object_suffix}`.

    If a matching class is not found, this will return `None`.

    Args:
        model (Union[BaseModel, str]): A model class, instance, or dotted representation
        module_name (str): The name of the module to search for the object class
        object_suffix (str): The suffix to append to the model name to find the object class

    Returns:
        (Union[BaseModel, str]): Either the matching object class or None
    """
    if isinstance(model, str):
        model = get_model_from_name(model)
    if isinstance(model, Model):
        model = type(model)
    if not inspect.isclass(model):
        raise TypeError(f"{model!r} is not a Django Model class")
    if not issubclass(model, Model):
        raise TypeError(f"{model!r} is not a subclass of a Django Model class")

    # e.g. "nautobot.dcim.forms.DeviceFilterForm"
    app_label = model._meta.app_label
    object_name = f"{model.__name__}{object_suffix}"
    object_path = f"{app_label}.{module_name}.{object_name}"
    if app_label not in settings.PLUGINS:
        object_path = f"nautobot.{object_path}"

    try:
        return import_string(object_path)
    # The name of the module is not correct or unable to find the desired object for this model
    except (AttributeError, ImportError, ModuleNotFoundError):
        pass

    return None

nautobot.apps.utils.get_route_for_model(model, action, api=False)

Return the URL route name for the given model and action. Does not perform any validation. Supports both core and plugin routes.

Parameters:

Name Type Description Default
model (models.Model, str)

Class, Instance, or dotted string of a Django Model

 required
action str

name of the action in the route

 required
api bool

If set, return an API route.

 False

Returns:

Type Description
str

return the name of the view for the model/action provided.

Examples:

>>> get_route_for_model(Device, "list")
"dcim:device_list"
>>> get_route_for_model(Device, "list", api=True)
"dcim-api:device-list"
>>> get_route_for_model("dcim.location", "list")
"dcim:location_list"
>>> get_route_for_model("dcim.location", "list", api=True)
"dcim-api:location-list"
>>> get_route_for_model(ExampleModel, "list")
"plugins:example_plugin:examplemodel_list"
>>> get_route_for_model(ExampleModel, "list", api=True)
"plugins-api:example_plugin-api:examplemodel-list"
Source code in nautobot/core/utils/lookup.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def get_route_for_model(model, action, api=False):
    """
    Return the URL route name for the given model and action. Does not perform any validation.
    Supports both core and plugin routes.

    Args:
        model (models.Model, str): Class, Instance, or dotted string of a Django Model
        action (str): name of the action in the route
        api (bool): If set, return an API route.

    Returns:
        (str): return the name of the view for the model/action provided.

    Examples:
        >>> get_route_for_model(Device, "list")
        "dcim:device_list"
        >>> get_route_for_model(Device, "list", api=True)
        "dcim-api:device-list"
        >>> get_route_for_model("dcim.location", "list")
        "dcim:location_list"
        >>> get_route_for_model("dcim.location", "list", api=True)
        "dcim-api:location-list"
        >>> get_route_for_model(ExampleModel, "list")
        "plugins:example_plugin:examplemodel_list"
        >>> get_route_for_model(ExampleModel, "list", api=True)
        "plugins-api:example_plugin-api:examplemodel-list"
    """

    if isinstance(model, str):
        model = get_model_from_name(model)

    suffix = "" if not api else "-api"
    # The `contenttypes` and `auth` app doesn't provide REST API endpoints,
    # but Nautobot provides one for the ContentType model in our `extras` and Group model in `users` app.
    if model is ContentType:
        app_label = "extras"
    elif model is Group:
        app_label = "users"
    else:
        app_label = model._meta.app_label
    prefix = f"{app_label}{suffix}:{model._meta.model_name}"
    sep = ""
    if action != "":
        sep = "_" if not api else "-"
    viewname = f"{prefix}{sep}{action}"

    if model._meta.app_label in settings.PLUGINS:
        viewname = f"plugins{suffix}:{viewname}"

    return viewname

nautobot.apps.utils.get_settings_or_config(variable_name)

Get a value from Django settings (if specified there) or Constance configuration (otherwise).

Source code in nautobot/core/utils/config.py 
 7
 8
 9
10
11
12
def get_settings_or_config(variable_name):
    """Get a value from Django settings (if specified there) or Constance configuration (otherwise)."""
    # Explicitly set in settings.py or nautobot_config.py takes precedence, for now
    if hasattr(settings, variable_name):
        return getattr(settings, variable_name)
    return getattr(config, variable_name)

nautobot.apps.utils.get_table_for_model(model)

Return the Table class associated with a given model.

The Table class is expected to be in the tables module within the application associated with the model and its name is expected to be {ModelName}Table.

If a matching Table is not found, this will return None.

Parameters:

Name Type Description Default
model BaseModel

A model class

 required

Returns:

Type Description
Union[Table, None]

Either the Table class or None
Source code in nautobot/core/utils/lookup.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
def get_table_for_model(model):
    """Return the `Table` class associated with a given `model`.

    The `Table` class is expected to be in the `tables` module within the application
    associated with the model and its name is expected to be `{ModelName}Table`.

    If a matching `Table` is not found, this will return `None`.

    Args:
        model (BaseModel): A model class

    Returns:
        (Union[Table, None]): Either the `Table` class or `None`
    """
    return get_related_class_for_model(model, module_name="tables", object_suffix="Table")

nautobot.apps.utils.get_worker_count(request=None, queue=None)

Return a count of the active Celery workers in a specified queue. Defaults to the CELERY_TASK_DEFAULT_QUEUE setting.

Source code in nautobot/extras/utils.py 
302
303
304
305
306
307
308
309
def get_worker_count(request=None, queue=None):
    """
    Return a count of the active Celery workers in a specified queue. Defaults to the `CELERY_TASK_DEFAULT_QUEUE` setting.
    """
    celery_queues = get_celery_queues()
    if not queue:
        queue = settings.CELERY_TASK_DEFAULT_QUEUE
    return celery_queues.get(queue, 0)

nautobot.apps.utils.hex_to_rgb(hex_str)

Map a hex string like "00ff00" to individual r, g, b integer values.

Source code in nautobot/core/utils/color.py 
4
5
6
7
8
def hex_to_rgb(hex_str):
    """
    Map a hex string like "00ff00" to individual r, g, b integer values.
    """
    return [int(hex_str[c : c + 2], 16) for c in (0, 2, 4)]  # noqa: E203

nautobot.apps.utils.image_upload(instance, filename)

Return a path for uploading image attachments.

Source code in nautobot/extras/utils.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def image_upload(instance, filename):
    """
    Return a path for uploading image attachments.
    """
    path = "image-attachments/"

    # Rename the file to the provided name, if any. Attempt to preserve the file extension.
    extension = filename.rsplit(".")[-1].lower()
    if instance.name and extension in ["bmp", "gif", "jpeg", "jpg", "png"]:
        filename = ".".join([instance.name, extension])
    elif instance.name:
        filename = instance.name

    return f"{path}{instance.content_type.name}_{instance.object_id}_{filename}"

nautobot.apps.utils.is_url(value)

Validate whether a value is a URL.

Parameters:

Name Type Description Default
value str

String to validate.

 required

Returns:

Type Description
bool

True if the value is a valid URL, False otherwise.
Source code in nautobot/core/utils/data.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def is_url(value):
    """
    Validate whether a value is a URL.

    Args:
        value (str): String to validate.

    Returns:
        (bool): True if the value is a valid URL, False otherwise.
    """
    try:
        return validators.URLValidator()(value) is None
    except validators.ValidationError:
        return False

nautobot.apps.utils.lighten_color(r, g, b, factor)

Make a given RGB color lighter (closer to white).

Source code in nautobot/core/utils/color.py 
30
31
32
33
34
35
36
37
38
def lighten_color(r, g, b, factor):
    """
    Make a given RGB color lighter (closer to white).
    """
    return [
        int(255 - (255 - r) * (1.0 - factor)),
        int(255 - (255 - g) * (1.0 - factor)),
        int(255 - (255 - b) * (1.0 - factor)),
    ]

nautobot.apps.utils.merge_dicts_without_collision(d1, d2)

Merge two dicts into a new dict, but raise a ValueError if any key exists with differing values across both dicts.

Source code in nautobot/core/utils/data.py 
86
87
88
89
90
91
92
93
94
def merge_dicts_without_collision(d1, d2):
    """
    Merge two dicts into a new dict, but raise a ValueError if any key exists with differing values across both dicts.
    """
    intersection = d1.keys() & d2.keys()
    for k in intersection:
        if d1[k] != d2[k]:
            raise ValueError(f'Conflicting values for key "{k}": ({d1[k]!r}, {d2[k]!r})')
    return {**d1, **d2}

nautobot.apps.utils.migrate_role_data(model_to_migrate, *, from_role_field_name, from_role_model=None, from_role_choiceset=None, to_role_field_name, to_role_model=None, to_role_choiceset=None, is_m2m_field=False)

Update all model_to_migrate with a value for to_role_field based on from_role_field values.

Parameters:

Name Type Description Default
model_to_migrate Model

Model with role fields to alter

 required
from_role_field_name str

Name of the field on model_to_migrate to use as source data

 required
from_role_model Model

If from_role_field is a ForeignKey or M2M field, the corresponding model for it

 None
from_role_choiceset ChoiceSet

If from_role_field is a choices field, the corresponding ChoiceSet for it

 None
to_role_field_name str

Name of the field on model_to_migrate to update based on the from_role_field

 required
to_role_model Model

If to_role_field is a ForeignKey or M2M field, the corresponding model for it

 None
to_role_choiceset ChoiceSet

If to_role_field is a choices field, the corresponding ChoiceSet for it

 None
is_m2m_field bool

True if the role fields are both ManyToManyFields, else False

 False
Source code in nautobot/extras/utils.py 
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
def migrate_role_data(
    model_to_migrate,
    *,
    from_role_field_name,
    from_role_model=None,
    from_role_choiceset=None,
    to_role_field_name,
    to_role_model=None,
    to_role_choiceset=None,
    is_m2m_field=False,
):
    """
    Update all `model_to_migrate` with a value for `to_role_field` based on `from_role_field` values.

    Args:
        model_to_migrate (Model): Model with role fields to alter
        from_role_field_name (str): Name of the field on `model_to_migrate` to use as source data
        from_role_model (Model): If `from_role_field` is a ForeignKey or M2M field, the corresponding model for it
        from_role_choiceset (ChoiceSet): If `from_role_field` is a choices field, the corresponding ChoiceSet for it
        to_role_field_name (str): Name of the field on `model_to_migrate` to update based on the `from_role_field`
        to_role_model (Model): If `to_role_field` is a ForeignKey or M2M field, the corresponding model for it
        to_role_choiceset (ChoiceSet): If `to_role_field` is a choices field, the corresponding ChoiceSet for it
        is_m2m_field (bool): True if the role fields are both ManyToManyFields, else False
    """
    if from_role_model is not None:
        assert from_role_choiceset is None
        if to_role_model is not None:
            assert to_role_choiceset is None
            # Mapping "from" model instances to corresponding "to" model instances
            roles_translation_mapping = {
                # Use .filter().first(), not .get() because "to" role might not exist, especially on reverse migrations
                from_role: to_role_model.objects.filter(name=from_role.name).first()
                for from_role in from_role_model.objects.all()
            }
        else:
            assert to_role_choiceset is not None
            # Mapping "from" model instances to corresponding "to" choices
            # We need to use `label` to look up the from_role instance, but `value` is what we set for the to_role_field
            inverted_to_role_choiceset = {label: value for value, label in to_role_choiceset.CHOICES}
            roles_translation_mapping = {
                from_role: inverted_to_role_choiceset.get(from_role.name, None)
                for from_role in from_role_model.objects.all()
            }
    else:
        assert from_role_choiceset is not None
        if to_role_model is not None:
            assert to_role_choiceset is None
            # Mapping "from" choices to corresponding "to" model instances
            roles_translation_mapping = {
                # Use .filter().first(), not .get() because "to" role might not exist, especially on reverse migrations
                from_role_value: to_role_model.objects.filter(name=from_role_label).first()
                for from_role_value, from_role_label in from_role_choiceset.CHOICES
            }
        else:
            assert to_role_choiceset is not None
            # Mapping "from" choices to corresponding "to" choices; we don't currently use this case, but it should work
            # We need to use `label` to look up the from_role instance, but `value` is what we set for the to_role_field
            inverted_to_role_choiceset = {label: value for value, label in to_role_choiceset.CHOICES}
            roles_translation_mapping = {
                from_role_value: inverted_to_role_choiceset.get(from_role_label, None)
                for from_role_value, from_role_label in from_role_choiceset.CHOICES
            }

    if not is_m2m_field:
        # Bulk updates of a single field are easy enough...
        for from_role_value, to_role_value in roles_translation_mapping.items():
            if to_role_value is not None:
                updated_count = model_to_migrate.objects.filter(**{from_role_field_name: from_role_value}).update(
                    **{to_role_field_name: to_role_value}
                )
                logger.info(
                    'Updated %d %s records to reference %s "%s"',
                    updated_count,
                    model_to_migrate._meta.label,
                    to_role_field_name,
                    to_role_value.name if to_role_model else to_role_value,
                )
    else:
        # ...but we have to update each instance's M2M field independently?
        for instance in model_to_migrate.objects.all():
            to_role_set = {
                roles_translation_mapping[from_role_value]
                for from_role_value in getattr(instance, from_role_field_name).all()
            }
            # Discard any null values
            to_role_set.discard(None)
            getattr(instance, to_role_field_name).set(to_role_set)
        logger.info(
            "Updated %d %s record %s M2M fields",
            model_to_migrate.objects.count(),
            model_to_migrate._meta.label,
            to_role_field_name,
        )

nautobot.apps.utils.normalize_querydict(querydict, form_class=None)

Convert a QueryDict to a normal, mutable dictionary, preserving list values. For example,

QueryDict('foo=1&bar=2&bar=3&baz=')
becomes

{'foo': '1', 'bar': ['2', '3'], 'baz': ''}

This function is necessary because QueryDict does not provide any built-in mechanism which preserves multiple values.

A form_class can be provided as a way to hint which query parameters should be treated as lists.

Source code in nautobot/core/utils/requests.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
def normalize_querydict(querydict, form_class=None):
    """
    Convert a QueryDict to a normal, mutable dictionary, preserving list values. For example,

        QueryDict('foo=1&bar=2&bar=3&baz=')

    becomes:

        {'foo': '1', 'bar': ['2', '3'], 'baz': ''}

    This function is necessary because QueryDict does not provide any built-in mechanism which preserves multiple
    values.

    A `form_class` can be provided as a way to hint which query parameters should be treated as lists.
    """
    result = {}
    if querydict:
        for key, value_list in querydict.lists():
            if len(value_list) > 1:
                # More than one value in the querydict for this key, so keep it as a list
                # TODO: we could check here and de-listify value_list if the form_class field is a single-value one?
                result[key] = value_list
            elif (
                form_class is not None
                and key in form_class.base_fields
                # ModelMultipleChoiceField is *not* itself a subclass of MultipleChoiceField, thanks Django!
                and isinstance(form_class.base_fields[key], (forms.MultipleChoiceField, forms.ModelMultipleChoiceField))
            ):
                # Even though there's only a single value in the querydict for this key, the form wants it as a list
                result[key] = value_list
            else:
                # Only a single value in the querydict for this key, and no guidance otherwise, so make it single
                result[key] = value_list[0]
    return result

nautobot.apps.utils.permission_is_exempt(name)

Determine whether a specified permission is exempt from evaluation.

:param name: Permission name in the format ._

Source code in nautobot/core/utils/permissions.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def permission_is_exempt(name):
    """
    Determine whether a specified permission is exempt from evaluation.

    :param name: Permission name in the format <app_label>.<action>_<model>
    """
    app_label, action, model_name = resolve_permission(name)

    if action == "view":
        if (
            # All models (excluding those in EXEMPT_EXCLUDE_MODELS) are exempt from view permission enforcement
            "*" in settings.EXEMPT_VIEW_PERMISSIONS
            and (app_label, model_name) not in settings.EXEMPT_EXCLUDE_MODELS
        ) or (
            # This specific model is exempt from view permission enforcement
            f"{app_label}.{model_name}"
            in settings.EXEMPT_VIEW_PERMISSIONS
        ):
            return True

    return False

nautobot.apps.utils.populate_model_features_registry(refresh=False)

Populate the registry model features with new apps.

This function updates the registry model features.

Behavior: - Defines a list of dictionaries called lookup_confs. Each dictionary contains: - 'feature_name': The name of the feature to be updated in the registry. - 'field_names': A list of names of fields that must be present in order for the model to be considered a valid model_feature. - 'field_attributes': Optional dictionary of attributes to filter the fields by. Only model which fields match all the attributes specified in the dictionary will be considered. This parameter can be useful to narrow down the search for fields that match certain criteria. For example, if field_attributes is set to {"related_model": RelationshipAssociation}, only fields with a related model of RelationshipAssociation will be considered. - Looks up all the models in the installed apps. - For each dictionary in lookup_confs, calls lookup_by_field() function to look for all models that have fields with the names given in the dictionary. - Groups the results by app and updates the registry model features for each app.

Source code in nautobot/extras/utils.py 
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
def populate_model_features_registry(refresh=False):
    """
    Populate the registry model features with new apps.

    This function updates the registry model features.

    Behavior:
    - Defines a list of dictionaries called lookup_confs. Each dictionary contains:
        - 'feature_name': The name of the feature to be updated in the registry.
        - 'field_names': A list of names of fields that must be present in order for the model to be considered
                        a valid model_feature.
        - 'field_attributes': Optional dictionary of attributes to filter the fields by. Only model which fields match
                            all the attributes specified in the dictionary will be considered. This parameter can be
                            useful to narrow down the search for fields that match certain criteria. For example, if
                            `field_attributes` is set to {"related_model": RelationshipAssociation}, only fields with
                            a related model of RelationshipAssociation will be considered.
    - Looks up all the models in the installed apps.
    - For each dictionary in lookup_confs, calls lookup_by_field() function to look for all models that have fields with the names given in the dictionary.
    - Groups the results by app and updates the registry model features for each app.
    """
    if registry.get("populate_model_features_registry_called", False) and not refresh:
        return

    RelationshipAssociation = apps.get_model(app_label="extras", model_name="relationshipassociation")

    lookup_confs = [
        {
            "feature_name": "custom_fields",
            "field_names": ["_custom_field_data"],
        },
        {
            "feature_name": "relationships",
            "field_names": ["source_for_associations", "destination_for_associations"],
            "field_attributes": {"related_model": RelationshipAssociation},
        },
    ]

    app_models = apps.get_models()
    for lookup_conf in lookup_confs:
        registry_items = find_models_with_matching_fields(
            app_models=app_models,
            field_names=lookup_conf["field_names"],
            field_attributes=lookup_conf.get("field_attributes"),
        )
        feature_name = lookup_conf["feature_name"]
        registry["model_features"][feature_name] = registry_items

    if not registry.get("populate_model_features_registry_called", False):
        registry["populate_model_features_registry_called"] = True

nautobot.apps.utils.refresh_job_model_from_job_class(job_model_class, job_class)

Create or update a job_model record based on the metadata of the provided job_class.

Note that job_model_class is a parameter (rather than doing a "from nautobot.extras.models import Job") because this function may be called from various initialization processes (such as the "nautobot_database_ready" signal) and in that case we need to not import models ourselves.

Source code in nautobot/extras/utils.py 
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
def refresh_job_model_from_job_class(job_model_class, job_class):
    """
    Create or update a job_model record based on the metadata of the provided job_class.

    Note that job_model_class is a parameter (rather than doing a "from nautobot.extras.models import Job") because
    this function may be called from various initialization processes (such as the "nautobot_database_ready" signal)
    and in that case we need to not import models ourselves.
    """
    from nautobot.extras.jobs import (
        JobHookReceiver,
        JobButtonReceiver,
    )  # imported here to prevent circular import problem

    # Unrecoverable errors
    if len(job_class.__module__) > JOB_MAX_NAME_LENGTH:
        logger.error(
            'Unable to store Jobs from module "%s" as Job models because the module exceeds %d characters in length!',
            job_class.__module__,
            JOB_MAX_NAME_LENGTH,
        )
        return (None, False)
    if len(job_class.__name__) > JOB_MAX_NAME_LENGTH:
        logger.error(
            'Unable to represent Job class "%s" as a Job model because the class name exceeds %d characters in length!',
            job_class.__name__,
            JOB_MAX_NAME_LENGTH,
        )
        return (None, False)
    if issubclass(job_class, JobHookReceiver) and issubclass(job_class, JobButtonReceiver):
        logger.error(
            'Job class "%s" must not sub-class from both JobHookReceiver and JobButtonReceiver!',
            job_class.__name__,
        )
        return (None, False)

    # Recoverable errors
    if len(job_class.grouping) > JOB_MAX_GROUPING_LENGTH:
        logger.warning(
            'Job class "%s" grouping "%s" exceeds %d characters in length, it will be truncated in the database.',
            job_class.__name__,
            job_class.grouping,
            JOB_MAX_GROUPING_LENGTH,
        )
    if len(job_class.name) > JOB_MAX_NAME_LENGTH:
        logger.warning(
            'Job class "%s" name "%s" exceeds %d characters in length, it will be truncated in the database.',
            job_class.__name__,
            job_class.name,
            JOB_MAX_NAME_LENGTH,
        )

    # handle duplicate names by appending an incrementing counter to the end
    default_job_name = job_class.name[:JOB_MAX_NAME_LENGTH]
    job_name = default_job_name
    append_counter = 2
    existing_job_names = (
        job_model_class.objects.filter(name__startswith=job_name)
        .exclude(
            module_name=job_class.__module__[:JOB_MAX_NAME_LENGTH],
            job_class_name=job_class.__name__[:JOB_MAX_NAME_LENGTH],
        )
        .values_list("name", flat=True)
    )
    while job_name in existing_job_names:
        job_name_append = f" ({append_counter})"
        max_name_length = JOB_MAX_NAME_LENGTH - len(job_name_append)
        job_name = default_job_name[:max_name_length] + job_name_append
        append_counter += 1
    if job_name != default_job_name and "test" not in sys.argv:
        logger.warning(
            'Job class "%s" name "%s" is not unique, changing to "%s".',
            job_class.__name__,
            default_job_name,
            job_name,
        )

    try:
        with transaction.atomic():
            job_model, created = job_model_class.objects.get_or_create(
                module_name=job_class.__module__[:JOB_MAX_NAME_LENGTH],
                job_class_name=job_class.__name__[:JOB_MAX_NAME_LENGTH],
                defaults={
                    "grouping": job_class.grouping[:JOB_MAX_GROUPING_LENGTH],
                    "name": job_name,
                    "is_job_hook_receiver": issubclass(job_class, JobHookReceiver),
                    "is_job_button_receiver": issubclass(job_class, JobButtonReceiver),
                    "read_only": job_class.read_only,
                    "supports_dryrun": job_class.supports_dryrun,
                    "installed": True,
                    "enabled": False,
                },
            )

            if job_name != default_job_name:
                job_model.name_override = True

            if created and job_model.module_name.startswith("nautobot."):
                # System jobs should be enabled by default when first created
                job_model.enabled = True

            for field_name in JOB_OVERRIDABLE_FIELDS:
                # Was this field directly inherited from the job before, or was it overridden in the database?
                if not getattr(job_model, f"{field_name}_override", False):
                    # It was inherited and not overridden
                    setattr(job_model, field_name, getattr(job_class, field_name))

            if not created:
                # Mark it as installed regardless
                job_model.installed = True
                # Update the non-overridable flags in case they've changed in the source
                job_model.is_job_hook_receiver = issubclass(job_class, JobHookReceiver)
                job_model.is_job_button_receiver = issubclass(job_class, JobButtonReceiver)
                job_model.read_only = job_class.read_only
                job_model.supports_dryrun = job_class.supports_dryrun

            job_model.save()

    except Exception as exc:
        logger.error(
            'Exception while trying to create/update a database record for Job class "%s": %s', job_class.__name__, exc
        )
        return (None, False)

    logger.info(
        '%s Job "%s: %s" from <%s>',
        "Created" if created else "Refreshed",
        job_model.grouping,
        job_model.name,
        job_class.__name__,
    )

    return (job_model, created)

nautobot.apps.utils.remove_prefix_from_cf_key(field_name)

field_name (str): f"cf_{cf.key}"

Helper method to remove the "cf_" prefix

Source code in nautobot/extras/utils.py 
466
467
468
469
470
471
472
def remove_prefix_from_cf_key(field_name):
    """
    field_name (str): f"cf_{cf.key}"

    Helper method to remove the "cf_" prefix
    """
    return field_name[3:]

nautobot.apps.utils.render_jinja2(template_code, context)

Render a Jinja2 template with the provided context. Return the rendered content.

Source code in nautobot/core/utils/data.py 
 97
 98
 99
100
101
102
103
104
105
106
107
def render_jinja2(template_code, context):
    """
    Render a Jinja2 template with the provided context. Return the rendered content.
    """
    rendering_engine = engines["jinja"]
    template = rendering_engine.from_string(template_code)
    # For reasons unknown to me, django-jinja2 `template.render()` implicitly calls `mark_safe()` on the rendered text.
    # This is a security risk in general, especially so in our case because we're often using this function to render
    # a user-provided template and don't want to open ourselves up to script injection or similar issues.
    # There's no `mark_unsafe()` function, but concatenating a SafeString to an ordinary string (even "") suffices.
    return "" + template.render(context=context)

nautobot.apps.utils.resolve_permission(name)

Given a permission name, return the app_label, action, and model_name components. For example, "dcim.view_location" returns ("dcim", "view", "location").

:param name: Permission name in the format ._

Source code in nautobot/core/utils/permissions.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def resolve_permission(name):
    """
    Given a permission name, return the app_label, action, and model_name components. For example, "dcim.view_location"
    returns ("dcim", "view", "location").

    :param name: Permission name in the format <app_label>.<action>_<model>
    """
    try:
        app_label, codename = name.split(".")
        action, model_name = codename.rsplit("_", 1)
    except ValueError:
        raise ValueError(f"Invalid permission name: {name}. Must be in the format <app_label>.<action>_<model>")

    return app_label, action, model_name

nautobot.apps.utils.resolve_permission_ct(name)

Given a permission name, return the relevant ContentType and action. For example, "dcim.view_location" returns (Location, "view").

:param name: Permission name in the format ._

Source code in nautobot/core/utils/permissions.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def resolve_permission_ct(name):
    """
    Given a permission name, return the relevant ContentType and action. For example, "dcim.view_location" returns
    (Location, "view").

    :param name: Permission name in the format <app_label>.<action>_<model>
    """
    app_label, action, model_name = resolve_permission(name)
    try:
        content_type = ContentType.objects.get(app_label=app_label, model=model_name)
    except ContentType.DoesNotExist:
        raise ValueError(f"Unknown app_label/model_name for {name}")

    return content_type, action

nautobot.apps.utils.rgb_to_hex(r, g, b)

Map r, g, b values to a hex string.

Source code in nautobot/core/utils/color.py 
11
12
13
14
15
def rgb_to_hex(r, g, b):
    """
    Map r, g, b values to a hex string.
    """
    return "%02x%02x%02x" % (r, g, b)  # pylint: disable=consider-using-f-string

nautobot.apps.utils.sanitize(string, replacement='(redacted)')

Make an attempt at stripping potentially-sensitive information from the given string.

Obviously this will never be 100% foolproof but we can at least try.

Uses settings.SANITIZER_PATTERNS as the list of (regexp, repl) tuples to apply.

Source code in nautobot/core/utils/logging.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
def sanitize(string, replacement="(redacted)"):
    """
    Make an attempt at stripping potentially-sensitive information from the given string.

    Obviously this will never be 100% foolproof but we can at least try.

    Uses settings.SANITIZER_PATTERNS as the list of (regexp, repl) tuples to apply.
    """
    # Don't allow regex match groups to be referenced in the replacement string!
    assert not re.search(r"\\\d|\\g<\d+>", replacement)

    for sanitizer, repl in settings.SANITIZER_PATTERNS:
        try:
            string = sanitizer.sub(repl.format(replacement=replacement), string)
        except re.error:
            logger.error('Error in string sanitization using "%s"', sanitizer)

    return string

nautobot.apps.utils.shallow_compare_dict(source_dict, destination_dict, exclude=None)

Return a new dictionary of the different keys. The values of destination_dict are returned. Only the equality of the first layer of keys/values is checked. exclude is a list or tuple of keys to be ignored.

Source code in nautobot/core/utils/data.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def shallow_compare_dict(source_dict, destination_dict, exclude=None):
    """
    Return a new dictionary of the different keys. The values of `destination_dict` are returned. Only the equality of
    the first layer of keys/values is checked. `exclude` is a list or tuple of keys to be ignored.
    """
    difference = {}

    for key in destination_dict:
        if source_dict.get(key) != destination_dict[key]:
            if isinstance(exclude, (list, tuple)) and key in exclude:
                continue
            difference[key] = destination_dict[key]

    return difference

nautobot.apps.utils.swap_status_initials(data)

Swap Git status initials with its equivalent.

Source code in nautobot/core/utils/git.py 
33
34
35
36
def swap_status_initials(data):
    """Swap Git status initials with its equivalent."""
    initial, text = data.split("\t")
    return GitDiffLog(status=GIT_STATUS_MAP.get(initial), text=text)

nautobot.apps.utils.task_queues_as_choices(task_queues)

Returns a list of 2-tuples for use in the form field choices argument. Appends worker count to the description.

Source code in nautobot/extras/utils.py 
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
def task_queues_as_choices(task_queues):
    """
    Returns a list of 2-tuples for use in the form field `choices` argument. Appends
    worker count to the description.
    """
    if not task_queues:
        task_queues = [settings.CELERY_TASK_DEFAULT_QUEUE]

    choices = []
    celery_queues = get_celery_queues()
    for queue in task_queues:
        if not queue:
            worker_count = celery_queues.get(settings.CELERY_TASK_DEFAULT_QUEUE, 0)
        else:
            worker_count = celery_queues.get(queue, 0)
        description = f"{queue if queue else 'default queue'} ({worker_count} worker{'s'[:worker_count^1]})"
        choices.append((queue, description))
    return choices

nautobot.apps.utils.to_meters(length, unit)

Convert the given length to meters.

Source code in nautobot/core/utils/data.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def to_meters(length, unit):
    """
    Convert the given length to meters.
    """
    length = int(length)
    if length < 0:
        raise ValueError("Length must be a positive integer")

    valid_units = choices.CableLengthUnitChoices.values()
    if unit not in valid_units:
        raise ValueError(f"Unknown unit {unit}. Must be one of the following: {', '.join(valid_units)}")

    if unit == choices.CableLengthUnitChoices.UNIT_METER:
        return length
    if unit == choices.CableLengthUnitChoices.UNIT_CENTIMETER:
        return length / 100
    if unit == choices.CableLengthUnitChoices.UNIT_FOOT:
        return length * Decimal("0.3048")
    if unit == choices.CableLengthUnitChoices.UNIT_INCH:
        return length * Decimal("0.3048") * 12
    raise ValueError(f"Unknown unit {unit}. Must be 'm', 'cm', 'ft', or 'in'.")

nautobot.apps.utils.wrap_model_clean_methods()

Helper function that wraps plugin model validator registered clean methods for all applicable models

Source code in nautobot/extras/plugins/validators.py 
39
40
41
42
43
44
45
46
def wrap_model_clean_methods():
    """
    Helper function that wraps plugin model validator registered clean methods for all applicable models
    """
    for app_label, models in FeatureQuery("custom_validators").as_dict():
        for model in models:
            model_class = apps.get_model(app_label=app_label, model_name=model)
            model_class.clean = custom_validator_clean(model_class.clean)