NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-08-08 00:28:16 -06:00
Code Issues Actions 15 Packages Projects Releases Wiki Activity

17034 refactor from review feedback

Browse Source
This commit is contained in:
Arthur Hanson 2024-07-31 20:46:58 +07:00
parent f53ff24885
commit 0440e7e7b9
3 changed files with 96 additions and 92 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

158
netbox/core/plugins.py
View File

@ -69,16 +69,16 @@ class Plugin:
installed_version: str = '' installed_version: str = ''
def get_local_plugins(): def get_local_plugins(plugins):
""" """
Return a dictionary of all locally-installed plugins, mapped by name. Return a dictionary of all locally-installed plugins, mapped by name.
""" """
plugins = {} local_plugins = {}
for plugin_name in settings.PLUGINS: for plugin_name in settings.PLUGINS:
plugin = importlib.import_module(plugin_name) plugin = importlib.import_module(plugin_name)
plugin_config: PluginConfig = plugin.config plugin_config: PluginConfig = plugin.config
plugins[plugin_config.name] = Plugin( local_plugins[plugin_config.name] = Plugin(
slug=plugin_config.name, slug=plugin_config.name,
title_short=plugin_config.verbose_name, title_short=plugin_config.verbose_name,
tag_line=plugin_config.description, tag_line=plugin_config.description,
@ -88,6 +88,13 @@ def get_local_plugins():
installed_version=plugin_config.version, installed_version=plugin_config.version,
) )
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 return plugins
@ -96,7 +103,6 @@ def get_catalog_plugins():
Return a dictionary of all entries in the plugins catalog, mapped by name. Return a dictionary of all entries in the plugins catalog, mapped by name.
""" """
session = requests.Session() session = requests.Session()
plugins = {}
def get_pages(): def get_pages():
# TODO: pagination is currently broken in API # TODO: pagination is currently broken in API
@ -122,94 +128,80 @@ def get_catalog_plugins():
).json() ).json()
yield next_page yield next_page
for page in get_pages(): def make_plugin_dict():
for data in page['data']: plugins = {}
# Populate releases for page in get_pages():
releases = [] for data in page['data']:
for version in data['release_recent_history']:
releases.append( # Populate releases
PluginVersion( releases = []
date=datetime_from_timestamp(version['date']), for version in data['release_recent_history']:
version=version['version'], releases.append(
netbox_min_version=version['netbox_min_version'], PluginVersion(
netbox_max_version=version['netbox_max_version'], date=datetime_from_timestamp(version['date']),
has_model=version['has_model'], version=version['version'],
is_certified=version['is_certified'], netbox_min_version=version['netbox_min_version'],
is_feature=version['is_feature'], netbox_max_version=version['netbox_max_version'],
is_integration=version['is_integration'], has_model=version['has_model'],
is_netboxlabs_supported=version['is_netboxlabs_supported'], 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'],
) )
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) # Populate author (if any)
if data['author']: if data['author']:
author = PluginAuthor( author = PluginAuthor(
name=data['author']['name'], name=data['author']['name'],
org_id=data['author']['org_id'], org_id=data['author']['org_id'],
url=data['author']['url'], 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,
) )
else:
author = None
# Populate plugin data return plugins
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(request):
"""
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', default={}) catalog_plugins = cache.get('plugins-catalog-feed', default={})
catalog_plugins_error = cache.get('plugins-catalog-error', default=False) if not catalog_plugins:
if not catalog_plugins and not catalog_plugins_error:
try: try:
catalog_plugins = get_catalog_plugins() catalog_plugins = make_plugin_dict()
cache.set('plugins-catalog-feed', catalog_plugins, 3600) cache.set('plugins-catalog-feed', catalog_plugins, 3600)
except requests.exceptions.RequestException: except requests.exceptions.RequestException:
# Cache for 15 minutes to avoid spamming connection pass
cache.set('plugins-catalog-error', True, 900)
messages.warning(request, _("Plugins catalog could not be loaded"))
plugins = catalog_plugins return 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

28
netbox/core/views.py
View File

@ -37,7 +37,7 @@ from . import filtersets, forms, tables
from .choices import DataSourceStatusChoices from .choices import DataSourceStatusChoices
from .jobs import SyncDataSourceJob from .jobs import SyncDataSourceJob
from .models import * from .models import *
from .plugins import get_plugins from .plugins import get_catalog_plugins, get_local_plugins
from .tables import CatalogPluginTable, PluginVersionTable from .tables import CatalogPluginTable, PluginVersionTable
@ -650,15 +650,30 @@ class SystemView(UserPassesTestMixin, View):
# Plugins # Plugins
# #
class PluginListView(UserPassesTestMixin, View): class BasePluginView(UserPassesTestMixin, View):
def test_func(self): def test_func(self):
return self.request.user.is_staff return self.request.user.is_staff
def get_cached_plugins(self, request):
catalog_plugins = {}
catalog_plugins_error = cache.get('plugins-catalog-error', default=False)
if not catalog_plugins_error:
catalog_plugins = get_catalog_plugins()
if not catalog_plugins:
# Cache for 5 minutes to avoid spamming connection
cache.set('plugins-catalog-error', True, 300)
messages.warning(request, _("Plugins catalog could not be loaded"))
return get_local_plugins(catalog_plugins)
class PluginListView(BasePluginView):
def get(self, request): def get(self, request):
q = request.GET.get('q', None) q = request.GET.get('q', None)
plugins = get_plugins(request).values() plugins = self.get_cached_plugins(request).values()
if q: if q:
plugins = [obj for obj in plugins if q.casefold() in obj.title_short.casefold()] plugins = [obj for obj in plugins if q.casefold() in obj.title_short.casefold()]
@ -676,14 +691,11 @@ class PluginListView(UserPassesTestMixin, View):
}) })
class PluginView(UserPassesTestMixin, View): class PluginView(BasePluginView):
def test_func(self):
return self.request.user.is_staff
def get(self, request, name): def get(self, request, name):
plugins = get_plugins(request) plugins = self.get_cached_plugins(request)
if name not in plugins: if name not in plugins:
raise Http404(_("Plugin {name} not found").format(name=name)) raise Http404(_("Plugin {name} not found").format(name=name))
plugin = plugins[name] plugin = plugins[name]

2
netbox/netbox/settings.py
View File

@ -774,7 +774,7 @@ STRAWBERRY_DJANGO = {
# Plugins # Plugins
# #
PLUGIN_CATALOG_URL = 'https://api.netbox.oss.netboxlabs.com/v1/plugins' PLUGIN_CATALOG_URL = 'httpxs://api.netbox.oss.netboxlabs.com/v1/plugins'
# Register any configured plugins # Register any configured plugins
for plugin_name in PLUGINS: for plugin_name in PLUGINS: