123 lines
4.4 KiB
Python
123 lines
4.4 KiB
Python
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,Ingress_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 button
|
|
from core.admin import UniversalModelAdmin
|
|
|
|
|
|
|
|
@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
|
|
|
|
|
|
@admin.register(Deployment_Data)
|
|
class DeploymentDataAdmin(UniversalModelAdmin):
|
|
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]
|
|
ordering = ('name',)
|
|
@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(UniversalModelAdmin):
|
|
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(UniversalModelAdmin):
|
|
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(UniversalModelAdmin):
|
|
list_display = ('namespace',)
|
|
|
|
|
|
@button()
|
|
def import_namespace(self, request):
|
|
# your business logic here
|
|
import_namespace()
|
|
self.message_user(request, 'Namespaces imported')
|
|
|
|
@admin.register(Ingress_Data)
|
|
class IngressAdmin(UniversalModelAdmin):
|
|
list_display = ('url','namespace') |