netbox/netbox/core/views.py

import json
import platform
from copy import deepcopy

from django import __version__ as django_version
from django.apps import apps as django_apps_registry
from django.conf import settings
from django.contrib import messages
from django.contrib.auth.mixins import UserPassesTestMixin
from django.core.cache import cache
from django.db import DatabaseError, connection
from django.http import Http404, HttpResponse, HttpResponseForbidden
from django.shortcuts import get_object_or_404, redirect, render
from django.urls import reverse
from django.utils.translation import gettext_lazy as _
from django.views.generic import View
from django_rq.queues import get_connection, get_queue_by_index, get_redis_connection
from django_rq.settings import get_queues_list, get_queues_map
from django_rq.utils import get_statistics
from rq.exceptions import NoSuchJobError
from rq.job import Job as RQ_Job
from rq.job import JobStatus as RQJobStatus
from rq.worker import Worker
from rq.worker_registration import clean_worker_registry

from core.utils import (
    delete_rq_job,
    enqueue_rq_job,
    get_db_schema,
    get_rq_jobs_from_status,
    requeue_rq_job,
    stop_rq_job,
)
from extras.ui.panels import CustomFieldsPanel, TagsPanel
from netbox.config import PARAMS, get_config
from netbox.object_actions import AddObject, BulkDelete, BulkExport, DeleteObject
from netbox.plugins import PluginConfig
from netbox.plugins.utils import get_installed_plugins
from netbox.ui import layout
from netbox.ui.panels import (
    CommentsPanel,
    ContextTablePanel,
    JSONPanel,
    ObjectsTablePanel,
    PluginContentPanel,
    RelatedObjectsPanel,
    TemplatePanel,
)
from netbox.views import generic
from netbox.views.generic.base import BaseObjectView
from netbox.views.generic.mixins import TableMixin
from utilities.apps import get_installed_apps
from utilities.data import deep_compare_dict
from utilities.forms import ConfirmationForm
from utilities.htmx import htmx_partial
from utilities.json import ConfigJSONEncoder
from utilities.query import count_related
from utilities.views import (
    ContentTypePermissionRequiredMixin,
    GetRelatedModelsMixin,
    GetReturnURLMixin,
    ViewTab,
    register_model_view,
)

from . import filtersets, forms, tables
from .jobs import SyncDataSourceJob
from .models import *
from .plugins import get_catalog_plugins, get_local_plugins
from .tables import CatalogPluginTable, JobLogEntryTable, PluginVersionTable
from .ui import panels

#
# Data sources
#


@register_model_view(DataSource, 'list', path='', detail=False)
class DataSourceListView(generic.ObjectListView):
    queryset = DataSource.objects.annotate(
        file_count=count_related(DataFile, 'source')
    )
    filterset = filtersets.DataSourceFilterSet
    filterset_form = forms.DataSourceFilterForm
    table = tables.DataSourceTable


@register_model_view(DataSource)
class DataSourceView(GetRelatedModelsMixin, generic.ObjectView):
    queryset = DataSource.objects.all()
    layout = layout.SimpleLayout(
        left_panels=[
            panels.DataSourcePanel(),
            TagsPanel(),
            CommentsPanel(),
        ],
        right_panels=[
            panels.DataSourceBackendPanel(),
            RelatedObjectsPanel(),
            CustomFieldsPanel(),
        ],
        bottom_panels=[
            ObjectsTablePanel(
                model='core.DataFile',
                filters={'source_id': lambda ctx: ctx['object'].pk},
                exclude_columns=['source'],
            ),
        ],
    )

    def get_extra_context(self, request, instance):
        return {
            'related_models': self.get_related_models(request, instance),
        }


@register_model_view(DataSource, 'sync')
class DataSourceSyncView(GetReturnURLMixin, BaseObjectView):
    queryset = DataSource.objects.all()

    def get_required_permission(self):
        return 'core.sync_datasource'

    def get(self, request, pk):
        # Redirect GET requests to the object view
        datasource = get_object_or_404(self.queryset, pk=pk)
        return redirect(datasource.get_absolute_url())

    def post(self, request, pk):
        datasource = get_object_or_404(self.queryset, pk=pk)
        # Enqueue the sync job
        job = SyncDataSourceJob.enqueue(instance=datasource, user=request.user)
        messages.success(
            request,
            _("Queued job #{id} to sync {datasource}").format(id=job.pk, datasource=datasource)
        )
        return redirect(self.get_return_url(request, datasource))


