from django.contrib import admin,messages from django.utils.translation import ngettext from .models import Deployment_Data,Schedule_Day_Data,Schedule_Week_Data,Namespace_Data from .utils import import_namespace,import_deployment,generate_default_day_schedule,generate_default_week_schedule,Check_server_supposed_XXX,retrieve_and_apply from apps.kubernetes_connection.utils import scale_up,scale_down from admin_extra_buttons.api import ExtraButtonsMixin, button from import_export.admin import ImportExportModelAdmin DB = 'cloudmanager_db' @admin.action(description="scale up") def scale_up_deploy(modeladmin, request, queryset): for query in queryset.values(): scale_up(namespace_name=query['namespace'],deploy_name=query['name'],replicas=query['replicas']) modeladmin.message_user( request, ngettext( "%d server was successfully scaled up.", "%d servers were successfully scaled up.", queryset.values().count(), ) % queryset.values().count(), messages.SUCCESS, ) @admin.action(description="scale down") def scale_down_deploy(modeladmin, request, queryset): for query in queryset.values(): scale_down(namespace_name=query['namespace'],deploy_name=query['name']) modeladmin.message_user( request, ngettext( "%d server was successfully scaled down.", "%d servers were successfully scaled down.", queryset.values().count(), ) % queryset.values().count(), messages.SUCCESS, ) @admin.action(description="Expected False") def Check_server_supposed_False(modeladmin, request, queryset): servers_supposed_false = Check_server_supposed_XXX(False) for server in servers_supposed_false: messages.add_message(request, messages.INFO, server) @admin.action(description="Expected True") def Check_server_supposed_true(modeladmin, request, queryset): servers_supposed_false = Check_server_supposed_XXX(True) for server in servers_supposed_false: messages.add_message(request, messages.INFO, server) class ActiveAtHourFilter(admin.SimpleListFilter): title = 'Active at hour' parameter_name = 'active_at_hour' def lookups(self, request, model_admin): return [(str(h), f'{h:02d}:00') for h in range(24)] def queryset(self, request, queryset): if self.value() is not None: hour = self.value() return queryset.filter(**{f'schedule_day__state_at_{hour}': True}) return queryset class MultiDBModelAdmin(admin.ModelAdmin): """Base class that pins all admin operations to DB.""" db = DB def get_queryset(self, request): return super().get_queryset(request).using(self.db) def save_model(self, request, obj, form, change): obj.save(using=self.db) def delete_model(self, request, obj): obj.delete(using=self.db) def delete_queryset(self, request, queryset): queryset.using(self.db).delete() @admin.register(Deployment_Data) class DeploymentDataAdmin(ImportExportModelAdmin,ExtraButtonsMixin, MultiDBModelAdmin): list_display = ('name', 'namespace', 'schedule_week', 'schedule_day', 'replicas') list_filter = (ActiveAtHourFilter,) actions = [Check_server_supposed_False,Check_server_supposed_true,scale_up_deploy,scale_down_deploy] @button() def retrieve_and_apply(self, request): # your business logic here retrieve_and_apply() self.message_user(request, 'Operation started') @button() def import_deployment(self, request): # your business logic here import_deployment() self.message_user(request, 'Deployments imported') @admin.register(Schedule_Day_Data) class ScheduleDayAdmin(ImportExportModelAdmin,ExtraButtonsMixin,MultiDBModelAdmin): list_display = ['name_schedule'] + [f'state_at_{h}' for h in range(24)] @button() def generate_default_day_schedule_admin(self, request): generate_default_day_schedule() generate_default_week_schedule() self.message_user(request, 'Created default schedules') @admin.register(Schedule_Week_Data) class ScheduleWeekAdmin(ImportExportModelAdmin,ExtraButtonsMixin,MultiDBModelAdmin): list_display = ( 'name_schedule', 'state_Monday', 'state_Tuesday', 'state_Wednesday', 'state_Thursday', 'state_Friday', 'state_Saturday', 'state_Sunday', ) @button() def generate_default_day_schedule_admin(self, request): generate_default_day_schedule() generate_default_week_schedule() self.message_user(request, 'Created default schedules') @admin.register(Namespace_Data) class NamespaceAdmin(ImportExportModelAdmin,ExtraButtonsMixin, MultiDBModelAdmin): list_display = ('namespace',) @button() def import_namespace(self, request): # your business logic here import_namespace() self.message_user(request, 'Namespaces imported')