"""Admin configuration for user app."""
from contextlib import suppress
from functools import wraps
from apps.core.exceptions import BaseAPIException
from apps.core.models import BackgroundJob
from apps.user.management import lock_managed_users, validate_access_change
from apps.user.models import Profile, User
from apps.user.services import UserService
from django.contrib import admin, messages
from django.contrib.admin.sites import NotRegistered
from django.contrib.admin.utils import unquote
from django.contrib.auth.admin import UserAdmin as BaseUserAdmin
from django.contrib.auth.models import Group
from django.db import transaction
from django.http import HttpResponseRedirect
from django.urls import reverse
from django.utils.html import format_html
from django.utils.translation import gettext_lazy as _
try:
from django_celery_beat.models import (
ClockedSchedule,
CrontabSchedule,
IntervalSchedule,
PeriodicTask,
SolarSchedule,
)
except ImportError: # pragma: no cover - optional dependency import guard
ClockedSchedule = None
CrontabSchedule = None
IntervalSchedule = None
PeriodicTask = None
SolarSchedule = None
try:
from rest_framework_simplejwt.token_blacklist.models import (
BlacklistedToken,
OutstandingToken,
)
except ImportError: # pragma: no cover - optional dependency import guard
BlacklistedToken = None
OutstandingToken = None
def _unregister(model) -> None:
if model is None:
return
with suppress(NotRegistered):
admin.site.unregister(model)
_unregister(User)
def _management_errors(view):
"""Present guarded account-change failures through Django admin messages."""
@wraps(view)
def wrapped(self, request, *args, **kwargs):
try:
return view(self, request, *args, **kwargs)
except BaseAPIException as exc:
self.message_user(request, exc.message, level=messages.ERROR)
return HttpResponseRedirect(reverse("admin:user_user_changelist"))
return wrapped
class ProfileInline(admin.StackedInline):
"""Inline для профиля пользователя."""
model = Profile
extra = 0
max_num = 1
can_delete = False
verbose_name_plural = "Профиль"
fk_name = "user"
fields = ["first_name", "mid_name", "last_name", "bio", "avatar", "date_of_birth"]
@admin.register(User)
class UserAdmin(BaseUserAdmin):
"""Admin для пользователей без групп."""
inlines = [ProfileInline]
list_display = [
"username",
"email",
"phone",
"is_verified_badge",
"is_active_badge",
"is_staff",
"created_at",
]
list_filter = ["is_staff", "is_superuser", "is_active", "is_verified", "created_at"]
search_fields = ["username", "email", "phone"]
ordering = ["-created_at"]
list_per_page = 50
date_hierarchy = "created_at"
fieldsets = (
(None, {"fields": ("username", "password")}),
(_("Personal info"), {"fields": ("email", "phone")}),
(
_("Permissions"),
{
"fields": (
"is_active",
"is_staff",
"is_superuser",
"is_verified",
"user_permissions",
),
"classes": ("collapse",),
},
),
(
_("Important dates"),
{"fields": ("last_login", "date_joined", "created_at", "updated_at")},
),
)
add_fieldsets = (
(
None,
{
"classes": ("wide",),
"fields": (
"username",
"email",
"password1",
"password2",
"is_staff",
"is_active",
),
},
),
)
readonly_fields = ["created_at", "updated_at", "last_login", "date_joined"]
actions = ["verify_users", "unverify_users", "activate_users", "deactivate_users"]
@_management_errors
def changeform_view(self, request, object_id=None, form_url="", extra_context=None):
return super().changeform_view(request, object_id, form_url, extra_context)
@_management_errors
def delete_view(self, request, object_id, extra_context=None):
return super().delete_view(request, object_id, extra_context)
@_management_errors
def user_change_password(self, request, id, form_url=""):
user = self.get_object(request, unquote(id))
if request.method == "POST" and user is not None:
with lock_managed_users(actor_id=request.user.pk, user_ids=[user.pk]):
return super().user_change_password(request, id, form_url)
return super().user_change_password(request, id, form_url)
@_management_errors
def changelist_view(self, request, extra_context=None):
# delete_selected writes LogEntry rows before delete_queryset. Roll those
# back too if the complete selection fails an account-management guard.
with transaction.atomic():
return super().changelist_view(request, extra_context)
def save_model(self, request, obj, form, change):
with lock_managed_users(
actor_id=request.user.pk, user_ids=[obj.pk] if change else []
) as users:
if change:
validate_access_change(
users[obj.pk],
actor_id=request.user.pk,
is_active=obj.is_active,
is_staff=obj.is_staff,
)
super().save_model(request, obj, form, change)
def delete_model(self, request, obj):
UserService.delete_user(obj.pk, actor_id=request.user.pk)
def delete_queryset(self, request, queryset):
UserService.delete_users(
queryset.values_list("pk", flat=True), actor_id=request.user.pk
)
def is_verified_badge(self, obj):
if obj.is_verified:
return format_html(
'✓'
)
return format_html(
'✗'
)
is_verified_badge.short_description = "Верифицирован"
is_verified_badge.admin_order_field = "is_verified"
def is_active_badge(self, obj):
if obj.is_active:
return format_html(
'Активен'
)
return format_html(
'Неактивен'
)
is_active_badge.short_description = "Статус"
is_active_badge.admin_order_field = "is_active"
@admin.action(description="Верифицировать выбранных пользователей")
def verify_users(self, request, queryset):
updated = queryset.update(is_verified=True)
self.message_user(request, f"Верифицировано {updated} пользователей")
@admin.action(description="Снять верификацию")
def unverify_users(self, request, queryset):
updated = queryset.update(is_verified=False)
self.message_user(request, f"Снята верификация у {updated} пользователей")
@admin.action(description="Активировать пользователей")
def activate_users(self, request, queryset):
updated = UserService.set_users_active(
queryset.values_list("pk", flat=True),
actor_id=request.user.pk,
is_active=True,
)
self.message_user(request, f"Активировано {updated} пользователей")
@admin.action(description="Деактивировать пользователей")
def deactivate_users(self, request, queryset):
updated = UserService.set_users_active(
queryset.values_list("pk", flat=True),
actor_id=request.user.pk,
is_active=False,
)
self.message_user(request, f"Деактивировано {updated} пользователей")
for model in (
Group,
BackgroundJob,
OutstandingToken,
BlacklistedToken,
PeriodicTask,
CrontabSchedule,
IntervalSchedule,
SolarSchedule,
ClockedSchedule,
):
_unregister(model)