@register_model_view(DataSource, 'add', detail=False)
@register_model_view(DataSource, 'edit')
class DataSourceEditView(generic.ObjectEditView):
    queryset = DataSource.objects.all()
    form = forms.DataSourceForm


@register_model_view(DataSource, 'delete')
class DataSourceDeleteView(generic.ObjectDeleteView):
    queryset = DataSource.objects.all()


@register_model_view(DataSource, 'bulk_import', path='import', detail=False)
class DataSourceBulkImportView(generic.BulkImportView):
    queryset = DataSource.objects.all()
    model_form = forms.DataSourceImportForm


@register_model_view(DataSource, 'bulk_edit', path='edit', detail=False)
class DataSourceBulkEditView(generic.BulkEditView):
    queryset = DataSource.objects.annotate(
        count_files=count_related(DataFile, 'source')
    )
    filterset = filtersets.DataSourceFilterSet
    table = tables.DataSourceTable
    form = forms.DataSourceBulkEditForm


@register_model_view(DataSource, 'bulk_rename', path='rename', detail=False)
class DataSourceBulkRenameView(generic.BulkRenameView):
    queryset = DataSource.objects.all()
    filterset = filtersets.DataSourceFilterSet


@register_model_view(DataSource, 'bulk_delete', path='delete', detail=False)
class DataSourceBulkDeleteView(generic.BulkDeleteView):
    queryset = DataSource.objects.annotate(
        count_files=count_related(DataFile, 'source')
    )
    filterset = filtersets.DataSourceFilterSet
    table = tables.DataSourceTable


#
# Data files
#

@register_model_view(DataFile, 'list', path='', detail=False)
class DataFileListView(generic.ObjectListView):
    queryset = DataFile.objects.defer('data')
    filterset = filtersets.DataFileFilterSet
    filterset_form = forms.DataFileFilterForm
    table = tables.DataFileTable
    actions = (BulkDelete,)


@register_model_view(DataFile)
class DataFileView(generic.ObjectView):
    queryset = DataFile.objects.all()
    actions = (DeleteObject,)
    layout = layout.Layout(
        layout.Row(
            layout.Column(
                panels.DataFilePanel(),
                panels.DataFileContentPanel(),
                PluginContentPanel('left_page'),
            ),
        ),
        layout.Row(
            layout.Column(
                PluginContentPanel('full_width_page'),
            ),
        ),
    )


@register_model_view(DataFile, 'delete')
class DataFileDeleteView(generic.ObjectDeleteView):
    queryset = DataFile.objects.all()


@register_model_view(DataFile, 'bulk_delete', path='delete', detail=False)
class DataFileBulkDeleteView(generic.BulkDeleteView):
    queryset = DataFile.objects.defer('data')
    filterset = filtersets.DataFileFilterSet
    table = tables.DataFileTable


#
# Jobs
#

@register_model_view(Job, 'list', path='', detail=False)
class JobListView(generic.ObjectListView):
    queryset = Job.objects.defer('data')
    filterset = filtersets.JobFilterSet
    filterset_form = forms.JobFilterForm
    table = tables.JobTable
    actions = (BulkExport, BulkDelete)


@register_model_view(Job)
class JobView(generic.ObjectView):
    queryset = Job.objects.all()
    actions = (DeleteObject,)
    layout = layout.SimpleLayout(
        left_panels=[
            panels.JobPanel(),
        ],
        right_panels=[
            panels.JobSchedulingPanel(),
        ],
        bottom_panels=[
            JSONPanel('data', title=_('Data')),
        ],
    )


@register_model_view(Job, 'log')
class JobLogView(generic.ObjectView):
    queryset = Job.objects.all()
    actions = (DeleteObject,)
    template_name = 'core/job/log.html'
    tab = ViewTab(
        label=_('Log'),
        badge=lambda obj: len(obj.log_entries),
        weight=500,
    )
    layout = layout.Layout(
        layout.Row(
            layout.Column(
                ContextTablePanel('table', title=_('Log Entries')),
                PluginContentPanel('left_page'),
            ),
        ),
        layout.Row(
            layout.Column(
                PluginContentPanel('full_width_page'),
            ),
        ),
    )

    def get_extra_context(self, request, instance):
        table = JobLogEntryTable(instance.log_entries)
        table.configure(request)
        return {
            'table': table,
        }


@register_model_view(Job, 'delete')
class JobDeleteView(generic.ObjectDeleteView):
    queryset = Job.objects.defer('data')


