|
|
@@ -0,0 +1,209 @@
|
|
|
+import datetime
|
|
|
+import importlib
|
|
|
+import importlib.util
|
|
|
+from dataclasses import dataclass, field
|
|
|
+from typing import Optional
|
|
|
+
|
|
|
+import requests
|
|
|
+from django.conf import settings
|
|
|
+from django.core.cache import cache
|
|
|
+from django.utils.translation import gettext_lazy as _
|
|
|
+
|
|
|
+from netbox.plugins import PluginConfig
|
|
|
+from utilities.datetime import datetime_from_timestamp
|
|
|
+
|
|
|
+USER_AGENT_STRING = f'NetBox/{settings.RELEASE.version} {settings.RELEASE.edition}'
|
|
|
+
|
|
|
+
|
|
|
+@dataclass
|
|
|
+class PluginAuthor:
|
|
|
+ """
|
|
|
+ Identifying information for the author of a plugin.
|
|
|
+ """
|
|
|
+ name: str
|
|
|
+ org_id: str = ''
|
|
|
+ url: str = ''
|
|
|
+
|
|
|
+
|
|
|
+@dataclass
|
|
|
+class PluginVersion:
|
|
|
+ """
|
|
|
+ Details for a specific versioned release of a plugin.
|
|
|
+ """
|
|
|
+ date: datetime.datetime = None
|
|
|
+ version: str = ''
|
|
|
+ netbox_min_version: str = ''
|
|
|
+ netbox_max_version: str = ''
|
|
|
+ has_model: bool = False
|
|
|
+ is_certified: bool = False
|
|
|
+ is_feature: bool = False
|
|
|
+ is_integration: bool = False
|
|
|
+ is_netboxlabs_supported: bool = False
|
|
|
+
|
|
|
+
|
|
|
+@dataclass
|
|
|
+class Plugin:
|
|
|
+ """
|
|
|
+ The representation of a NetBox plugin in the catalog API.
|
|
|
+ """
|
|
|
+ id: str = ''
|
|
|
+ status: str = ''
|
|
|
+ title_short: str = ''
|
|
|
+ title_long: str = ''
|
|
|
+ tag_line: str = ''
|
|
|
+ description_short: str = ''
|
|
|
+ slug: str = ''
|
|
|
+ author: Optional[PluginAuthor] = None
|
|
|
+ created_at: datetime.datetime = None
|
|
|
+ updated_at: datetime.datetime = None
|
|
|
+ license_type: str = ''
|
|
|
+ homepage_url: str = ''
|
|
|
+ package_name_pypi: str = ''
|
|
|
+ config_name: str = ''
|
|
|
+ is_certified: bool = False
|
|
|
+ release_latest: PluginVersion = field(default_factory=PluginVersion)
|
|
|
+ release_recent_history: list[PluginVersion] = field(default_factory=list)
|
|
|
+ is_local: bool = False # extra field for locally installed plugins
|
|
|
+ is_installed: bool = False
|
|
|
+ installed_version: str = ''
|
|
|
+
|
|
|
+
|
|
|
+def get_local_plugins():
|
|
|
+ """
|
|
|
+ Return a dictionary of all locally-installed plugins, mapped by name.
|
|
|
+ """
|
|
|
+ plugins = {}
|
|
|
+ for plugin_name in settings.PLUGINS:
|
|
|
+ plugin = importlib.import_module(plugin_name)
|
|
|
+ plugin_config: PluginConfig = plugin.config
|
|
|
+
|
|
|
+ plugins[plugin_config.name] = Plugin(
|
|
|
+ slug=plugin_config.name,
|
|
|
+ title_short=plugin_config.verbose_name,
|
|
|
+ tag_line=plugin_config.description,
|
|
|
+ description_short=plugin_config.description,
|
|
|
+ is_local=True,
|
|
|
+ is_installed=True,
|
|
|
+ installed_version=plugin_config.version,
|
|
|
+ )
|
|
|
+
|
|
|
+ return plugins
|
|
|
+
|
|
|
+
|
|
|
+def get_catalog_plugins():
|
|
|
+ """
|
|
|
+ Return a dictionary of all entries in the plugins catalog, mapped by name.
|
|
|
+ """
|
|
|
+ session = requests.Session()
|
|
|
+ plugins = {}
|
|
|
+
|
|
|
+ def get_pages():
|
|
|
+ # TODO: pagination is currently broken in API
|
|
|
+ payload = {'page': '1', 'per_page': '50'}
|
|
|
+ first_page = session.get(
|
|
|
+ settings.PLUGIN_CATALOG_URL,
|
|
|
+ headers={'User-Agent': USER_AGENT_STRING},
|
|
|
+ proxies=settings.HTTP_PROXIES,
|
|
|
+ timeout=3,
|
|
|
+ params=payload
|
|
|
+ ).json()
|
|
|
+ yield first_page
|
|
|
+ num_pages = first_page['metadata']['pagination']['last_page']
|
|
|
+
|
|
|
+ for page in range(2, num_pages + 1):
|
|
|
+ payload['page'] = page
|
|
|
+ next_page = session.get(
|
|
|
+ settings.PLUGIN_CATALOG_URL,
|
|
|
+ headers={'User-Agent': USER_AGENT_STRING},
|
|
|
+ proxies=settings.HTTP_PROXIES,
|
|
|
+ timeout=3,
|
|
|
+ params=payload
|
|
|
+ ).json()
|
|
|
+ yield next_page
|
|
|
+
|
|
|
+ for page in get_pages():
|
|
|
+ for data in page['data']:
|
|
|
+
|
|
|
+ # Populate releases
|
|
|
+ releases = []
|
|
|
+ for version in data['release_recent_history']:
|
|
|
+ releases.append(
|
|
|
+ PluginVersion(
|
|
|
+ date=datetime_from_timestamp(version['date']),
|
|
|
+ version=version['version'],
|
|
|
+ netbox_min_version=version['netbox_min_version'],
|
|
|
+ netbox_max_version=version['netbox_max_version'],
|
|
|
+ has_model=version['has_model'],
|
|
|
+ is_certified=version['is_certified'],
|
|
|
+ is_feature=version['is_feature'],
|
|
|
+ is_integration=version['is_integration'],
|
|
|
+ is_netboxlabs_supported=version['is_netboxlabs_supported'],
|
|
|
+ )
|
|
|
+ )
|
|
|
+ releases = sorted(releases, key=lambda x: x.date, reverse=True)
|
|
|
+ latest_release = PluginVersion(
|
|
|
+ date=datetime_from_timestamp(data['release_latest']['date']),
|
|
|
+ version=data['release_latest']['version'],
|
|
|
+ netbox_min_version=data['release_latest']['netbox_min_version'],
|
|
|
+ netbox_max_version=data['release_latest']['netbox_max_version'],
|
|
|
+ has_model=data['release_latest']['has_model'],
|
|
|
+ is_certified=data['release_latest']['is_certified'],
|
|
|
+ is_feature=data['release_latest']['is_feature'],
|
|
|
+ is_integration=data['release_latest']['is_integration'],
|
|
|
+ is_netboxlabs_supported=data['release_latest']['is_netboxlabs_supported'],
|
|
|
+ )
|
|
|
+
|
|
|
+ # Populate author (if any)
|
|
|
+ if data['author']:
|
|
|
+ print(data['author'])
|
|
|
+ author = PluginAuthor(
|
|
|
+ name=data['author']['name'],
|
|
|
+ org_id=data['author']['org_id'],
|
|
|
+ url=data['author']['url'],
|
|
|
+ )
|
|
|
+ else:
|
|
|
+ author = None
|
|
|
+
|
|
|
+ # Populate plugin data
|
|
|
+ plugins[data['slug']] = Plugin(
|
|
|
+ id=data['id'],
|
|
|
+ status=data['status'],
|
|
|
+ title_short=data['title_short'],
|
|
|
+ title_long=data['title_long'],
|
|
|
+ tag_line=data['tag_line'],
|
|
|
+ description_short=data['description_short'],
|
|
|
+ slug=data['slug'],
|
|
|
+ author=author,
|
|
|
+ created_at=datetime_from_timestamp(data['created_at']),
|
|
|
+ updated_at=datetime_from_timestamp(data['updated_at']),
|
|
|
+ license_type=data['license_type'],
|
|
|
+ homepage_url=data['homepage_url'],
|
|
|
+ package_name_pypi=data['package_name_pypi'],
|
|
|
+ config_name=data['config_name'],
|
|
|
+ is_certified=data['is_certified'],
|
|
|
+ release_latest=latest_release,
|
|
|
+ release_recent_history=releases,
|
|
|
+ )
|
|
|
+
|
|
|
+ return plugins
|
|
|
+
|
|
|
+
|
|
|
+def get_plugins():
|
|
|
+ """
|
|
|
+ Return a dictionary of all plugins (both catalog and locally installed), mapped by name.
|
|
|
+ """
|
|
|
+ local_plugins = get_local_plugins()
|
|
|
+ catalog_plugins = cache.get('plugins-catalog-feed')
|
|
|
+ if not catalog_plugins:
|
|
|
+ catalog_plugins = get_catalog_plugins()
|
|
|
+ cache.set('plugins-catalog-feed', catalog_plugins, 3600)
|
|
|
+
|
|
|
+ plugins = catalog_plugins
|
|
|
+ for k, v in local_plugins.items():
|
|
|
+ if k in plugins:
|
|
|
+ plugins[k].is_local = True
|
|
|
+ plugins[k].is_installed = True
|
|
|
+ else:
|
|
|
+ plugins[k] = v
|
|
|
+
|
|
|
+ return plugins
|