14731 plugins catalog

netbox/core/plugins.py

@@ -0,0 +1,195 @@
+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
+
+
+@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, 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, 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

netbox/core/tables/plugins.py

@@ -1,39 +1,80 @@
 import django_tables2 as tables
 from django.utils.translation import gettext_lazy as _
-from netbox.tables import BaseTable
+
+from netbox.tables import BaseTable, columns
 
 __all__ = (
-    'PluginTable',
+    'CatalogPluginTable',
+    'PluginVersionTable',
 )
 
 
-class PluginTable(BaseTable):
-    name = tables.Column(
-        accessor=tables.A('verbose_name'),
-        verbose_name=_('Name')
-    )
+class PluginVersionTable(BaseTable):
     version = tables.Column(
         verbose_name=_('Version')
     )
-    package = tables.Column(
-        accessor=tables.A('name'),
-        verbose_name=_('Package')
+    last_updated = columns.DateTimeColumn(
+        accessor=tables.A('date'),
+        timespec='minutes',
+        verbose_name=_('Last Updated')
+    )
+    min_version = tables.Column(
+        accessor=tables.A('netbox_min_version'),
+        verbose_name=_('Minimum NetBox Version')
+    )
+    max_version = tables.Column(
+        accessor=tables.A('netbox_max_version'),
+        verbose_name=_('Maximum NetBox Version')
+    )
+
+    class Meta(BaseTable.Meta):
+        empty_text = _('No plugin data found')
+        fields = (
+            'version', 'last_updated', 'min_version', 'max_version',
+        )
+        default_columns = (
+            'version', 'last_updated', 'min_version', 'max_version',
+        )
+        orderable = False
+
+
+class CatalogPluginTable(BaseTable):
+    title_short = tables.Column(
+        linkify=('core:plugin', [tables.A('slug')]),
+        verbose_name=_('Name')
     )
     author = tables.Column(
+        accessor=tables.A('author.name'),
         verbose_name=_('Author')
     )
-    author_email = tables.Column(
-        verbose_name=_('Author Email')
+    is_local = columns.BooleanColumn(
+        verbose_name=_('Local')
+    )
+    is_installed = columns.BooleanColumn(
+        verbose_name=_('Installed')
+    )
+    is_certified = columns.BooleanColumn(
+        verbose_name=_('Certified')
+    )
+    created_at = columns.DateTimeColumn(
+        verbose_name=_('Published')
+    )
+    updated_at = columns.DateTimeColumn(
+        verbose_name=_('Updated')
     )
-    description = tables.Column(
-        verbose_name=_('Description')
+    installed_version = tables.Column(
+        verbose_name=_('Installed version')
     )
 
     class Meta(BaseTable.Meta):
-        empty_text = _('No plugins found')
+        empty_text = _('No plugin data found')
         fields = (
-            'name', 'version', 'package', 'author', 'author_email', 'description',
+            'title_short', 'author', 'is_local', 'is_installed', 'is_certified', 'created_at', 'updated_at',
+            'installed_version',
         )
         default_columns = (
-            'name', 'version', 'package', 'description',
+            'title_short', 'author', 'is_local', 'is_installed', 'is_certified', 'created_at', 'updated_at',
         )
+        # List installed plugins first, then certified plugins, then
+        # everything else (with each tranche ordered alphabetically)
+        order_by = ('-is_installed', '-is_certified', 'name')

netbox/core/urls.py

@@ -49,4 +49,8 @@
 
     # System
     path('system/', views.SystemView.as_view(), name='system'),
+
+    # Plugins
+    path('plugins/', views.PluginListView.as_view(), name='plugin_list'),
+    path('plugins/<str:name>/', views.PluginView.as_view(), name='plugin'),
 )