@register_model_view(Job, 'bulk_delete', path='delete', detail=False)
class JobBulkDeleteView(generic.BulkDeleteView):
    queryset = Job.objects.defer('data')
    filterset = filtersets.JobFilterSet
    table = tables.JobTable


#
# Change logging
#

@register_model_view(ObjectChange, 'list', path='', detail=False)
class ObjectChangeListView(generic.ObjectListView):
    queryset = None
    filterset = filtersets.ObjectChangeFilterSet
    filterset_form = forms.ObjectChangeFilterForm
    table = tables.ObjectChangeTable
    template_name = 'core/objectchange_list.html'
    actions = (BulkExport,)

    def get_queryset(self, request):
        return ObjectChange.objects.valid_models()


@register_model_view(ObjectChange)
class ObjectChangeView(generic.ObjectView):
    queryset = None
    layout = layout.Layout(
        layout.Row(
            layout.Column(panels.ObjectChangePanel()),
            layout.Column(TemplatePanel('core/panels/objectchange_difference.html')),
        ),
        layout.Row(
            layout.Column(TemplatePanel('core/panels/objectchange_prechange.html')),
            layout.Column(TemplatePanel('core/panels/objectchange_postchange.html')),
        ),
        layout.Row(
            layout.Column(PluginContentPanel('left_page')),
            layout.Column(PluginContentPanel('right_page')),
        ),
        layout.Row(
            layout.Column(
                TemplatePanel('core/panels/objectchange_related.html'),
                PluginContentPanel('full_width_page'),
            ),
        ),
    )

    def get_queryset(self, request):
        return ObjectChange.objects.valid_models()

    def get_extra_context(self, request, instance):
        related_changes = ObjectChange.objects.valid_models().restrict(request.user, 'view').filter(
            request_id=instance.request_id
        ).exclude(
            pk=instance.pk
        )
        related_changes_table = tables.ObjectChangeTable(
            data=related_changes[:50],
            orderable=False
        )
        related_changes_table.configure(request)

        objectchanges = ObjectChange.objects.valid_models().restrict(request.user, 'view').filter(
            changed_object_type=instance.changed_object_type,
            changed_object_id=instance.changed_object_id,
        )

        next_change = objectchanges.filter(time__gt=instance.time).order_by('time').first()
        prev_change = objectchanges.filter(time__lt=instance.time).order_by('-time').first()

        if not instance.prechange_data and instance.action in ['update', 'delete'] and prev_change:
            non_atomic_change = True
            prechange_data = prev_change.postchange_data_clean
        else:
            non_atomic_change = False
            prechange_data = instance.prechange_data_clean

        if prechange_data and instance.postchange_data:
            diff_added, diff_removed = deep_compare_dict(
                prechange_data,
                instance.postchange_data_clean,
                exclude=['last_updated'],
            )
        else:
            diff_added = {}
            diff_removed = {}

        return {
            'diff_added': diff_added,
            'diff_removed': diff_removed,
            'next_change': next_change,
            'prev_change': prev_change,
            'related_changes_table': related_changes_table,
            'related_changes_count': related_changes.count(),
            'non_atomic_change': non_atomic_change
        }


#
# Config Revisions
#

@register_model_view(ConfigRevision, 'list', path='', detail=False)
class ConfigRevisionListView(generic.ObjectListView):
    queryset = ConfigRevision.objects.all()
    filterset = filtersets.ConfigRevisionFilterSet
    filterset_form = forms.ConfigRevisionFilterForm
    table = tables.ConfigRevisionTable
    actions = (AddObject, BulkExport)


@register_model_view(ConfigRevision)
class ConfigRevisionView(generic.ObjectView):
    queryset = ConfigRevision.objects.all()
    layout = layout.Layout(
        layout.Row(
            layout.Column(
                TemplatePanel('core/panels/configrevision_data.html'),
                TemplatePanel('core/panels/configrevision_comment.html'),
                PluginContentPanel('left_page'),
            ),
        ),
        layout.Row(
            layout.Column(
                PluginContentPanel('full_width_page'),
            ),
        ),
    )

    def get_extra_context(self, request, instance):
        """
        Retrieve additional context for a given request and instance.
        """
        # Copy the revision data to avoid modifying the original
        config = deepcopy(instance.data or {})

        # Serialize any JSON-based classes
        for attr in ['CUSTOM_VALIDATORS', 'DEFAULT_USER_PREFERENCES', 'PROTECTION_RULES']:
            if attr in config:
                config[attr] = json.dumps(config[attr], cls=ConfigJSONEncoder, indent=4)

        return {
            'config': config,
        }


