Skip to content

nautobot.apps.datasources

Utilities for an app to register datasource contents (for Git, etc.).

nautobot.apps.datasources.DatasourceContent

Parameters:

Name Type Description Default
name str

Human-readable name for this content type, such as "config contexts"

required
content_identifier str

Brief unique identifier of this content type; by convention a string such as "extras.configcontext"

required
icon str

Material Design Icons icon name, such as "mdi-code-json" or "mdi-script-text"

required
callback callable

Callback function to invoke whenever a given datasource is created, updated, or deleted. This callback should take three arguments (record, job_result, delete) where "record" is the GitRepository, etc. that is being refreshed, "job_result" is an extras.JobResult record for logging purposes, and "delete" is a boolean flag to distinguish between the "create/update" and "delete" cases.

required
weight int

Defines the order in which datasources will be loaded.

1000
Source code in nautobot/extras/registry.py
class DatasourceContent:
    """
    Args:
      name (str): Human-readable name for this content type, such as "config contexts"
      content_identifier (str): Brief unique identifier of this content type; by convention a string such as "extras.configcontext"
      icon (str): Material Design Icons icon name, such as "mdi-code-json" or "mdi-script-text"
      callback (callable): Callback function to invoke whenever a given datasource is created, updated, or deleted.
          This callback should take three arguments (record, job_result, delete) where "record" is the GitRepository, etc.
          that is being refreshed, "job_result" is an extras.JobResult record for logging purposes, and
          "delete" is a boolean flag to distinguish between the "create/update" and "delete" cases.
      weight (int): Defines the order in which datasources will be loaded.
    """

    __slots__ = ["name", "content_identifier", "icon", "callback", "weight"]

    def __init__(self, name, content_identifier, icon, callback, weight=1000):
        """Ensure datasource properties."""
        self.name = name
        self.content_identifier = content_identifier
        self.icon = icon
        self.callback = callback
        self.weight = weight

    def __repr__(self):
        return f"<{self.__class__.__name__}: {self.content_identifier}>"

__init__(name, content_identifier, icon, callback, weight=1000)

Ensure datasource properties.

Source code in nautobot/extras/registry.py
def __init__(self, name, content_identifier, icon, callback, weight=1000):
    """Ensure datasource properties."""
    self.name = name
    self.content_identifier = content_identifier
    self.icon = icon
    self.callback = callback
    self.weight = weight

nautobot.apps.datasources.get_repo_access_url(repository_record)

Returns the repo url with and without token when present

Returns:

Type Description
str

The url used to connect to the git repo, with credentials as applicable.

Source code in nautobot/extras/datasources/git.py
def get_repo_access_url(repository_record):
    """Returns the repo url with and without token when present
    Returns:
        (str): The url used to connect to the git repo, with credentials as applicable.
    """
    # Inject username and/or token into source URL if necessary
    from_url = repository_record.remote_url

    user = None
    token = None
    if repository_record.secrets_group:
        # In addition to ObjectDoesNotExist, get_secret_value() may also raise a SecretError if a secret is mis-defined;
        # we don't catch that here but leave it up to the caller to handle as part of general exception handling.
        try:
            token = repository_record.secrets_group.get_secret_value(
                SecretsGroupAccessTypeChoices.TYPE_HTTP,
                SecretsGroupSecretTypeChoices.TYPE_TOKEN,
                obj=repository_record,
            )
        except ObjectDoesNotExist:
            # No defined secret, fall through to legacy behavior
            pass
        try:
            user = repository_record.secrets_group.get_secret_value(
                SecretsGroupAccessTypeChoices.TYPE_HTTP,
                SecretsGroupSecretTypeChoices.TYPE_USERNAME,
                obj=repository_record,
            )
        except ObjectDoesNotExist:
            # No defined secret, fall through to legacy behavior
            pass

    if token and token not in from_url:
        # Some git repositories require a user as well as a token.
        if user:
            from_url = re.sub("//", f"//{quote(user, safe='')}:{quote(token, safe='')}@", from_url)
        else:
            from_url = re.sub("//", f"//{quote(token, safe='')}@", from_url)
    return from_url