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 netbox.plugins import PluginConfig from utilities.datetime import datetime_from_timestamp USER_AGENT_STRING = f'NetBox/{settings.RELEASE.version} {settings.RELEASE.edition}' CACHE_KEY_CATALOG_FEED = 'plugins-catalog-feed' @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(plugins=None): """ Return a dictionary of all locally-installed plugins, mapped by name. """ plugins = plugins or {} local_plugins = {} # Gather all locally-installed plugins for plugin_name in settings.PLUGINS: plugin = importlib.import_module(plugin_name) plugin_config: PluginConfig = plugin.config local_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, ) # Update catalog entries for local plugins, or add them to the list if not listed 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 def get_catalog_plugins(): """ Return a dictionary of all entries in the plugins catalog, mapped by name. """ session = requests.Session() # Disable catalog fetching for isolated deployments if settings.ISOLATED_DEPLOYMENT: return {} 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 def make_plugin_dict(): plugins = {} 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']: 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 catalog_plugins = cache.get(CACHE_KEY_CATALOG_FEED, default={}) if not catalog_plugins: try: catalog_plugins = make_plugin_dict() cache.set(CACHE_KEY_CATALOG_FEED, catalog_plugins, 3600) except requests.exceptions.RequestException: pass return catalog_plugins