@register_model_view(ConfigRevision, 'add', detail=False)
class ConfigRevisionEditView(generic.ObjectEditView):
    queryset = ConfigRevision.objects.all()
    form = forms.ConfigRevisionForm


@register_model_view(ConfigRevision, 'delete')
class ConfigRevisionDeleteView(generic.ObjectDeleteView):
    queryset = ConfigRevision.objects.all()


@register_model_view(ConfigRevision, 'bulk_delete', path='delete', detail=False)
class ConfigRevisionBulkDeleteView(generic.BulkDeleteView):
    queryset = ConfigRevision.objects.all()
    filterset = filtersets.ConfigRevisionFilterSet
    table = tables.ConfigRevisionTable


@register_model_view(ConfigRevision, 'restore')
class ConfigRevisionRestoreView(ContentTypePermissionRequiredMixin, View):

    def get_required_permission(self):
        return 'core.configrevision_edit'

    def get(self, request, pk):
        candidate_config = get_object_or_404(ConfigRevision, pk=pk)

        # Get the current ConfigRevision
        config_version = get_config().version
        current_config = ConfigRevision.objects.filter(pk=config_version).first()

        params = []
        for param in PARAMS:
            params.append((
                param.name,
                current_config.data.get(param.name, None) if current_config else None,
                candidate_config.data.get(param.name, None)
            ))

        return render(request, 'core/configrevision_restore.html', {
            'object': candidate_config,
            'params': params,
        })

    def post(self, request, pk):
        if not request.user.has_perm('core.configrevision_edit'):
            return HttpResponseForbidden()

        candidate_config = get_object_or_404(ConfigRevision, pk=pk)
        candidate_config.activate()
        messages.success(request, _("Restored configuration revision #{id}").format(id=pk))

        return redirect(candidate_config.get_absolute_url())


#
# Background Tasks (RQ)
#

class BaseRQView(UserPassesTestMixin, View):

    def test_func(self):
        return self.request.user.is_superuser


class BackgroundQueueListView(TableMixin, BaseRQView):
    table = tables.BackgroundQueueTable

    def get(self, request):
        data = get_statistics(run_maintenance_tasks=True)["queues"]
        table = self.get_table(data, request, bulk_actions=False)

        return render(request, 'core/rq_queue_list.html', {
            'table': table,
        })


class BackgroundTaskListView(TableMixin, BaseRQView):
    table = tables.BackgroundTaskTable

    def get_table_data(self, request, queue, status):

        # Call get_jobs() to returned queued tasks
        if status == RQJobStatus.QUEUED:
            return queue.get_jobs()

        return get_rq_jobs_from_status(queue, status)

    def get(self, request, queue_index, status):
        queue = get_queue_by_index(queue_index)
        data = self.get_table_data(request, queue, status)
        table = self.get_table(data, request, False)

        # If this is an HTMX request, return only the rendered table HTML
        if htmx_partial(request):
            return render(request, 'htmx/table.html', {
                'table': table,
            })

        return render(request, 'core/rq_task_list.html', {
            'table': table,
            'queue': queue,
            'status': status,
        })


class BackgroundTaskView(BaseRQView):

    def get(self, request, job_id):
        # all the RQ queues should use the same connection
        config = get_queues_list()[0]
        try:
            job = RQ_Job.fetch(job_id, connection=get_redis_connection(config['connection_config']),)
        except NoSuchJobError:
            raise Http404(_("Job {job_id} not found").format(job_id=job_id))

        queue_index = get_queues_map()[job.origin]
        queue = get_queue_by_index(queue_index)

        try:
            exc_info = job._exc_info
        except AttributeError:
            exc_info = None

        return render(request, 'core/rq_task.html', {
            'queue': queue,
            'job': job,
            'queue_index': queue_index,
            'dependency_id': job._dependency_id,
            'exc_info': exc_info,
        })


class BackgroundTaskDeleteView(BaseRQView):

    def get(self, request, job_id):
        if not request.htmx:
            return redirect(reverse('core:background_queue_list'))

        form = ConfirmationForm(initial=request.GET)

        return render(request, 'htmx/delete_form.html', {
            'object_type': 'background task',
            'object': job_id,
            'form': form,
            'form_url': reverse('core:background_task_delete', kwargs={'job_id': job_id})
        })

    def post(self, request, job_id):
        form = ConfirmationForm(request.POST)

        if form.is_valid():
            delete_rq_job(job_id)
            messages.success(request, _('Job {id} has been deleted.').format(id=job_id))
        else:
            messages.error(request, _('Error deleting job {id}: {error}').format(id=job_id, error=form.errors[0]))

        return redirect(reverse('core:background_queue_list'))


