Jan Dittberner
c9a9fa11b2
This commit isolates the celery task invocations of the userdbs app into signal handlers. All celery interaction is now asynchronously handled in userdbs.signals.
278 lines
8 KiB
Python
278 lines
8 KiB
Python
"""
|
|
Admin functionality for the :py:mod:`userdbs.models` models.
|
|
|
|
"""
|
|
from __future__ import absolute_import
|
|
|
|
from django import forms
|
|
from django.contrib import admin
|
|
from django.utils.translation import ugettext_lazy as _
|
|
|
|
from .models import (
|
|
DatabaseUser,
|
|
UserDatabase,
|
|
)
|
|
|
|
|
|
class DatabaseUserCreationForm(forms.ModelForm):
|
|
"""
|
|
A form for creating :py:class:`database users
|
|
<userdbs.models.DatabaseUser>`
|
|
|
|
"""
|
|
|
|
class Meta:
|
|
model = DatabaseUser
|
|
fields = ['osuser', 'db_type']
|
|
|
|
def save(self, commit=True):
|
|
"""
|
|
Save the database user.
|
|
|
|
:param boolean commit: whether to save the created database user
|
|
:return: database user instance
|
|
:rtype: :py:class:`userdbs.models.DatabaseUser`
|
|
|
|
"""
|
|
dbuser = DatabaseUser.objects.create_database_user(
|
|
osuser=self.cleaned_data['osuser'],
|
|
db_type=self.cleaned_data['db_type'], commit=commit)
|
|
return dbuser
|
|
|
|
def save_m2m(self):
|
|
"""
|
|
Noop.
|
|
|
|
"""
|
|
|
|
|
|
class UserDatabaseCreationForm(forms.ModelForm):
|
|
"""
|
|
A form for creating :py:class:`user databases
|
|
<userdbs.models.UserDatabase>`
|
|
|
|
"""
|
|
|
|
class Meta:
|
|
model = UserDatabase
|
|
fields = ['db_user']
|
|
|
|
def save(self, commit=True):
|
|
"""
|
|
Save the user database.
|
|
|
|
:param boolean commit: whether to save the created user database
|
|
:return: user database instance
|
|
:rtype: :py:class:`userdbs.models.UserDatabase`
|
|
|
|
"""
|
|
database = UserDatabase.objects.create_userdatabase(
|
|
db_user=self.cleaned_data['db_user'], commit=commit)
|
|
return database
|
|
|
|
def save_m2m(self):
|
|
"""
|
|
Noop.
|
|
|
|
"""
|
|
|
|
|
|
class DatabaseUserAdmin(admin.ModelAdmin):
|
|
"""
|
|
Admin class for working with :py:class:`database users
|
|
<userdbs.models.DatabaseUser>`
|
|
|
|
"""
|
|
actions = ['perform_delete_selected']
|
|
add_form = DatabaseUserCreationForm
|
|
|
|
def get_form(self, request, obj=None, **kwargs):
|
|
"""
|
|
Use special form for database user creation.
|
|
|
|
:param request: the current HTTP request
|
|
:param obj: either a :py:class:`Database user
|
|
<userdbs.models.DatabaseUser>` instance or None for a new database
|
|
user
|
|
:param kwargs: keyword arguments to be passed to
|
|
:py:meth:`django.contrib.admin.ModelAdmin.get_form`
|
|
:return: form instance
|
|
|
|
"""
|
|
defaults = {}
|
|
if obj is None:
|
|
defaults.update({
|
|
'form': self.add_form,
|
|
})
|
|
defaults.update(kwargs)
|
|
return super(DatabaseUserAdmin, self).get_form(
|
|
request, obj, **defaults)
|
|
|
|
def get_readonly_fields(self, request, obj=None):
|
|
"""
|
|
Make sure that osuser, name and db_type are not editable for existing
|
|
database users.
|
|
|
|
:param request: the current HTTP request
|
|
:param obj: either a :py:class:`Database user
|
|
<userdbs.models.DatabaseUser>` instance or None for a new database
|
|
user
|
|
:return: a list of fields
|
|
:rtype: list
|
|
|
|
"""
|
|
if obj:
|
|
return ['osuser', 'name', 'db_type']
|
|
return []
|
|
|
|
def save_model(self, request, obj, form, change):
|
|
"""
|
|
Make sure that the user is created in the target database.
|
|
|
|
:param request: the current HTTP request
|
|
:param obj: a :py:class:`Database user <userdbs.models.DatabaseUser>`
|
|
instance
|
|
:param form: the form instance
|
|
:param boolean change: whether this is a change operation or not
|
|
|
|
"""
|
|
if not change:
|
|
obj.create_in_database()
|
|
super(DatabaseUserAdmin, self).save_model(request, obj, form, change)
|
|
|
|
def perform_delete_selected(self, request, queryset):
|
|
"""
|
|
Action to delete a list of selected database users.
|
|
|
|
This action calls the delete method of each selected database user in
|
|
contrast to the default `delete_selected`
|
|
|
|
:param request: the current HTTP request
|
|
:param queryset: Django ORM queryset representing the selected database
|
|
users
|
|
|
|
"""
|
|
for dbuser in queryset.all():
|
|
dbuser.delete()
|
|
perform_delete_selected.short_description = _(
|
|
'Delete selected database users')
|
|
|
|
def get_actions(self, request):
|
|
"""
|
|
Get the available actions for database users.
|
|
|
|
This overrides the default behavior to remove the default
|
|
`delete_selected` action.
|
|
|
|
:param request: the current HTTP request
|
|
:return: list of actions
|
|
:rtype: list
|
|
|
|
"""
|
|
actions = super(DatabaseUserAdmin, self).get_actions(request)
|
|
if 'delete_selected' in actions: # pragma: no cover
|
|
del actions['delete_selected']
|
|
return actions
|
|
|
|
|
|
class UserDatabaseAdmin(admin.ModelAdmin):
|
|
"""
|
|
Admin class for working with :py:class:`user databases
|
|
<userdbs.models.UserDatabase>`
|
|
|
|
"""
|
|
actions = ['perform_delete_selected']
|
|
add_form = UserDatabaseCreationForm
|
|
|
|
def get_form(self, request, obj=None, **kwargs):
|
|
"""
|
|
Use special form for user database creation.
|
|
|
|
:param request: the current HTTP request
|
|
:param obj: either a :py:class:`User database
|
|
<userdbs.models.UserDatabase>` instance or None for a new user
|
|
database
|
|
:param kwargs: keyword arguments to be passed to
|
|
:py:meth:`django.contrib.admin.ModelAdmin.get_form`
|
|
:return: form instance
|
|
|
|
"""
|
|
defaults = {}
|
|
if obj is None:
|
|
defaults.update({
|
|
'form': self.add_form,
|
|
})
|
|
defaults.update(kwargs)
|
|
return super(UserDatabaseAdmin, self).get_form(
|
|
request, obj, **defaults)
|
|
|
|
def get_readonly_fields(self, request, obj=None):
|
|
"""
|
|
Make sure that db_name and db_user are not editable for existing user
|
|
databases.
|
|
|
|
:param request: the current HTTP request
|
|
:param obj: either a :py:class:`User database
|
|
<userdbs.models.UserDatabase>` instance or None for a new user
|
|
database
|
|
:return: a list of fields
|
|
:rtype: list
|
|
|
|
"""
|
|
if obj:
|
|
return ['db_name', 'db_user']
|
|
return []
|
|
|
|
def save_model(self, request, obj, form, change):
|
|
"""
|
|
Make sure that the database is created in the target database server.
|
|
|
|
:param request: the current HTTP request
|
|
:param obj: a :py:class:`Database user <userdbs.models.DatabaseUser>`
|
|
instance
|
|
:param form: the form instance
|
|
:param boolean change: whether this is a change operation or not
|
|
|
|
"""
|
|
if not change:
|
|
obj.create_in_database()
|
|
super(UserDatabaseAdmin, self).save_model(request, obj, form, change)
|
|
|
|
def perform_delete_selected(self, request, queryset):
|
|
"""
|
|
Action to delete a list of selected user databases.
|
|
|
|
This action calls the delete method of each selected user database in
|
|
contrast to the default `delete_selected`
|
|
|
|
:param request: the current HTTP request
|
|
:param queryset: Django ORM queryset representing the selected user
|
|
databases
|
|
|
|
"""
|
|
for database in queryset.all():
|
|
database.delete()
|
|
perform_delete_selected.short_description = _(
|
|
'Delete selected user databases')
|
|
|
|
def get_actions(self, request):
|
|
"""
|
|
Get the available actions for user databases.
|
|
|
|
This overrides the default behavior to remove the default
|
|
`delete_selected` action.
|
|
|
|
:param request: the current HTTP request
|
|
:return: list of actions
|
|
:rtype: list
|
|
|
|
"""
|
|
actions = super(UserDatabaseAdmin, self).get_actions(request)
|
|
if 'delete_selected' in actions: # pragma: no cover
|
|
del actions['delete_selected']
|
|
return actions
|
|
|
|
|
|
admin.site.register(DatabaseUser, DatabaseUserAdmin)
|
|
admin.site.register(UserDatabase, UserDatabaseAdmin)
|