class BackgroundTaskRequeueView(BaseRQView):

    def get(self, request, job_id):
        requeue_rq_job(job_id)
        messages.success(request, _('Job {id} has been re-enqueued.').format(id=job_id))
        return redirect(reverse('core:background_task', args=[job_id]))


class BackgroundTaskEnqueueView(BaseRQView):

    def get(self, request, job_id):
        # all the RQ queues should use the same connection
        enqueue_rq_job(job_id)
        messages.success(request, _('Job {id} has been enqueued.').format(id=job_id))
        return redirect(reverse('core:background_task', args=[job_id]))


class BackgroundTaskStopView(BaseRQView):

    def get(self, request, job_id):
        stopped_jobs = stop_rq_job(job_id)
        if len(stopped_jobs) == 1:
            messages.success(request, _('Job {id} has been stopped.').format(id=job_id))
        else:
            messages.error(request, _('Failed to stop job {id}').format(id=job_id))

        return redirect(reverse('core:background_task', args=[job_id]))


class WorkerListView(TableMixin, BaseRQView):
    table = tables.WorkerTable

    def get_table_data(self, request, queue):
        clean_worker_registry(queue)
        all_workers = Worker.all(queue.connection)
        workers = [worker for worker in all_workers if queue.name in worker.queue_names()]
        return workers

    def get(self, request, queue_index):
        queue = get_queue_by_index(queue_index)
        data = self.get_table_data(request, queue)

        table = self.get_table(data, request, False)

        # If this is an HTMX request, return only the rendered table HTML
        if htmx_partial(request):
            if not request.htmx.target:
                table.embedded = True
                # Hide selection checkboxes
                if 'pk' in table.base_columns:
                    table.columns.hide('pk')
            return render(request, 'htmx/table.html', {
                'table': table,
                'queue': queue,
            })

        return render(request, 'core/rq_worker_list.html', {
            'table': table,
            'queue': queue,
        })


class WorkerView(BaseRQView):

    def get(self, request, key):
        # all the RQ queues should use the same connection
        config = get_queues_list()[0]
        worker = Worker.find_by_key('rq:worker:' + key, connection=get_redis_connection(config['connection_config']))
        # Convert microseconds to milliseconds
        worker.total_working_time = worker.total_working_time / 1000

        return render(request, 'core/rq_worker.html', {
            'worker': worker,
            'job': worker.get_current_job(),
            'total_working_time': worker.total_working_time * 1000,
        })


#
# System
#


class SystemView(UserPassesTestMixin, View):

    def test_func(self):
        return self.request.user.is_superuser

    def _get_stats(self):
        psql_version = db_name = db_size = None
        try:
            with connection.cursor() as cursor:
                cursor.execute("SELECT version()")
                psql_version = cursor.fetchone()[0]
                psql_version = psql_version.split('(')[0].strip()
                cursor.execute("SELECT current_database()")
                db_name = cursor.fetchone()[0]
                cursor.execute("SELECT pg_size_pretty(pg_database_size(current_database()))")
                db_size = cursor.fetchone()[0]
        except (DatabaseError, IndexError):
            pass
        return {
            'netbox_release': settings.RELEASE,
            'django_version': django_version,
            'python_version': platform.python_version(),
            'postgresql_version': psql_version,
            'database_name': db_name,
            'database_size': db_size,
            'rq_worker_count': Worker.count(get_connection('default')),
        }

    def _get_object_counts(self):
        objects = {}
        for ot in ObjectType.objects.public().order_by('app_label', 'model'):
            if model := ot.model_class():
                objects[ot] = model.objects.count()
        return objects

    def get(self, request):
        stats = self._get_stats()
        django_apps = get_installed_apps()
        config = get_config()
        plugins = get_installed_plugins()
        objects = self._get_object_counts()

        # Raw data export
        if 'export' in request.GET:
            db_schema = get_db_schema()
            stats['netbox_release'] = stats['netbox_release'].asdict()
            params = [param.name for param in PARAMS]
            data = {
                **stats,
                'django_apps': django_apps,
                'plugins': plugins,
                'config': {
                    k: getattr(config, k) for k in sorted(params)
                },
                'objects': {
                    f'{ot.app_label}.{ot.model}': count for ot, count in objects.items()
                },
                'db_schema': {
                    table['name']: {
                        'columns': table['columns'],
                        'indexes': table['indexes'],
                    } for table in db_schema
                },
            }
            response = HttpResponse(json.dumps(data, cls=ConfigJSONEncoder, indent=4), content_type='text/json')
            response['Content-Disposition'] = 'attachment; filename="netbox.json"'
            return response

        # Serialize any JSON-based classes
        for attr in ['CUSTOM_VALIDATORS', 'DEFAULT_USER_PREFERENCES', 'PROTECTION_RULES']:
            if hasattr(config, attr) and getattr(config, attr, None):
                setattr(config, attr, json.dumps(getattr(config, attr), cls=ConfigJSONEncoder, indent=4))

        return render(request, 'core/system.html', {
            'stats': stats,
            'django_apps': django_apps,
            'config': config,
            'plugins': plugins,
            'objects': objects,
        })


class SystemDBSchemaView(UserPassesTestMixin, View):

    def test_func(self):
        return self.request.user.is_superuser

    @staticmethod
    def _get_db_schema_groups(db_schema):
        plugin_app_labels = {
            app_config.label
            for app_config in django_apps_registry.get_app_configs()
            if isinstance(app_config, PluginConfig)
        }
        # Sort longest-first so "netbox_branching" matches before "netbox"
        sorted_plugin_labels = sorted(plugin_app_labels, key=len, reverse=True)
        groups = {}
        for table in db_schema:
            matched_plugin = next(
                (label for label in sorted_plugin_labels if table['name'].startswith(label + '_')),
                None,
            )
            if matched_plugin:
                prefix = matched_plugin
            elif '_' in table['name']:
                prefix = table['name'].split('_')[0]
            else:
                prefix = 'other'
            groups.setdefault(prefix, []).append(table)
        return sorted(
            [
                {
                    'name': name,
                    'tables': tables,
                    'index_count': sum(len(t['indexes']) for t in tables),
                    'is_plugin': name in plugin_app_labels,
                }
                for name, tables in groups.items()
            ],
            key=lambda g: (g['is_plugin'], g['name']),
        )

    def get(self, request):
        db_schema = get_db_schema()
        db_schema_groups = self._get_db_schema_groups(db_schema)
        db_schema_stats = {
            'total_tables': len(db_schema),
            'total_columns': sum(len(t['columns']) for t in db_schema),
            'total_indexes': sum(len(t['indexes']) for t in db_schema),
        }
        return render(request, 'core/htmx/system_db_schema.html', {
            'db_schema': db_schema,
            'db_schema_groups': db_schema_groups,
            'db_schema_stats': db_schema_stats,
        })


#
# Plugins
#

class BasePluginView(UserPassesTestMixin, View):
    CACHE_KEY_CATALOG_ERROR = 'plugins-catalog-error'

    def test_func(self):
        return self.request.user.is_superuser

    def get_cached_plugins(self, request):
        catalog_plugins = {}
        catalog_plugins_error = cache.get(self.CACHE_KEY_CATALOG_ERROR, default=False)
        if not catalog_plugins_error:
            catalog_plugins = get_catalog_plugins()
            if not catalog_plugins and not settings.ISOLATED_DEPLOYMENT:
                # Cache for 5 minutes to avoid spamming connection
                cache.set(self.CACHE_KEY_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):
        q = request.GET.get('q', None)

        plugins = self.get_cached_plugins(request).values()
        if q:
            plugins = [obj for obj in plugins if q.casefold() in obj.title_short.casefold()]

        plugins = [plugin for plugin in plugins if not plugin.hidden]

        table = CatalogPluginTable(plugins)
        table.configure(request)

        # If this is an HTMX request, return only the rendered table HTML
        if htmx_partial(request):
            return render(request, 'htmx/table.html', {
                'table': table,
            })

        return render(request, 'core/plugin_list.html', {
            'table': table,
        })


class PluginView(BasePluginView):

    def get(self, request, name):

        plugins = self.get_cached_plugins(request)
        if name not in plugins:
            raise Http404(_("Plugin {name} not found").format(name=name))
        plugin = plugins[name]

        table = PluginVersionTable(plugin.release_recent_history)
        table.configure(request)

        return render(request, 'core/plugin.html', {
            'plugin': plugin,
            'table': table,
        })